From 943f30f19b136a62dedcb9c33fc52b6d6a500443 Mon Sep 17 00:00:00 2001 From: Fury03 Date: Mon, 27 Jul 2026 16:17:41 +0100 Subject: [PATCH 1/2] feat: answer feedback capture and a quality-improvement loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Maintainers had no signal about which answers were failing or why. This closes that loop end-to-end. - Every chat answer now carries a stable message_id so the frontend can address one turn, tracked in structures parallel to active_chats rather than restructuring it (the streaming and memory paths depend on its current shape). - POST /feedback attaches an up/down rating and validated failure categories to a specific answer. The prompt and the *displayed* answer (after safety/hadith/confidence shaping) are snapshotted server-side, so the stored record is what the user actually saw; on a free-tier restart the snapshot is gone and the client supplies prompt/answer, else 422. Idempotent per (chat_id, message_id); rate-limited per IP. - feedback.py stores records in SQLite locally or Redis when REDIS_URL is set β€” the same store direction the session and scholar-review work use. - Two admin endpoints (GET /feedback/stats, /feedback/records) expose the aggregate signal, gated on X-Admin-Token with constant-time comparison and disabled until ADMIN_TOKEN is set. - scripts/export_eval_candidates.py turns down-rated records into evaluation-dataset candidates (#16 format), deduplicated, always needs_review:true, never fabricating an expected answer for religious content. It reads whichever store the service is configured to use. This is a rebuild of #45 onto current dev (which had moved far past that branch) with the review feedback applied: blocking SQLite/Redis I/O in the async endpoints is offloaded via run_in_threadpool; the export honours the configured backend instead of hardcoding SQLite; the shared rate limiter has a reset() and tests clear it so limiter state never leaks between them; and no virtualenv is committed. Tests (tests/test_feedback.py, 38 offline): store upsert/idempotency/ filtering/stats, rate-limiter allow/block/reset/expiry, backend selection, export mapping/dedup/backend, and the endpoints β€” message_id on responses and history, snapshot resolution and client fallback, validation, admin auth (401/403/503), and rate limiting. No live model or Redis calls. Closes #43 --- .env.example | 6 + .github/workflows/ci.yml | 7 +- .gitignore | 7 + README.md | 61 ++++ feedback.py | 506 ++++++++++++++++++++++++++++++ main.py | 248 ++++++++++++++- scripts/export_eval_candidates.py | 188 +++++++++++ tests/test_feedback.py | 448 ++++++++++++++++++++++++++ 8 files changed, 1465 insertions(+), 6 deletions(-) create mode 100644 feedback.py create mode 100644 scripts/export_eval_candidates.py create mode 100644 tests/test_feedback.py diff --git a/.env.example b/.env.example index ec70d97..8f62bfb 100644 --- a/.env.example +++ b/.env.example @@ -33,3 +33,9 @@ GEMINI_API_KEY=your_api_key_here # ZAKAT_NISAB_USD=6000 # fallback when no gold price is reachable # NISAB_CACHE_TTL_SECONDS=21600 # how long a fetched gold price is reused # GOLD_PRICE_TIMEOUT=8 + +# Answer feedback loop (optional) +# ADMIN_TOKEN=generate_a_long_random_value # enables /feedback/stats and /feedback/records +# FEEDBACK_DB_PATH=feedback.db # SQLite path when REDIS_URL is unset +# FEEDBACK_RATE_LIMIT_MAX=20 +# FEEDBACK_RATE_LIMIT_WINDOW=60 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9da3257..ec62779 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,10 +28,10 @@ jobs: pip install pytest flake8 pytest-asyncio - name: Run linting - run: flake8 main.py memory stellar.py nisab.py safety telemetry.py tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py tests/test_confidence.py tests/test_review_queue.py tests/test_tafsir.py tests/test_zakat.py tests/test_telemetry.py tests/test_multilingual.py tests/test_memory_profile.py tests/test_memory_extraction.py tests/test_memory_integration.py scripts/build_hadith_data.py scripts/build_surah_index.py --max-line-length=120 --ignore=E501,W503 + run: flake8 main.py memory stellar.py nisab.py safety telemetry.py tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py feedback.py scripts/export_eval_candidates.py tests/test_feedback.py tests/test_confidence.py tests/test_review_queue.py tests/test_tafsir.py tests/test_zakat.py tests/test_telemetry.py tests/test_multilingual.py tests/test_memory_profile.py tests/test_memory_extraction.py tests/test_memory_integration.py scripts/build_hadith_data.py scripts/build_surah_index.py --max-line-length=120 --ignore=E501,W503 - name: Check syntax - run: python -m compileall -q main.py memory stellar.py nisab.py safety telemetry.py tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py scripts/build_hadith_data.py scripts/build_surah_index.py + run: python -m compileall -q main.py memory stellar.py nisab.py safety telemetry.py tests/redteam study.py fiqh.py hadith.py confidence.py review.py review_store.py tafsir.py semantic_cache.py feedback.py scripts/export_eval_candidates.py scripts/build_hadith_data.py scripts/build_surah_index.py - name: Run offline safety and red-team tests run: pytest -q tests/redteam @@ -60,6 +60,9 @@ jobs: - name: Run zakat and nisab tests run: pytest -q tests/test_zakat.py + - name: Run feedback loop tests + run: pytest -q tests/test_feedback.py + - name: Run LLM telemetry tests run: pytest -q tests/test_telemetry.py diff --git a/.gitignore b/.gitignore index e80b856..1615903 100644 --- a/.gitignore +++ b/.gitignore @@ -38,3 +38,10 @@ ENV/ *.log # Scholar-review export (contains user questions) data/review/ + +# Local feedback database (SQLite fallback store) +feedback.db +feedback.db-* + +# Committed virtual environments are machine-specific β€” never commit them +venv_linux/ diff --git a/README.md b/README.md index f037960..77a5412 100644 --- a/README.md +++ b/README.md @@ -58,6 +58,9 @@ The platform is composed of three services: | `GET` | `/review/reviewed` | Answers that already carry a verdict (reviewer token) | | `GET` | `/review/{id}` | A single review item (reviewer token) | | `POST` | `/review/{id}/verdict` | Record approve / correct / reject (reviewer token) | +| `POST` | `/feedback` | Rate a specific answer and flag failure categories | +| `GET` | `/feedback/stats` | Aggregate answer-quality metrics (admin token) | +| `GET` | `/feedback/records` | Browse flagged records, filterable (admin token) | ## πŸš€ Getting Started @@ -146,6 +149,10 @@ services: | `ZAKAT_NISAB_USD` | Fallback nisab when no gold price can be fetched | `6000` | | `NISAB_CACHE_TTL_SECONDS` | How long a fetched gold price is reused | `21600` (6h) | | `GOLD_PRICE_TIMEOUT` | Gold price request timeout in seconds | `8` | +| `ADMIN_TOKEN` | Enables the feedback admin endpoints; required as `X-Admin-Token` | β€” (endpoints disabled) | +| `FEEDBACK_DB_PATH` | SQLite path for the feedback store (when Redis is not used) | `feedback.db` | +| `FEEDBACK_RATE_LIMIT_MAX` | Max feedback submissions per IP per window | `20` | +| `FEEDBACK_RATE_LIMIT_WINDOW` | Feedback rate-limit window in seconds | `60` | | `QURAN_API_BASE` | Base URL for tafsir/ayah retrieval | `https://api.quran.com/api/v4` | | `QURAN_API_TIMEOUT` | Tafsir request timeout in seconds | `15` | | `TAFSIR_MAX_AYAT` | Maximum ayat per `/tafsir` request | `10` | @@ -379,6 +386,60 @@ validation like any other malformed input, so one can never reach Horizon. If a message looks like it contains a secret key, the assistant refuses to use it and warns the user to treat it as compromised β€” without repeating it back. +### Answer feedback & the quality loop + +Every chat answer carries a stable `message_id`, so the frontend can rate a +specific turn. `POST /feedback` attaches an up/down rating and optional failure +categories to that answer: + +```bash +curl -sX POST http://localhost:8000/feedback \ + -H 'Content-Type: application/json' \ + -d '{"chat_id": "…", "message_id": "…", "rating": "down", + "categories": ["wrong_or_missing_citation"], "comment": "Ayah number is off"}' +``` + +- **Snapshot resolution.** The prompt and the *displayed* answer (after any + safety, hadith, or confidence shaping) are resolved server-side from what the + user actually saw β€” the client never has to be trusted for them. On a + free-tier restart the snapshot is gone; the client then supplies `prompt` and + `answer`, and a request missing both is a `422`. +- **Validation.** `rating` must be `up`/`down`, categories are checked against a + fixed taxonomy, and `comment` is length-capped β€” bad input is a `422`. +- **Idempotent.** Resubmitting for the same `(chat_id, message_id)` overwrites, + so a user changing their mind never double-counts. +- **Rate-limited** per IP (in-process sliding window), and durably stored in + SQLite locally or Redis when `REDIS_URL` is set β€” the same store direction the + session and scholar-review work use, not a parallel one. + +Maintainers read the aggregate signal through two **admin** endpoints, gated on +`X-Admin-Token` and disabled entirely until `ADMIN_TOKEN` is set: + +```bash +curl -s http://localhost:8000/feedback/stats -H "X-Admin-Token: $ADMIN_TOKEN" +curl -s "http://localhost:8000/feedback/records?rating=down&category=too_vague" \ + -H "X-Admin-Token: $ADMIN_TOKEN" +``` + +#### Eval-candidate export + +Down-rated answers become candidates for the evaluation dataset (issue #16 +format). Each carries `needs_review: true` and an `answer_draft` for the +reviewer to judge β€” the script **never** fabricates an expected answer for +religious content: + +```bash +# Reads whichever store the service is configured to use (Redis or SQLite): +python scripts/export_eval_candidates.py --output candidates.jsonl +REDIS_URL=redis://localhost:6379 python scripts/export_eval_candidates.py --output candidates.jsonl + +# …or force a specific SQLite file: +python scripts/export_eval_candidates.py --db feedback.db --output candidates.jsonl +``` + +Near-duplicate prompts are deduplicated; approved candidates feed the harness +and, via #56, the semantic cache. + ### Content-safety testing The versioned policy lives in [`safety/policy.yaml`](safety/policy.yaml), with diff --git a/feedback.py b/feedback.py new file mode 100644 index 0000000..22a237a --- /dev/null +++ b/feedback.py @@ -0,0 +1,506 @@ +"""Answer-feedback capture for the Deen Bridge AI service. + +Stores per-message ratings and failure categories so the team can measure +answer quality and grow the evaluation dataset from real user pain rather than +guesses. This is the capture-and-storage half of issue #43; the scholar-review +queue (#56) owns human vetting of low-confidence answers, and they share +storage direction (Redis when configured) rather than inventing parallel ones. + +Storage backends (selected at import time): + - Redis β€” when REDIS_URL is set (aligns with the session/queue store direction) + - SQLite β€” fallback for local dev and free-tier Render + +Abuse resistance: + - One record per (chat_id, message_id): resubmission overwrites (idempotent) + - comment capped at COMMENT_MAX_CHARS characters (validated server-side) + - categories validated against FEEDBACK_TAXONOMY + - per-IP rate limiting via an in-process sliding-window counter + (stopgap until real auth/rate-limiting infrastructure lands) + - SQLite bounded by SQLITE_MAX_RECORDS; Redis keys carry a TTL + +Admin endpoints are protected by ADMIN_TOKEN (stopgap). +""" + +from __future__ import annotations + +import json +import logging +import os +import sqlite3 +import threading +import time +from collections import defaultdict, deque +from dataclasses import dataclass, field +from typing import Any, Deque, Dict, List, Optional + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Constants +# --------------------------------------------------------------------------- + +FEEDBACK_TAXONOMY = { + "incorrect_information", + "wrong_or_missing_citation", + "one_sided_fiqh_answer", + "too_vague", + "too_long", + "wrong_language", + "poor_adab", + "refused_unnecessarily", + "other", +} + +COMMENT_MAX_CHARS = 1000 + +# Redis TTL for feedback records (30 days). +REDIS_TTL_SECONDS = 60 * 60 * 24 * 30 + +# SQLite cap β€” oldest records are pruned when this is exceeded. +SQLITE_MAX_RECORDS = 50_000 + +# Rate limiting: max submissions per IP per window. +RATE_LIMIT_MAX = int(os.getenv("FEEDBACK_RATE_LIMIT_MAX", "20")) +RATE_LIMIT_WINDOW_SECONDS = int(os.getenv("FEEDBACK_RATE_LIMIT_WINDOW", "60")) + + +# --------------------------------------------------------------------------- +# Rate limiter (in-process sliding window β€” stopgap) +# --------------------------------------------------------------------------- + + +class RateLimiter: + """Per-IP sliding-window rate limiter (in-process, non-persistent).""" + + def __init__( + self, + max_calls: int = RATE_LIMIT_MAX, + window_seconds: float = RATE_LIMIT_WINDOW_SECONDS, + ) -> None: + self._max = max_calls + self._window = window_seconds + self._buckets: Dict[str, Deque[float]] = defaultdict(deque) + self._lock = threading.Lock() + + def is_allowed(self, ip: str) -> bool: + now = time.monotonic() + cutoff = now - self._window + with self._lock: + bucket = self._buckets[ip] + while bucket and bucket[0] < cutoff: + bucket.popleft() + if len(bucket) >= self._max: + return False + bucket.append(now) + return True + + def reset(self) -> None: + """Clear all buckets. Used by tests so limiter state never leaks between them.""" + with self._lock: + self._buckets.clear() + + +rate_limiter = RateLimiter() + + +# --------------------------------------------------------------------------- +# Feedback record +# --------------------------------------------------------------------------- + + +@dataclass +class FeedbackRecord: + feedback_id: str + chat_id: str + message_id: str + rating: str # "up" | "down" + categories: List[str] = field(default_factory=list) + comment: Optional[str] = None + prompt: Optional[str] = None + answer: Optional[str] = None + model_name: Optional[str] = None + generation_config: Optional[Dict[str, Any]] = None + created_at: str = "" # ISO-8601 UTC + + def to_dict(self) -> Dict[str, Any]: + return { + "feedback_id": self.feedback_id, + "chat_id": self.chat_id, + "message_id": self.message_id, + "rating": self.rating, + "categories": self.categories, + "comment": self.comment, + "prompt": self.prompt, + "answer": self.answer, + "model_name": self.model_name, + "generation_config": self.generation_config, + "created_at": self.created_at, + } + + @staticmethod + def from_dict(d: Dict[str, Any]) -> "FeedbackRecord": + gen_cfg = d.get("generation_config") + if isinstance(gen_cfg, str): + try: + gen_cfg = json.loads(gen_cfg) if gen_cfg else None + except (json.JSONDecodeError, TypeError): + gen_cfg = None + cats = d.get("categories", []) + if isinstance(cats, str): + try: + cats = json.loads(cats) if cats else [] + except (json.JSONDecodeError, TypeError): + cats = [] + return FeedbackRecord( + feedback_id=d["feedback_id"], + chat_id=d["chat_id"], + message_id=d["message_id"], + rating=d["rating"], + categories=cats, + comment=d.get("comment") or None, + prompt=d.get("prompt") or None, + answer=d.get("answer") or None, + model_name=d.get("model_name") or None, + generation_config=gen_cfg, + created_at=d.get("created_at", ""), + ) + + +# --------------------------------------------------------------------------- +# Storage back-ends +# --------------------------------------------------------------------------- + + +class FeedbackStore: + """Abstract interface β€” concrete implementations below.""" + + def upsert(self, record: FeedbackRecord) -> None: + raise NotImplementedError + + def get(self, chat_id: str, message_id: str) -> Optional[FeedbackRecord]: + raise NotImplementedError + + def list_records( + self, + rating: Optional[str] = None, + category: Optional[str] = None, + limit: int = 100, + ) -> List[FeedbackRecord]: + raise NotImplementedError + + def stats(self) -> Dict[str, Any]: + raise NotImplementedError + + +# -- SQLite store ----------------------------------------------------------- + +_SQLITE_PATH = os.getenv("FEEDBACK_DB_PATH", "feedback.db") + +_CREATE_TABLE = """ +CREATE TABLE IF NOT EXISTS feedback ( + feedback_id TEXT NOT NULL, + chat_id TEXT NOT NULL, + message_id TEXT NOT NULL, + rating TEXT NOT NULL, + categories TEXT NOT NULL DEFAULT '[]', + comment TEXT, + prompt TEXT, + answer TEXT, + model_name TEXT, + generation_config TEXT, + created_at TEXT NOT NULL, + PRIMARY KEY (chat_id, message_id) +); +CREATE INDEX IF NOT EXISTS idx_feedback_rating ON feedback(rating); +CREATE INDEX IF NOT EXISTS idx_feedback_created ON feedback(created_at); +CREATE INDEX IF NOT EXISTS idx_feedback_model ON feedback(model_name); +""" + + +class SQLiteFeedbackStore(FeedbackStore): + """Thread-safe SQLite store; prunes oldest rows past SQLITE_MAX_RECORDS.""" + + def __init__(self, db_path: str = _SQLITE_PATH) -> None: + self._db_path = db_path + self._local = threading.local() + self._init_db() + + def _conn(self) -> sqlite3.Connection: + if getattr(self._local, "conn", None) is None: + conn = sqlite3.connect(self._db_path, check_same_thread=False) + conn.row_factory = sqlite3.Row + self._local.conn = conn + return self._local.conn + + def _init_db(self) -> None: + conn = sqlite3.connect(self._db_path, check_same_thread=False) + conn.row_factory = sqlite3.Row + conn.executescript(_CREATE_TABLE) + conn.commit() + conn.close() + + def _prune(self, conn: sqlite3.Connection) -> None: + count = conn.execute("SELECT COUNT(*) FROM feedback").fetchone()[0] + if count > SQLITE_MAX_RECORDS: + excess = count - SQLITE_MAX_RECORDS + conn.execute( + "DELETE FROM feedback WHERE rowid IN " + "(SELECT rowid FROM feedback ORDER BY created_at ASC LIMIT ?)", + (excess,), + ) + + def upsert(self, record: FeedbackRecord) -> None: + conn = self._conn() + conn.execute( + """ + INSERT INTO feedback + (feedback_id, chat_id, message_id, rating, categories, + comment, prompt, answer, model_name, generation_config, created_at) + VALUES (?,?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(chat_id, message_id) DO UPDATE SET + feedback_id = excluded.feedback_id, + rating = excluded.rating, + categories = excluded.categories, + comment = excluded.comment, + prompt = excluded.prompt, + answer = excluded.answer, + model_name = excluded.model_name, + generation_config = excluded.generation_config, + created_at = excluded.created_at + """, + ( + record.feedback_id, + record.chat_id, + record.message_id, + record.rating, + json.dumps(record.categories), + record.comment, + record.prompt, + record.answer, + record.model_name, + json.dumps(record.generation_config) if record.generation_config else None, + record.created_at, + ), + ) + self._prune(conn) + conn.commit() + + def get(self, chat_id: str, message_id: str) -> Optional[FeedbackRecord]: + conn = self._conn() + row = conn.execute( + "SELECT * FROM feedback WHERE chat_id=? AND message_id=?", + (chat_id, message_id), + ).fetchone() + return FeedbackRecord.from_dict(dict(row)) if row else None + + def list_records( + self, + rating: Optional[str] = None, + category: Optional[str] = None, + limit: int = 100, + ) -> List[FeedbackRecord]: + conn = self._conn() + sql = "SELECT * FROM feedback WHERE 1=1" + params: list = [] + if rating: + sql += " AND rating=?" + params.append(rating) + if category: + # categories stored as a JSON array string β€” LIKE on the quoted token + sql += " AND categories LIKE ?" + params.append(f'%"{category}"%') + sql += " ORDER BY created_at DESC LIMIT ?" + params.append(limit) + rows = conn.execute(sql, params).fetchall() + return [FeedbackRecord.from_dict(dict(r)) for r in rows] + + def stats(self) -> Dict[str, Any]: + conn = self._conn() + + total = conn.execute("SELECT COUNT(*) FROM feedback").fetchone()[0] + up = conn.execute("SELECT COUNT(*) FROM feedback WHERE rating='up'").fetchone()[0] + down = conn.execute("SELECT COUNT(*) FROM feedback WHERE rating='down'").fetchone()[0] + + cat_counts: Dict[str, Dict[str, int]] = {} + for row in conn.execute("SELECT categories, rating FROM feedback").fetchall(): + try: + cats = json.loads(row["categories"]) if row["categories"] else [] + except (json.JSONDecodeError, TypeError): + cats = [] + for cat in cats: + bucket = cat_counts.setdefault(cat, {"up": 0, "down": 0}) + bucket[row["rating"]] = bucket.get(row["rating"], 0) + 1 + + model_rows = conn.execute( + "SELECT model_name, rating, COUNT(*) as cnt " + "FROM feedback GROUP BY model_name, rating" + ).fetchall() + model_counts: Dict[str, Dict[str, int]] = {} + for r in model_rows: + name = r["model_name"] or "unknown" + bucket = model_counts.setdefault(name, {"up": 0, "down": 0}) + bucket[r["rating"]] = r["cnt"] + + day_rows = conn.execute( + "SELECT substr(created_at,1,10) as day, rating, COUNT(*) as cnt " + "FROM feedback GROUP BY day, rating ORDER BY day DESC LIMIT 14" + ).fetchall() + by_day: Dict[str, Dict[str, int]] = {} + for r in day_rows: + bucket = by_day.setdefault(r["day"], {"up": 0, "down": 0}) + bucket[r["rating"]] = r["cnt"] + + return { + "total": total, + "up": up, + "down": down, + "up_ratio": round(up / total, 4) if total else None, + "by_category": cat_counts, + "by_model": model_counts, + "by_day": by_day, + } + + +# -- Redis store ------------------------------------------------------------ + + +class RedisFeedbackStore(FeedbackStore): + """Redis-backed store. + + Key layout: + feedback:: -> JSON hash (TTL REDIS_TTL_SECONDS) + feedback:index:rating: -> sorted set, score = unix timestamp + feedback:index:cat: -> sorted set, score = unix timestamp + feedback:index:model: -> sorted set, score = unix timestamp + """ + + _PREFIX = "feedback" + + def __init__(self, client: Any) -> None: + self._r = client + + def _record_key(self, chat_id: str, message_id: str) -> str: + return f"{self._PREFIX}:{chat_id}:{message_id}" + + def upsert(self, record: FeedbackRecord) -> None: + key = self._record_key(record.chat_id, record.message_id) + ts = time.time() + data = record.to_dict() + data["categories"] = json.dumps(data["categories"]) + data["generation_config"] = ( + json.dumps(data["generation_config"]) if data["generation_config"] else "" + ) + pipe = self._r.pipeline() + pipe.hset(key, mapping={k: (v if v is not None else "") for k, v in data.items()}) + pipe.expire(key, REDIS_TTL_SECONDS) + pipe.zadd(f"{self._PREFIX}:index:rating:{record.rating}", {key: ts}) + for cat in record.categories: + pipe.zadd(f"{self._PREFIX}:index:cat:{cat}", {key: ts}) + pipe.zadd(f"{self._PREFIX}:index:model:{record.model_name or 'unknown'}", {key: ts}) + pipe.execute() + + def get(self, chat_id: str, message_id: str) -> Optional[FeedbackRecord]: + data = self._r.hgetall(self._record_key(chat_id, message_id)) + return FeedbackRecord.from_dict(data) if data else None + + def _fetch_keys(self, index_key: str, limit: int) -> List[str]: + return self._r.zrevrange(index_key, 0, limit - 1) + + def _fetch_records(self, keys: List[str]) -> List[FeedbackRecord]: + if not keys: + return [] + pipe = self._r.pipeline() + for k in keys: + pipe.hgetall(k) + records = [] + for data in pipe.execute(): + if data: + try: + records.append(FeedbackRecord.from_dict(data)) + except (KeyError, TypeError): + continue + return records + + def list_records( + self, + rating: Optional[str] = None, + category: Optional[str] = None, + limit: int = 100, + ) -> List[FeedbackRecord]: + if rating: + keys = self._fetch_keys(f"{self._PREFIX}:index:rating:{rating}", limit) + elif category: + keys = self._fetch_keys(f"{self._PREFIX}:index:cat:{category}", limit) + else: + up_keys = self._fetch_keys(f"{self._PREFIX}:index:rating:up", limit) + down_keys = self._fetch_keys(f"{self._PREFIX}:index:rating:down", limit) + seen: set = set() + keys = [] + for k in up_keys + down_keys: + if k not in seen: + seen.add(k) + keys.append(k) + keys = keys[:limit] + if category and rating: + cat_keys = set(self._fetch_keys(f"{self._PREFIX}:index:cat:{category}", limit * 2)) + keys = [k for k in keys if k in cat_keys][:limit] + return self._fetch_records(keys) + + def stats(self) -> Dict[str, Any]: + up = self._r.zcard(f"{self._PREFIX}:index:rating:up") + down = self._r.zcard(f"{self._PREFIX}:index:rating:down") + total = up + down + + cat_counts: Dict[str, Dict[str, int]] = {} + for cat in FEEDBACK_TAXONOMY: + n = self._r.zcard(f"{self._PREFIX}:index:cat:{cat}") + if n: + cat_counts[cat] = {"total": n} + + return { + "total": total, + "up": up, + "down": down, + "up_ratio": round(up / total, 4) if total else None, + "by_category": cat_counts, + "by_model": {}, # full per-model aggregation omitted for Redis brevity + "by_day": {}, + } + + +# --------------------------------------------------------------------------- +# Backend selection +# --------------------------------------------------------------------------- + + +def _build_redis_store() -> Optional[RedisFeedbackStore]: + redis_url = os.getenv("REDIS_URL") + if not redis_url: + return None + try: + import redis as _redis # type: ignore + + client = _redis.from_url(redis_url, decode_responses=True) + client.ping() + logger.info("Feedback store: Redis (%s)", redis_url.split("@")[-1]) + return RedisFeedbackStore(client) + except Exception as exc: # noqa: BLE001 - any Redis failure degrades to SQLite + logger.warning("Redis unavailable (%s); falling back to SQLite.", exc) + return None + + +def build_store() -> FeedbackStore: + """Return the configured feedback store: Redis when reachable, else SQLite. + + The single place backend selection happens, so the service, the export + script, and tests all agree on which store is live rather than each + hardcoding SQLite. + """ + redis_store = _build_redis_store() + if redis_store is not None: + return redis_store + logger.info("Feedback store: SQLite (%s)", _SQLITE_PATH) + return SQLiteFeedbackStore() + + +store: FeedbackStore = build_store() diff --git a/main.py b/main.py index 7f3487a..5337f03 100644 --- a/main.py +++ b/main.py @@ -2,13 +2,18 @@ import json import logging import os +import secrets from typing import Any, Dict, List, Optional import uuid +from datetime import datetime, timezone +from collections import OrderedDict from dotenv import load_dotenv -from fastapi import FastAPI, HTTPException, Request, Response +from fastapi import Depends, FastAPI, HTTPException, Request, Response +from fastapi.concurrency import run_in_threadpool from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse -from pydantic import BaseModel, Field +from fastapi.security import APIKeyHeader +from pydantic import BaseModel, Field, field_validator import google.generativeai as genai import time @@ -61,6 +66,13 @@ ) from review import enqueue_for_review, router as review_router from review_store import get_review_store +from feedback import ( + COMMENT_MAX_CHARS, + FEEDBACK_TAXONOMY, + FeedbackRecord, + rate_limiter, + store as feedback_store, +) from memory import ChatSummary, UserProfile, create_memory_store, render_user_context from memory.extraction import ( @@ -131,6 +143,7 @@ class ChatRequest(BaseModel): class Message(BaseModel): role: str content: str + message_id: Optional[str] = None # present on model turns, for feedback class Moderation(BaseModel): @@ -142,6 +155,7 @@ class ChatResponse(BaseModel): response: Optional[str] = None text: Optional[str] = None chat_id: str + message_id: Optional[str] = None # stable id of the answer just returned history: List[Message] = [] moderation: Optional[Moderation] = None fiqh: Optional[FiqhInfo] = None @@ -152,6 +166,46 @@ class ChatResponse(BaseModel): language: Optional[str] = None +class FeedbackRequest(BaseModel): + chat_id: str + message_id: str + rating: str = Field(..., description="'up' or 'down'") + categories: Optional[List[str]] = None + comment: Optional[str] = None + # Supplied by the client when the session is no longer in memory (restart). + prompt: Optional[str] = None + answer: Optional[str] = None + + @field_validator("rating") + @classmethod + def rating_must_be_valid(cls, v: str) -> str: + if v not in ("up", "down"): + raise ValueError("rating must be 'up' or 'down'") + return v + + @field_validator("categories") + @classmethod + def categories_must_be_valid(cls, v: Optional[List[str]]) -> Optional[List[str]]: + if v is None: + return v + invalid = set(v) - FEEDBACK_TAXONOMY + if invalid: + raise ValueError( + f"Unknown categories: {sorted(invalid)}. " + f"Valid choices: {sorted(FEEDBACK_TAXONOMY)}" + ) + return v + + @field_validator("comment") + @classmethod + def comment_length(cls, v: Optional[str]) -> Optional[str]: + if v and len(v) > COMMENT_MAX_CHARS: + raise ValueError( + f"comment must not exceed {COMMENT_MAX_CHARS} characters (got {len(v)})" + ) + return v + + def classify_for_safety(prompt: str, candidate_ids: List[str]): """Gemini classifier seam; offline tests replace this with a fixture.""" classifier_instruction = ( @@ -247,6 +301,66 @@ def get_safety_settings(): sessions: Dict[str, Any] = {} active_chats: Dict[str, Any] = {} +# --- Feedback support ------------------------------------------------------ +# The generation config captured into a feedback record so a flagged answer is +# reproducible evidence, kept beside the model name telemetry already tracks. +GENERATION_CONFIG: Dict[str, Any] = { + "temperature": 0.7, + "top_p": 0.8, + "top_k": 40, + "max_output_tokens": 2048, +} + +# Stable ids for model answers, so the frontend can address one turn and the +# feedback endpoint can locate what was said. Kept parallel to active_chats +# rather than restructuring it: chat_id -> [message_id per model turn, in order]. +chat_message_ids: Dict[str, List[str]] = {} + +# Faithful snapshot of what the user was actually shown for each answered turn, +# keyed by (chat_id, message_id). Feedback reads from here so the stored answer +# is the displayed text (post safety/hadith/abstention shaping), not the raw +# model output. Bounded LRU β€” on a free-tier restart it is empty, which is why +# the feedback endpoint also accepts a client-supplied prompt/answer. +FEEDBACK_SNAPSHOT_MAX = int(os.getenv("FEEDBACK_SNAPSHOT_MAX", "5000")) +answer_snapshots: "OrderedDict[tuple, Dict[str, str]]" = OrderedDict() + + +def _record_answer(chat_id: str, prompt: str, answer: str) -> str: + """Assign a message id to a fresh answer, snapshot it, and return the id.""" + message_id = str(uuid.uuid4()) + chat_message_ids.setdefault(chat_id, []).append(message_id) + answer_snapshots[(chat_id, message_id)] = {"prompt": prompt, "answer": answer} + while len(answer_snapshots) > FEEDBACK_SNAPSHOT_MAX: + answer_snapshots.popitem(last=False) + return message_id + + +def _tag_history_with_message_ids(chat_id: str, history: List["Message"]) -> None: + """Attach each model turn's stable id to the history returned to the client.""" + ids = chat_message_ids.get(chat_id, []) + model_turn = 0 + for message in history: + if message.role == "model": + if model_turn < len(ids): + message.message_id = ids[model_turn] + model_turn += 1 + + +# --- Admin auth (stopgap until real auth/rate-limiting infrastructure) ------ +ADMIN_TOKEN = os.getenv("ADMIN_TOKEN", "") +_admin_header = APIKeyHeader(name="X-Admin-Token", auto_error=False) + + +async def require_admin(token: Optional[str] = Depends(_admin_header)) -> None: + """Gate admin endpoints on ADMIN_TOKEN; closed by default when unset.""" + if not ADMIN_TOKEN: + raise HTTPException( + status_code=503, + detail="ADMIN_TOKEN is not configured on this server.", + ) + if not token or not secrets.compare_digest(token, ADMIN_TOKEN): + raise HTTPException(status_code=403, detail="Invalid or missing admin token.") + ISLAMIC_CONTEXT = ( "You are an AI assistant for Deen Bridge, a platform for authentic Islamic education. " "Provide respectful, accurate, and context-aware responses grounded in authentic Islamic knowledge.\n\n" @@ -521,11 +635,13 @@ def _finalize() -> None: ]) active_chats[chat_id] = chat_session logger.info("Semantic cache HIT for prompt: %s", prompt[:80]) + cached_message_id = _record_answer(chat_id, prompt, cached.response) _finalize() _succeeded = True return ChatResponse( response=cached.response, chat_id=chat_id, + message_id=cached_message_id, history=cached.history, fiqh=fiqh_info, hadith_references=annotate_hadith(cached.response), @@ -687,6 +803,12 @@ async def generate(safety_prompt: str) -> str: fastapi_response.headers["X-Semantic-Cache"] = "bypass" if is_bypass else "miss" + # Assign this answer a stable id and snapshot the displayed text, so a + # later /feedback call can reference exactly this turn and store what + # the user actually saw (post safety/hadith/abstention shaping). + message_id = _record_answer(chat_id, prompt, response_text) + _tag_history_with_message_ids(chat_id, history) + trace.add_span("post_processing", (time.perf_counter() - _pp_start) * 1000.0) logger.info("Chat response generated successfully") # Build the response before finalizing, so a construction/validation @@ -695,6 +817,7 @@ async def generate(safety_prompt: str) -> str: response_obj = ChatResponse( response=response_text, chat_id=chat_id, + message_id=message_id, history=history, moderation=Moderation( category_id=safety_result.category_id, @@ -949,8 +1072,13 @@ async def event_generator(): @app.delete("/chat/{chat_id}") async def delete_chat(chat_id: str): try: - if chat_id in active_chats: - del active_chats[chat_id] + existed = chat_id in active_chats + active_chats.pop(chat_id, None) + # Drop this session's feedback bookkeeping too, so the message-id list + # and answer snapshots do not outlive the conversation they describe. + for message_id in chat_message_ids.pop(chat_id, []): + answer_snapshots.pop((chat_id, message_id), None) + if existed: logger.info(f"Deleted chat session: {chat_id}") return {"message": "Chat session deleted successfully"} return {"message": "Chat session not found"} @@ -959,6 +1087,118 @@ async def delete_chat(chat_id: str): raise HTTPException(status_code=500, detail="Internal server error") from e +# --------------------------------------------------------------------------- +# Feedback: capture and admin views +# --------------------------------------------------------------------------- + +def _client_ip(request: Request) -> str: + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + return forwarded.split(",")[0].strip() + return request.client.host if request.client else "unknown" + + +@app.post("/feedback", status_code=200) +async def submit_feedback(request: Request, body: FeedbackRequest): + """Attach a rating and optional failure categories to one model answer. + + The prompt/answer snapshot is resolved server-side from what the user was + shown for that (chat_id, message_id). If the snapshot is gone β€” a free-tier + restart evicts it β€” the client MUST supply prompt and answer, or the + request is rejected with 422. + + Rate-limited per IP (in-process sliding window). Idempotent: resubmitting + for the same (chat_id, message_id) overwrites the earlier record. + """ + if not rate_limiter.is_allowed(_client_ip(request)): + raise HTTPException( + status_code=429, + detail="Too many feedback submissions. Please wait before trying again.", + ) + + snapshot = answer_snapshots.get((body.chat_id, body.message_id)) + prompt_text = snapshot["prompt"] if snapshot else body.prompt + answer_text = snapshot["answer"] if snapshot else body.answer + + if snapshot is None and (not prompt_text or not answer_text): + raise HTTPException( + status_code=422, + detail=( + "This answer is no longer in memory. Please supply 'prompt' and " + "'answer' in the request body so the feedback has context." + ), + ) + + record = FeedbackRecord( + feedback_id=str(uuid.uuid4()), + chat_id=body.chat_id, + message_id=body.message_id, + rating=body.rating, + categories=body.categories or [], + comment=body.comment, + prompt=prompt_text, + answer=answer_text, + model_name=telemetry.GEMINI_MODEL, + generation_config=GENERATION_CONFIG, + created_at=datetime.now(timezone.utc).isoformat(), + ) + + try: + # SQLite/Redis I/O is synchronous; keep it off the event loop. + await run_in_threadpool(feedback_store.upsert, record) + except Exception as exc: + logger.error("Failed to store feedback: %s", exc) + raise HTTPException(status_code=500, detail="Failed to store feedback.") + + logger.info( + "Feedback stored: chat_id=%s message_id=%s rating=%s", + body.chat_id, body.message_id, body.rating, + ) + return {"status": "ok", "feedback_id": record.feedback_id} + + +@app.get("/feedback/stats", dependencies=[Depends(require_admin)]) +async def feedback_stats(): + """Aggregate quality metrics: rating ratios, per-category, per-model, by day. + + Requires the X-Admin-Token header. + """ + try: + return await run_in_threadpool(feedback_store.stats) + except Exception as exc: + logger.error("Failed to fetch feedback stats: %s", exc) + raise HTTPException(status_code=500, detail="Failed to fetch stats.") + + +@app.get("/feedback/records", dependencies=[Depends(require_admin)]) +async def feedback_records( + rating: Optional[str] = None, + category: Optional[str] = None, + limit: int = 50, +): + """Recent flagged records, filterable by rating and category. + + Requires the X-Admin-Token header. + """ + if rating and rating not in ("up", "down"): + raise HTTPException(status_code=422, detail="rating must be 'up' or 'down'") + if category and category not in FEEDBACK_TAXONOMY: + raise HTTPException( + status_code=422, + detail=f"Unknown category. Valid: {sorted(FEEDBACK_TAXONOMY)}", + ) + if not (1 <= limit <= 500): + raise HTTPException(status_code=422, detail="limit must be between 1 and 500") + try: + records = await run_in_threadpool( + feedback_store.list_records, rating, category, limit + ) + return {"records": [r.to_dict() for r in records]} + except Exception as exc: + logger.error("Failed to fetch feedback records: %s", exc) + raise HTTPException(status_code=500, detail="Failed to fetch records.") + + @app.get("/ping") async def ping(): """Lightweight liveness probe for container healthchecks and keep-alive pings.""" diff --git a/scripts/export_eval_candidates.py b/scripts/export_eval_candidates.py new file mode 100644 index 0000000..0a55b73 --- /dev/null +++ b/scripts/export_eval_candidates.py @@ -0,0 +1,188 @@ +#!/usr/bin/env python3 +"""Export down-rated feedback records as evaluation-dataset candidates. + +Targets the evaluation-harness dataset format (issue #16). Each emitted entry +carries ``needs_review: true`` β€” a human curator MUST supply an +``expected_answer`` before any record enters the golden set. + +This script intentionally NEVER generates expected answers for religious +content. That decision belongs to qualified scholars and the maintainers of +the evaluation harness. ``answer_draft`` is included only so a reviewer can +assess the failure; it is never treated as ground truth. + +Backend: the export reads from whichever feedback store the service is +configured to use β€” Redis when ``REDIS_URL`` is set, SQLite otherwise β€” so a +Redis-backed deployment exports from the live store, not a stale local +``feedback.db``. Pass ``--db`` to force a specific SQLite file. + +Usage +----- + python scripts/export_eval_candidates.py [options] + + --output PATH Write JSONL here (default: stdout) + --db PATH Force this SQLite DB, ignoring REDIS_URL + --min-categories N Only include records with at least N categories tagged + --limit N Max records to read (default: 2000) + +Output (one JSON object per line): + {"question", "category", "categories", "needs_review", "source", + "feedback_id", "model_name", "answer_draft", "comment"} + +Near-duplicate prompts (same first 120 chars after normalization) are +deduplicated: the first occurrence wins. +""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import sys +from typing import Any, Dict, List, Optional + +# Allow running from the repo root without installing the package. +sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) + +from feedback import ( # noqa: E402 + FeedbackRecord, + FeedbackStore, + SQLiteFeedbackStore, + build_store, +) + +_WS = re.compile(r"\s+") + +_TAXONOMY_TO_HARNESS_CATEGORY: Dict[str, str] = { + "incorrect_information": "factual_accuracy", + "wrong_or_missing_citation": "citation_quality", + "one_sided_fiqh_answer": "fiqh_balance", + "too_vague": "answer_completeness", + "too_long": "answer_conciseness", + "wrong_language": "language", + "poor_adab": "adab", + "refused_unnecessarily": "refusal", + "other": "other", +} + + +def _normalise(text: str) -> str: + """Lower-case and collapse whitespace for near-duplicate detection.""" + return _WS.sub(" ", text.lower().strip()) + + +def _primary_category(categories: List[str]) -> str: + """First recognized category, mapped to the harness label, else 'other'.""" + for cat in categories: + if cat in _TAXONOMY_TO_HARNESS_CATEGORY: + return _TAXONOMY_TO_HARNESS_CATEGORY[cat] + return "other" + + +def to_candidate(record: FeedbackRecord) -> Optional[Dict[str, Any]]: + """Convert a FeedbackRecord to an eval-harness candidate, or None. + + Returns None when there is no prompt snapshot β€” without the question there + is no useful candidate to review. + """ + if not record.prompt: + return None + return { + "question": record.prompt, + "category": _primary_category(record.categories), + "categories": record.categories, + "needs_review": True, + "source": "user_feedback", + "feedback_id": record.feedback_id, + "model_name": record.model_name or "unknown", + "answer_draft": record.answer or "", + "comment": record.comment or "", + } + + +def build_candidates( + records: List[FeedbackRecord], min_categories: int = 0 +) -> List[Dict[str, Any]]: + """Deduplicated candidates from *records*, first occurrence winning.""" + seen_prompts: set = set() + candidates: List[Dict[str, Any]] = [] + for record in records: + if min_categories and len(record.categories) < min_categories: + continue + candidate = to_candidate(record) + if candidate is None: + continue + norm = _normalise(record.prompt or "")[:120] + if norm in seen_prompts: + continue + seen_prompts.add(norm) + candidates.append(candidate) + return candidates + + +def _select_store(db_path: Optional[str]) -> FeedbackStore: + """The SQLite file when --db is given, otherwise the configured backend.""" + if db_path: + return SQLiteFeedbackStore(db_path=db_path) + return build_store() + + +def export( + output_path: Optional[str], + min_categories: int = 0, + limit: int = 2000, + db_path: Optional[str] = None, +) -> int: + """Run the export; return the number of candidates written.""" + store = _select_store(db_path) + records = store.list_records(rating="down", limit=limit) + candidates = build_candidates(records, min_categories=min_categories) + + out = open(output_path, "w", encoding="utf-8") if output_path else sys.stdout + try: + for candidate in candidates: + out.write(json.dumps(candidate, ensure_ascii=False) + "\n") + finally: + if output_path: + out.close() + return len(candidates) + + +def main() -> None: + parser = argparse.ArgumentParser( + description="Export down-rated feedback as evaluation-dataset candidates." + ) + parser.add_argument("--output", metavar="PATH", help="Output JSONL file (default: stdout)") + parser.add_argument( + "--db", + metavar="PATH", + default=None, + help="Force this SQLite DB path, ignoring REDIS_URL (default: configured backend)", + ) + parser.add_argument( + "--min-categories", + metavar="N", + type=int, + default=0, + help="Only include records with at least N failure categories (default: 0)", + ) + parser.add_argument( + "--limit", + metavar="N", + type=int, + default=2000, + help="Max feedback records to read (default: 2000)", + ) + args = parser.parse_args() + + count = export( + output_path=args.output, + min_categories=args.min_categories, + limit=args.limit, + db_path=args.db, + ) + print(f"Exported {count} candidates.", file=sys.stderr) + + +if __name__ == "__main__": + main() diff --git a/tests/test_feedback.py b/tests/test_feedback.py new file mode 100644 index 0000000..1a83fd3 --- /dev/null +++ b/tests/test_feedback.py @@ -0,0 +1,448 @@ +"""Tests for the answer-feedback loop: store, rate limiter, endpoints, and export. + +Everything runs offline against a temporary SQLite store β€” no Redis, no live +model calls. The FastAPI app is exercised through httpx's ASGI transport with +Gemini stubbed, matching the existing endpoint tests. +""" + +import asyncio +import json +import uuid + +import pytest +from httpx import ASGITransport, AsyncClient + +import feedback +from feedback import ( + FeedbackRecord, + RateLimiter, + SQLiteFeedbackStore, + build_store, +) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def make_record(**overrides) -> FeedbackRecord: + fields = dict( + feedback_id=str(uuid.uuid4()), + chat_id="chat-1", + message_id="msg-1", + rating="down", + categories=["incorrect_information"], + comment="wrong ayah number", + prompt="What does Surah al-Asr say?", + answer="An incorrect paraphrase.", + model_name="gemini-test", + generation_config={"temperature": 0.7}, + created_at="2026-07-25T10:00:00+00:00", + ) + fields.update(overrides) + return FeedbackRecord(**fields) + + +@pytest.fixture() +def store(tmp_path): + return SQLiteFeedbackStore(db_path=str(tmp_path / "feedback.db")) + + +# --------------------------------------------------------------------------- +# SQLite store +# --------------------------------------------------------------------------- + + +class TestSQLiteStore: + def test_upsert_and_get_roundtrip(self, store): + record = make_record() + store.upsert(record) + loaded = store.get("chat-1", "msg-1") + assert loaded is not None + assert loaded.rating == "down" + assert loaded.categories == ["incorrect_information"] + assert loaded.generation_config == {"temperature": 0.7} + + def test_get_missing_returns_none(self, store): + assert store.get("nope", "nope") is None + + def test_upsert_is_idempotent_per_message(self, store): + """Resubmitting for the same (chat_id, message_id) overwrites, not appends.""" + store.upsert(make_record(rating="down")) + store.upsert(make_record(rating="up", comment="changed my mind")) + records = store.list_records() + assert len(records) == 1 + assert records[0].rating == "up" + assert records[0].comment == "changed my mind" + + def test_list_filters_by_rating(self, store): + store.upsert(make_record(message_id="a", rating="down")) + store.upsert(make_record(message_id="b", rating="up")) + down = store.list_records(rating="down") + assert [r.message_id for r in down] == ["a"] + + def test_list_filters_by_category(self, store): + store.upsert(make_record(message_id="a", categories=["too_long"])) + store.upsert(make_record(message_id="b", categories=["poor_adab"])) + assert [r.message_id for r in store.list_records(category="poor_adab")] == ["b"] + + def test_category_filter_is_exact_not_substring(self, store): + """A LIKE on the quoted token must not match a different category.""" + store.upsert(make_record(message_id="a", categories=["wrong_language"])) + assert store.list_records(category="language") == [] + + def test_list_orders_newest_first(self, store): + store.upsert(make_record(message_id="old", created_at="2026-07-01T00:00:00+00:00")) + store.upsert(make_record(message_id="new", created_at="2026-07-25T00:00:00+00:00")) + assert [r.message_id for r in store.list_records()] == ["new", "old"] + + def test_stats_counts_and_ratio(self, store): + for i in range(3): + store.upsert(make_record(chat_id=f"c{i}", rating="down", categories=["too_vague"])) + store.upsert(make_record(chat_id="up1", rating="up", categories=[])) + stats = store.stats() + assert stats["total"] == 4 + assert stats["down"] == 3 + assert stats["up"] == 1 + assert stats["up_ratio"] == 0.25 + assert stats["by_category"]["too_vague"]["down"] == 3 + + def test_stats_empty_store(self, store): + stats = store.stats() + assert stats["total"] == 0 + assert stats["up_ratio"] is None + + +# --------------------------------------------------------------------------- +# Rate limiter +# --------------------------------------------------------------------------- + + +class TestRateLimiter: + def test_allows_up_to_the_limit_then_blocks(self): + limiter = RateLimiter(max_calls=3, window_seconds=60) + assert [limiter.is_allowed("ip") for _ in range(4)] == [True, True, True, False] + + def test_buckets_are_per_ip(self): + limiter = RateLimiter(max_calls=1, window_seconds=60) + assert limiter.is_allowed("a") is True + assert limiter.is_allowed("b") is True + assert limiter.is_allowed("a") is False + + def test_reset_clears_state(self): + limiter = RateLimiter(max_calls=1, window_seconds=60) + assert limiter.is_allowed("a") is True + assert limiter.is_allowed("a") is False + limiter.reset() + assert limiter.is_allowed("a") is True + + def test_window_expiry_frees_slots(self): + limiter = RateLimiter(max_calls=1, window_seconds=0.05) + assert limiter.is_allowed("a") is True + assert limiter.is_allowed("a") is False + import time + + time.sleep(0.06) + assert limiter.is_allowed("a") is True + + +# --------------------------------------------------------------------------- +# Backend selection +# --------------------------------------------------------------------------- + + +class TestBackendSelection: + def test_build_store_is_sqlite_without_redis_url(self, monkeypatch): + monkeypatch.delenv("REDIS_URL", raising=False) + assert isinstance(build_store(), SQLiteFeedbackStore) + + def test_build_store_falls_back_when_redis_unreachable(self, monkeypatch): + monkeypatch.setenv("REDIS_URL", "redis://127.0.0.1:6390/0") # nothing listening + assert isinstance(build_store(), SQLiteFeedbackStore) + + +# --------------------------------------------------------------------------- +# Export script +# --------------------------------------------------------------------------- + + +class TestExport: + def _export_module(self): + import importlib + import scripts.export_eval_candidates as exp + + return importlib.reload(exp) + + def test_down_rated_records_become_candidates(self, store): + store.upsert(make_record(message_id="a", categories=["incorrect_information"])) + exp = self._export_module() + candidates = exp.build_candidates(store.list_records(rating="down")) + assert len(candidates) == 1 + c = candidates[0] + assert c["needs_review"] is True + assert c["source"] == "user_feedback" + assert c["category"] == "factual_accuracy" # taxonomy -> harness mapping + assert c["answer_draft"] == "An incorrect paraphrase." + assert "expected" not in c # never fabricates a ground-truth answer + + def test_records_without_a_prompt_are_skipped(self, store): + store.upsert(make_record(prompt=None)) + exp = self._export_module() + assert exp.build_candidates(store.list_records(rating="down")) == [] + + def test_near_duplicate_prompts_are_deduplicated(self, store): + store.upsert(make_record(message_id="a", prompt="What is zakat?")) + store.upsert(make_record(message_id="b", prompt="what is ZAKAT?")) + exp = self._export_module() + candidates = exp.build_candidates(store.list_records(rating="down")) + assert len(candidates) == 1 + + def test_min_categories_filter(self, store): + store.upsert(make_record(message_id="a", categories=["too_long"])) + store.upsert(make_record(message_id="b", categories=["too_long", "too_vague"])) + exp = self._export_module() + records = store.list_records(rating="down") + assert len(exp.build_candidates(records, min_categories=2)) == 1 + + def test_export_honors_db_path(self, store, tmp_path): + store.upsert(make_record()) + exp = self._export_module() + out = tmp_path / "candidates.jsonl" + count = exp.export(output_path=str(out), db_path=store._db_path) + assert count == 1 + line = json.loads(out.read_text().strip()) + assert line["question"] == "What does Surah al-Asr say?" + + def test_export_without_db_uses_configured_backend(self, tmp_path, monkeypatch): + """No --db must read the live store, not a hardcoded feedback.db.""" + monkeypatch.delenv("REDIS_URL", raising=False) + monkeypatch.setenv("FEEDBACK_DB_PATH", str(tmp_path / "live.db")) + import feedback as fb + + live = SQLiteFeedbackStore(db_path=str(tmp_path / "live.db")) + live.upsert(make_record()) + monkeypatch.setattr(fb, "build_store", lambda: live) + exp = self._export_module() + monkeypatch.setattr(exp, "build_store", lambda: live) + assert exp.export(output_path=None, db_path=None) == 1 + + +# --------------------------------------------------------------------------- +# Endpoints +# --------------------------------------------------------------------------- + +ADMIN_TOKEN = "test-admin-token" + + +@pytest.fixture() +def client(tmp_path, monkeypatch): + """A TestClient with feedback pointed at a temp DB and Gemini stubbed.""" + from unittest.mock import AsyncMock, MagicMock + + import main + + temp_store = SQLiteFeedbackStore(db_path=str(tmp_path / "feedback.db")) + monkeypatch.setattr(main, "feedback_store", temp_store) + monkeypatch.setattr(feedback, "store", temp_store) + monkeypatch.setattr(main, "ADMIN_TOKEN", ADMIN_TOKEN) + + # Reset the shared rate limiter so counts never leak between tests. + main.rate_limiter.reset() + + # Isolate chat state and neutralize the heavy model pipeline. + monkeypatch.setattr(main, "GEMINI_API_KEY", "test-key") + monkeypatch.setattr(main, "active_chats", {}) + monkeypatch.setattr(main, "chat_message_ids", {}) + monkeypatch.setattr(main, "answer_snapshots", main.OrderedDict()) + monkeypatch.setenv("SAFETY_PIPELINE_ENABLED", "false") + monkeypatch.setattr(main, "SEMANTIC_CACHE_ENABLED", False) + monkeypatch.setattr(main, "zakat_retriever", AsyncMock(return_value=None)) + monkeypatch.setattr(main, "tafsir_retriever", AsyncMock(return_value=None)) + monkeypatch.setattr(main, "enqueue_for_review", AsyncMock()) + + def fake_model(*args, **kwargs): + session = MagicMock() + session.history = [] + + async def send_message_async(message, **kw): + resp = MagicMock() + resp.text = "A model answer." + resp.candidates = [MagicMock(finish_reason="STOP")] + resp.prompt_feedback = None # avoid the safety-block false positive + session.history = [ + MagicMock(role="user", parts=[MagicMock(text=message)]), + MagicMock(role="model", parts=[MagicMock(text="A model answer.")]), + ] + return resp + + session.send_message_async = send_message_async + session.send_message = MagicMock() + model = MagicMock() + model.start_chat.return_value = session + return model + + monkeypatch.setattr(main.genai, "GenerativeModel", fake_model) + monkeypatch.setattr(main, "get_model", lambda *a, **k: fake_model()) + + transport = ASGITransport(app=main.app) + return AsyncClient(transport=transport, base_url="http://test") + + +def run(coro): + return asyncio.get_event_loop().run_until_complete(coro) + + +@pytest.mark.asyncio +class TestFeedbackEndpoint: + async def _one_answer(self, client): + """Drive a chat turn and return (chat_id, message_id).""" + resp = await client.post("/chat", json={"prompt": "What is zakat?"}) + assert resp.status_code == 200, resp.text + body = resp.json() + return body["chat_id"], body["message_id"] + + async def test_chat_response_carries_message_id(self, client): + async with client: + chat_id, message_id = await self._one_answer(client) + assert message_id + # And the model turn in history is tagged with it. + resp = await client.post( + "/chat", json={"prompt": "again", "chat_id": chat_id} + ) + history = resp.json()["history"] + model_turns = [m for m in history if m["role"] == "model"] + assert all(m["message_id"] for m in model_turns) + + async def test_submit_down_feedback_from_live_snapshot(self, client): + async with client: + chat_id, message_id = await self._one_answer(client) + resp = await client.post("/feedback", json={ + "chat_id": chat_id, + "message_id": message_id, + "rating": "down", + "categories": ["incorrect_information"], + "comment": "wrong", + }) + assert resp.status_code == 200 + fid = resp.json()["feedback_id"] + + import main + stored = main.feedback_store.get(chat_id, message_id) + assert stored.rating == "down" + assert stored.feedback_id == fid + # Snapshot was resolved server-side, not left to the client, and it + # captures the *displayed* answer β€” including any confidence or + # hadith note appended after generation, which is what the user saw. + assert stored.prompt == "What is zakat?" + assert stored.answer.startswith("A model answer.") + + async def test_up_feedback_needs_no_categories(self, client): + async with client: + chat_id, message_id = await self._one_answer(client) + resp = await client.post("/feedback", json={ + "chat_id": chat_id, "message_id": message_id, "rating": "up", + }) + assert resp.status_code == 200 + + @pytest.mark.parametrize("payload,field", [ + ({"rating": "sideways"}, "rating"), + ({"rating": "down", "categories": ["not_a_category"]}, "categories"), + ({"rating": "down", "comment": "x" * 1001}, "comment"), + ]) + async def test_invalid_body_is_422(self, client, payload, field): + async with client: + chat_id, message_id = await self._one_answer(client) + payload = {"chat_id": chat_id, "message_id": message_id, **payload} + resp = await client.post("/feedback", json=payload) + assert resp.status_code == 422 + + async def test_missing_snapshot_requires_client_prompt_and_answer(self, client): + async with client: + resp = await client.post("/feedback", json={ + "chat_id": "gone", "message_id": "gone", "rating": "down", + }) + assert resp.status_code == 422 + + async def test_missing_snapshot_accepts_client_supplied_pair(self, client): + async with client: + resp = await client.post("/feedback", json={ + "chat_id": "gone", "message_id": "gone", "rating": "down", + "prompt": "client prompt", "answer": "client answer", + }) + assert resp.status_code == 200 + import main + stored = main.feedback_store.get("gone", "gone") + assert stored.prompt == "client prompt" + assert stored.answer == "client answer" + + async def test_resubmission_overwrites(self, client): + async with client: + chat_id, message_id = await self._one_answer(client) + base = {"chat_id": chat_id, "message_id": message_id} + await client.post("/feedback", json={**base, "rating": "down"}) + await client.post("/feedback", json={**base, "rating": "up"}) + import main + assert main.feedback_store.get(chat_id, message_id).rating == "up" + assert len(main.feedback_store.list_records()) == 1 + + async def test_rate_limit_blocks_a_flood(self, client, monkeypatch): + import main + monkeypatch.setattr(main.rate_limiter, "_max", 3) + async with client: + chat_id, message_id = await self._one_answer(client) + body = {"chat_id": chat_id, "message_id": message_id, "rating": "up"} + statuses = [(await client.post("/feedback", json=body)).status_code for _ in range(5)] + assert 429 in statuses + + +@pytest.mark.asyncio +class TestAdminEndpoints: + async def test_stats_requires_a_token(self, client): + async with client: + assert (await client.get("/feedback/stats")).status_code == 403 + + async def test_records_requires_a_token(self, client): + async with client: + assert (await client.get("/feedback/records")).status_code == 403 + + async def test_wrong_token_is_403(self, client): + async with client: + resp = await client.get("/feedback/stats", headers={"X-Admin-Token": "wrong"}) + assert resp.status_code == 403 + + async def test_unconfigured_token_disables_admin(self, client, monkeypatch): + import main + monkeypatch.setattr(main, "ADMIN_TOKEN", "") + async with client: + resp = await client.get("/feedback/stats", headers={"X-Admin-Token": "anything"}) + assert resp.status_code == 503 + + async def test_stats_with_token(self, client): + async with client: + resp = await client.get("/feedback/stats", headers={"X-Admin-Token": ADMIN_TOKEN}) + assert resp.status_code == 200 + assert "total" in resp.json() + + async def test_records_filter_validation(self, client): + async with client: + headers = {"X-Admin-Token": ADMIN_TOKEN} + assert (await client.get("/feedback/records?rating=bogus", headers=headers)).status_code == 422 + assert (await client.get("/feedback/records?category=bogus", headers=headers)).status_code == 422 + assert (await client.get("/feedback/records?limit=0", headers=headers)).status_code == 422 + + async def test_records_returns_flagged_items(self, client): + async with client: + resp = await client.post("/chat", json={"prompt": "What is zakat?"}) + chat_id, message_id = resp.json()["chat_id"], resp.json()["message_id"] + await client.post("/feedback", json={ + "chat_id": chat_id, "message_id": message_id, + "rating": "down", "categories": ["too_vague"], + }) + resp = await client.get( + "/feedback/records?rating=down", headers={"X-Admin-Token": ADMIN_TOKEN} + ) + assert resp.status_code == 200 + records = resp.json()["records"] + assert len(records) == 1 + assert records[0]["rating"] == "down" From 8b619bd56a82a1addb03b2635feb9254d9724a17 Mon Sep 17 00:00:00 2001 From: Fury03 Date: Mon, 27 Jul 2026 16:56:48 +0100 Subject: [PATCH 2/2] fix: address review on the feedback loop (index hygiene, bounds, auth) Resolves the CodeRabbit findings on the feedback PR. feedback.py - env_int() guards the rate-limit knobs so a malformed value logs and falls back instead of raising at import and taking down boot (main.py imports this module). - RateLimiter sweeps buckets whose window has fully passed, so a flood of distinct client-controlled IPs can no longer grow _buckets without bound. - Redis upsert now reads the prior record and removes its stale rating/category/model index memberships in the same pipeline, so a re-rating (down->up) or a changed category set can't leave dangling members that double-count in stats() and mislead list_records(). - Daily stats limit distinct days, not grouped rows, so the "by day" window is a full 14 days rather than as few as 7. main.py - FeedbackRequest bounds chat_id, message_id, prompt, answer, and the categories list, so an anonymous caller on the snapshot-miss path can't fill the store with multi-megabyte bodies (the rate limiter caps count, not size). The redundant comment validator is dropped for a field cap. - GENERATION_CONFIG is now the single source consumed by generate(), so a feedback record can't attest to settings that were never used. - The admin token is compared as bytes; compare_digest raises on a non-ASCII str, which would have turned a crafted header into a 500. - X-Forwarded-For is honored only when TRUST_PROXY_HEADERS is set, and then the rightmost (proxy-observed) hop is used, so header rotation can't mint fresh rate-limit buckets on the unauthenticated write endpoint. - The three feedback handlers chain their 500s with `from exc`. export_eval_candidates.py - Rejects negative --limit / --min-categories (a negative limit reaches SQLite as "no limit"). Tests: +18 (56 total) covering env_int fallback, bucket reclamation, the Redis index-hygiene paths via an in-memory fake, oversized-field 422s, the non-ASCII-token 403, the proxy-trust gating, and the CLI guards. --- feedback.py | 63 +++++++++- main.py | 67 ++++++----- scripts/export_eval_candidates.py | 7 ++ tests/test_feedback.py | 187 ++++++++++++++++++++++++++++++ 4 files changed, 291 insertions(+), 33 deletions(-) diff --git a/feedback.py b/feedback.py index 22a237a..bab65a7 100644 --- a/feedback.py +++ b/feedback.py @@ -53,6 +53,28 @@ COMMENT_MAX_CHARS = 1000 + +def env_int(name: str, default: int, minimum: int = 1) -> int: + """Read a positive int from the environment, falling back on nonsense. + + A malformed tuning value must not crash boot: main.py imports this module, + so an unguarded ``int(os.getenv(...))`` here would take the whole app down + with a traceback instead of degrading to a sane default. + """ + raw = os.getenv(name) + if raw is None: + return default + try: + value = int(raw) + except ValueError: + logger.warning("%s=%r is not an integer; using %s", name, raw, default) + return default + if value < minimum: + logger.warning("%s=%s is below the minimum %s; using %s", name, value, minimum, default) + return default + return value + + # Redis TTL for feedback records (30 days). REDIS_TTL_SECONDS = 60 * 60 * 24 * 30 @@ -60,8 +82,8 @@ SQLITE_MAX_RECORDS = 50_000 # Rate limiting: max submissions per IP per window. -RATE_LIMIT_MAX = int(os.getenv("FEEDBACK_RATE_LIMIT_MAX", "20")) -RATE_LIMIT_WINDOW_SECONDS = int(os.getenv("FEEDBACK_RATE_LIMIT_WINDOW", "60")) +RATE_LIMIT_MAX = env_int("FEEDBACK_RATE_LIMIT_MAX", 20) +RATE_LIMIT_WINDOW_SECONDS = env_int("FEEDBACK_RATE_LIMIT_WINDOW", 60) # --------------------------------------------------------------------------- @@ -86,6 +108,7 @@ def is_allowed(self, ip: str) -> bool: now = time.monotonic() cutoff = now - self._window with self._lock: + self._sweep(cutoff) bucket = self._buckets[ip] while bucket and bucket[0] < cutoff: bucket.popleft() @@ -94,6 +117,19 @@ def is_allowed(self, ip: str) -> bool: bucket.append(now) return True + def _sweep(self, cutoff: float) -> None: + """Drop buckets whose newest timestamp is outside the window. + + Without this, one entry accumulates per distinct IP and is never + reclaimed. The key comes from a client-controlled X-Forwarded-For, so + an attacker could otherwise grow this dict without bound. A bucket + whose most-recent hit is older than the window can hold nothing live, + so it is safe to drop entirely. + """ + stale = [ip for ip, bucket in self._buckets.items() if not bucket or bucket[-1] < cutoff] + for ip in stale: + del self._buckets[ip] + def reset(self) -> None: """Clear all buckets. Used by tests so limiter state never leaks between them.""" with self._lock: @@ -341,9 +377,17 @@ def stats(self) -> Dict[str, Any]: bucket = model_counts.setdefault(name, {"up": 0, "down": 0}) bucket[r["rating"]] = r["cnt"] + # Limit distinct *days*, not grouped rows: GROUP BY day, rating yields + # up to two rows per day, so a plain LIMIT 14 would return as few as + # seven days when both ratings occur. day_rows = conn.execute( "SELECT substr(created_at,1,10) as day, rating, COUNT(*) as cnt " - "FROM feedback GROUP BY day, rating ORDER BY day DESC LIMIT 14" + "FROM feedback " + "WHERE substr(created_at,1,10) IN (" + " SELECT DISTINCT substr(created_at,1,10) FROM feedback " + " ORDER BY 1 DESC LIMIT 14" + ") " + "GROUP BY day, rating ORDER BY day DESC" ).fetchall() by_day: Dict[str, Dict[str, int]] = {} for r in day_rows: @@ -385,12 +429,25 @@ def _record_key(self, chat_id: str, message_id: str) -> str: def upsert(self, record: FeedbackRecord) -> None: key = self._record_key(record.chat_id, record.message_id) ts = time.time() + + # Idempotent overwrite must also fix the indexes: a re-rating (down->up) + # or a changed category set would otherwise leave the key in the old + # rating/category sorted sets forever, so list_records and stats would + # double-count it. Remove the previous memberships before re-adding. + previous = self.get(record.chat_id, record.message_id) + data = record.to_dict() data["categories"] = json.dumps(data["categories"]) data["generation_config"] = ( json.dumps(data["generation_config"]) if data["generation_config"] else "" ) pipe = self._r.pipeline() + if previous is not None: + pipe.zrem(f"{self._PREFIX}:index:rating:{previous.rating}", key) + for cat in previous.categories: + pipe.zrem(f"{self._PREFIX}:index:cat:{cat}", key) + pipe.zrem(f"{self._PREFIX}:index:model:{previous.model_name or 'unknown'}", key) + pipe.hset(key, mapping={k: (v if v is not None else "") for k, v in data.items()}) pipe.expire(key, REDIS_TTL_SECONDS) pipe.zadd(f"{self._PREFIX}:index:rating:{record.rating}", {key: ts}) diff --git a/main.py b/main.py index 5337f03..05051bd 100644 --- a/main.py +++ b/main.py @@ -70,6 +70,7 @@ COMMENT_MAX_CHARS, FEEDBACK_TAXONOMY, FeedbackRecord, + env_int, rate_limiter, store as feedback_store, ) @@ -167,14 +168,17 @@ class ChatResponse(BaseModel): class FeedbackRequest(BaseModel): - chat_id: str - message_id: str + chat_id: str = Field(..., max_length=200) + message_id: str = Field(..., max_length=200) rating: str = Field(..., description="'up' or 'down'") - categories: Optional[List[str]] = None - comment: Optional[str] = None - # Supplied by the client when the session is no longer in memory (restart). - prompt: Optional[str] = None - answer: Optional[str] = None + # At most one tag per taxonomy category; a caller cannot pad the list. + categories: Optional[List[str]] = Field(None, max_length=len(FEEDBACK_TAXONOMY)) + comment: Optional[str] = Field(None, max_length=COMMENT_MAX_CHARS) + # Supplied by the client when the snapshot is gone (restart). Bounded so an + # anonymous caller cannot fill the store with multi-megabyte bodies β€” the + # rate limiter caps request count, not request size. + prompt: Optional[str] = Field(None, max_length=8000) + answer: Optional[str] = Field(None, max_length=16000) @field_validator("rating") @classmethod @@ -196,15 +200,6 @@ def categories_must_be_valid(cls, v: Optional[List[str]]) -> Optional[List[str]] ) return v - @field_validator("comment") - @classmethod - def comment_length(cls, v: Optional[str]) -> Optional[str]: - if v and len(v) > COMMENT_MAX_CHARS: - raise ValueError( - f"comment must not exceed {COMMENT_MAX_CHARS} characters (got {len(v)})" - ) - return v - def classify_for_safety(prompt: str, candidate_ids: List[str]): """Gemini classifier seam; offline tests replace this with a fixture.""" @@ -321,9 +316,14 @@ def get_safety_settings(): # is the displayed text (post safety/hadith/abstention shaping), not the raw # model output. Bounded LRU β€” on a free-tier restart it is empty, which is why # the feedback endpoint also accepts a client-supplied prompt/answer. -FEEDBACK_SNAPSHOT_MAX = int(os.getenv("FEEDBACK_SNAPSHOT_MAX", "5000")) +FEEDBACK_SNAPSHOT_MAX = env_int("FEEDBACK_SNAPSHOT_MAX", 5000) answer_snapshots: "OrderedDict[tuple, Dict[str, str]]" = OrderedDict() +# Only honor X-Forwarded-For when the deployment actually sits behind a proxy we +# control; otherwise any client can rotate the header to mint a fresh rate-limit +# bucket on every request and defeat the only control on the write endpoint. +TRUST_PROXY_HEADERS = os.getenv("TRUST_PROXY_HEADERS", "false").lower() in {"1", "true", "yes"} + def _record_answer(chat_id: str, prompt: str, answer: str) -> str: """Assign a message id to a fresh answer, snapshot it, and return the id.""" @@ -358,7 +358,9 @@ async def require_admin(token: Optional[str] = Depends(_admin_header)) -> None: status_code=503, detail="ADMIN_TOKEN is not configured on this server.", ) - if not token or not secrets.compare_digest(token, ADMIN_TOKEN): + # Compare as bytes: secrets.compare_digest raises on non-ASCII str, which + # would turn a crafted header into a 500 instead of a clean 403. + if not token or not secrets.compare_digest(token.encode("utf-8"), ADMIN_TOKEN.encode("utf-8")): raise HTTPException(status_code=403, detail="Invalid or missing admin token.") ISLAMIC_CONTEXT = ( @@ -676,12 +678,7 @@ async def generate(safety_prompt: str) -> str: response = await send_message_with_retry( active_chats[chat_id], full_prompt, - generation_config={ - "temperature": 0.7, - "top_p": 0.8, - "top_k": 40, - "max_output_tokens": 2048, - }, + generation_config=GENERATION_CONFIG, ) telemetry.record_model_call( response, @@ -1092,9 +1089,19 @@ async def delete_chat(chat_id: str): # --------------------------------------------------------------------------- def _client_ip(request: Request) -> str: - forwarded = request.headers.get("X-Forwarded-For") - if forwarded: - return forwarded.split(",")[0].strip() + """Best-effort client IP for rate limiting. + + X-Forwarded-For is honored only when TRUST_PROXY_HEADERS is set β€” otherwise + a client can rotate the header to get a fresh bucket every request. Even + when trusted, take the rightmost hop (the address our proxy saw) rather than + the leftmost, which the client controls. + """ + if TRUST_PROXY_HEADERS: + forwarded = request.headers.get("X-Forwarded-For") + if forwarded: + hops = [h.strip() for h in forwarded.split(",") if h.strip()] + if hops: + return hops[-1] return request.client.host if request.client else "unknown" @@ -1148,7 +1155,7 @@ async def submit_feedback(request: Request, body: FeedbackRequest): await run_in_threadpool(feedback_store.upsert, record) except Exception as exc: logger.error("Failed to store feedback: %s", exc) - raise HTTPException(status_code=500, detail="Failed to store feedback.") + raise HTTPException(status_code=500, detail="Failed to store feedback.") from exc logger.info( "Feedback stored: chat_id=%s message_id=%s rating=%s", @@ -1167,7 +1174,7 @@ async def feedback_stats(): return await run_in_threadpool(feedback_store.stats) except Exception as exc: logger.error("Failed to fetch feedback stats: %s", exc) - raise HTTPException(status_code=500, detail="Failed to fetch stats.") + raise HTTPException(status_code=500, detail="Failed to fetch stats.") from exc @app.get("/feedback/records", dependencies=[Depends(require_admin)]) @@ -1196,7 +1203,7 @@ async def feedback_records( return {"records": [r.to_dict() for r in records]} except Exception as exc: logger.error("Failed to fetch feedback records: %s", exc) - raise HTTPException(status_code=500, detail="Failed to fetch records.") + raise HTTPException(status_code=500, detail="Failed to fetch records.") from exc @app.get("/ping") diff --git a/scripts/export_eval_candidates.py b/scripts/export_eval_candidates.py index 0a55b73..2f64d34 100644 --- a/scripts/export_eval_candidates.py +++ b/scripts/export_eval_candidates.py @@ -175,6 +175,13 @@ def main() -> None: ) args = parser.parse_args() + # A negative --limit reaches SQLite as "no limit" and loads the whole table + # before dedup; a negative --min-categories is meaningless. Reject both. + if args.limit < 0: + parser.error("--limit must be a non-negative integer") + if args.min_categories < 0: + parser.error("--min-categories must be a non-negative integer") + count = export( output_path=args.output, min_categories=args.min_categories, diff --git a/tests/test_feedback.py b/tests/test_feedback.py index 1a83fd3..97e1917 100644 --- a/tests/test_feedback.py +++ b/tests/test_feedback.py @@ -16,8 +16,10 @@ from feedback import ( FeedbackRecord, RateLimiter, + RedisFeedbackStore, SQLiteFeedbackStore, build_store, + env_int, ) @@ -137,6 +139,123 @@ def test_reset_clears_state(self): limiter.reset() assert limiter.is_allowed("a") is True + def test_stale_buckets_are_reclaimed(self): + """A flood of one-shot IPs must not grow _buckets without bound.""" + limiter = RateLimiter(max_calls=5, window_seconds=0.05) + for i in range(50): + limiter.is_allowed(f"ip-{i}") + assert len(limiter._buckets) == 50 + import time + + time.sleep(0.06) + # One more call sweeps everything now outside the window. + limiter.is_allowed("fresh") + assert len(limiter._buckets) == 1 + + +class TestEnvInt: + def test_reads_a_valid_value(self, monkeypatch): + monkeypatch.setenv("SOME_KNOB", "42") + assert env_int("SOME_KNOB", 10) == 42 + + @pytest.mark.parametrize("raw", ["not-a-number", "", "0", "-3"]) + def test_bad_value_falls_back_instead_of_raising(self, monkeypatch, raw): + monkeypatch.setenv("SOME_KNOB", raw) + assert env_int("SOME_KNOB", 10) == 10 + + def test_unset_uses_default(self, monkeypatch): + monkeypatch.delenv("SOME_KNOB", raising=False) + assert env_int("SOME_KNOB", 7) == 7 + + +class FakeRedisPipeline: + def __init__(self, client): + self._client = client + self._ops = [] + + def hset(self, key, mapping): + self._ops.append(("hset", key, mapping)) + return self + + def expire(self, key, ttl): + self._ops.append(("expire", key, ttl)) + return self + + def zadd(self, key, mapping): + self._ops.append(("zadd", key, mapping)) + return self + + def zrem(self, key, member): + self._ops.append(("zrem", key, member)) + return self + + def execute(self): + for op in self._ops: + if op[0] == "hset": + self._client.hashes[op[1]] = dict(op[2]) + elif op[0] == "zadd": + self._client.zsets.setdefault(op[1], {}).update(op[2]) + elif op[0] == "zrem": + self._client.zsets.get(op[1], {}).pop(op[2], None) + self._ops.clear() + + +class FakeRedis: + """Just enough Redis for RedisFeedbackStore's index bookkeeping.""" + + def __init__(self): + self.hashes = {} + self.zsets = {} + + def pipeline(self): + return FakeRedisPipeline(self) + + def hgetall(self, key): + return dict(self.hashes.get(key, {})) + + def zcard(self, key): + return len(self.zsets.get(key, {})) + + def zrevrange(self, key, start, stop): + members = sorted(self.zsets.get(key, {}).items(), key=lambda kv: kv[1], reverse=True) + return [m for m, _ in members][start:stop + 1] + + +class TestRedisIndexHygiene: + """Re-rating and re-tagging must not leave dangling index memberships.""" + + def _store(self): + return RedisFeedbackStore(FakeRedis()) + + def test_rerating_moves_the_index_membership(self): + store = self._store() + store.upsert(make_record(rating="down")) + assert store._r.zcard("feedback:index:rating:down") == 1 + assert store._r.zcard("feedback:index:rating:up") == 0 + + store.upsert(make_record(rating="up")) + # The old down membership is gone; not counted in both. + assert store._r.zcard("feedback:index:rating:down") == 0 + assert store._r.zcard("feedback:index:rating:up") == 1 + + def test_removed_category_is_dropped_from_its_index(self): + store = self._store() + store.upsert(make_record(categories=["too_long", "too_vague"])) + assert store._r.zcard("feedback:index:cat:too_long") == 1 + + store.upsert(make_record(categories=["too_vague"])) + assert store._r.zcard("feedback:index:cat:too_long") == 0 + assert store._r.zcard("feedback:index:cat:too_vague") == 1 + + def test_stats_are_not_double_counted_after_rerating(self): + store = self._store() + store.upsert(make_record(rating="down")) + store.upsert(make_record(rating="up")) + stats = store.stats() + assert stats["total"] == 1 + assert stats["down"] == 0 + assert stats["up"] == 1 + def test_window_expiry_frees_slots(self): limiter = RateLimiter(max_calls=1, window_seconds=0.05) assert limiter.is_allowed("a") is True @@ -227,6 +346,17 @@ def test_export_without_db_uses_configured_backend(self, tmp_path, monkeypatch): monkeypatch.setattr(exp, "build_store", lambda: live) assert exp.export(output_path=None, db_path=None) == 1 + @pytest.mark.parametrize("argv", [ + ["--limit", "-1"], + ["--min-categories", "-2"], + ]) + def test_cli_rejects_negative_values(self, argv, monkeypatch): + exp = self._export_module() + monkeypatch.setattr(exp.sys, "argv", ["export_eval_candidates.py", *argv]) + with pytest.raises(SystemExit) as excinfo: + exp.main() + assert excinfo.value.code != 0 + # --------------------------------------------------------------------------- # Endpoints @@ -349,6 +479,10 @@ async def test_up_feedback_needs_no_categories(self, client): ({"rating": "sideways"}, "rating"), ({"rating": "down", "categories": ["not_a_category"]}, "categories"), ({"rating": "down", "comment": "x" * 1001}, "comment"), + # Oversized snapshot fields must be rejected at the boundary, not stored. + ({"rating": "down", "prompt": "x" * 8001}, "prompt"), + ({"rating": "down", "answer": "x" * 16001}, "answer"), + ({"rating": "down", "categories": ["other"] * 50}, "categories"), ]) async def test_invalid_body_is_422(self, client, payload, field): async with client: @@ -395,6 +529,44 @@ async def test_rate_limit_blocks_a_flood(self, client, monkeypatch): statuses = [(await client.post("/feedback", json=body)).status_code for _ in range(5)] assert 429 in statuses + async def test_forwarded_header_ignored_unless_proxy_trusted(self, client, monkeypatch): + """Header rotation must not mint fresh buckets when proxy is untrusted.""" + import main + monkeypatch.setattr(main, "TRUST_PROXY_HEADERS", False) + monkeypatch.setattr(main.rate_limiter, "_max", 2) + async with client: + chat_id, message_id = await self._one_answer(client) + body = {"chat_id": chat_id, "message_id": message_id, "rating": "up"} + statuses = [] + for i in range(4): + resp = await client.post( + "/feedback", json=body, headers={"X-Forwarded-For": f"9.9.9.{i}"} + ) + statuses.append(resp.status_code) + # A rotating X-Forwarded-For gave no new buckets, so the limit bit. + assert 429 in statuses + + async def test_trusted_proxy_uses_rightmost_hop(self, client, monkeypatch): + import main + monkeypatch.setattr(main, "TRUST_PROXY_HEADERS", True) + captured = {} + real_is_allowed = main.rate_limiter.is_allowed + + def spy(ip): + captured["ip"] = ip + return real_is_allowed(ip) + + monkeypatch.setattr(main.rate_limiter, "is_allowed", spy) + async with client: + chat_id, message_id = await self._one_answer(client) + await client.post( + "/feedback", + json={"chat_id": chat_id, "message_id": message_id, "rating": "up"}, + headers={"X-Forwarded-For": "1.1.1.1, 2.2.2.2, 3.3.3.3"}, + ) + # Rightmost hop (what our proxy saw), not the client-controlled left. + assert captured["ip"] == "3.3.3.3" + @pytest.mark.asyncio class TestAdminEndpoints: @@ -411,6 +583,21 @@ async def test_wrong_token_is_403(self, client): resp = await client.get("/feedback/stats", headers={"X-Admin-Token": "wrong"}) assert resp.status_code == 403 + async def test_non_ascii_token_is_403_not_500(self, monkeypatch): + """compare_digest raises on non-ASCII str; the bytes compare must not. + + Starlette decodes headers as latin-1, so a non-ASCII token can reach + require_admin as a str. httpx itself won't send such a header, so this + exercises the dependency directly. + """ + import main + from fastapi import HTTPException + + monkeypatch.setattr(main, "ADMIN_TOKEN", ADMIN_TOKEN) + with pytest.raises(HTTPException) as exc: + await main.require_admin(token="tΓΈken-ΓΌnicode") + assert exc.value.status_code == 403 + async def test_unconfigured_token_disables_admin(self, client, monkeypatch): import main monkeypatch.setattr(main, "ADMIN_TOKEN", "")