diff --git a/.env.example b/.env.example index 92a58d4..cc43bf8 100644 --- a/.env.example +++ b/.env.example @@ -20,3 +20,8 @@ JWT_SECRET=change-me-too-long-random-string JWT_EXPIRE_HOURS=12 AUTO_START_WHISPER=true + +# Persistance des sessions (transcripts + claims + métriques) +PERSIST_SESSIONS=true +# URL SQLAlchemy (défaut : fichier SQLite local) +DATABASE_URL=sqlite:///./livefactchecker.db diff --git a/.gitignore b/.gitignore index 623423c..65b4c0b 100644 --- a/.gitignore +++ b/.gitignore @@ -62,6 +62,12 @@ local_settings.py db.sqlite3 db.sqlite3-journal +# Local SQLite session store +*.db +*.db-journal +*.sqlite3-wal +*.sqlite3-shm + # Flask stuff: instance/ .webassets-cache diff --git a/TODO.md b/TODO.md index 5d929e0..bcc9c02 100644 --- a/TODO.md +++ b/TODO.md @@ -8,25 +8,27 @@ La source de vérité reste le code, pas ce fichier (cf. CLAUDE.md racine). Le produit est aujourd'hui sans état : les claims vivent le temps d'une connexion WS, rien n'est conservé, un seul utilisateur, vérification figée en français. -- [ ] **Persistance des sessions & claims** : stocker chaque session (transcript + claims vérifiés) en base (SQLite suffit pour commencer) pour pouvoir rejouer, exporter et analyser après coup. Prérequis à la plupart des features ci-dessous. -- [ ] **Export d'une session** : générer un récap (Markdown / PDF / JSON) de tous les claims d'une session — texte, statut, explication, sources, score de confiance. -- [ ] **Historique consultable** : endpoint `/sessions` (liste) + `/sessions/{id}` (détail) pour relire une vérification passée hors live. +- [x] **Persistance des sessions & claims** : SQLAlchemy + SQLite (`app/db/`, modèles `app/models/`). Écriture best-effort au fil de l'eau depuis `session.py` via `app/services/session_store.py` (offload `to_thread`). Le segment porte les mesures de la passe (tokens, latences, `api_calls`, `web_search`) ; tout le reste est dérivé à la lecture. Création **paresseuse** : une session sans aucun transcript ne crée pas de ligne. Réglé par `PERSIST_SESSIONS` (défaut `true`) + `DATABASE_URL`. +- [x] **Export d'une session** : `GET /sessions/{id}/export?format=md|json` (formatteur `app/services/export.py`). PDF non fait (Markdown/JSON seulement). +- [x] **Historique consultable** : `GET /sessions` (liste) + `GET /sessions/{id}` (détail), **gated admin** (`require_admin`). - [ ] **Dédoublonnage des claims** : un même fait répété sur plusieurs chunks de 5 s crée aujourd'hui des claims distincts. Détecter les quasi-doublons (similarité du `text`) et fusionner / ne pas re-vérifier — économise des appels Anthropic. - [ ] **Cache de vérification** : mémoriser le résultat d'un claim déjà vérifié (clé = texte normalisé) pour ne pas repayer un appel sur une affirmation identique. - [ ] **Webhook / notification sur claim "false"** : pousser une alerte (webhook configurable) quand un fait est démenti, pour intégration externe (overlay OBS, Slack…). -- [ ] **Stats agrégées par session** : ratio vrai/faux/incertain, catégorie dominante, taux de recours au web — exposé en fin de session et dans l'admin. +- [x] **Stats agrégées par session** : `app/services/stats.py` (`compute_stats`) — ratio par statut, catégorie dominante, confiance moyenne, taux/usage web_search, tokens, latences, rejets, coût € estimé. Exposé dans `/sessions/{id}` et la page admin Sessions. ⚠️ Tarifs `PRICING` dans `stats.py` à vérifier (coût `None` si modèle non tarifé). - [ ] **Multilingue (prompt Claude)** : la transcription tourne toujours en auto-détection ; la langue choisie par session sert de *filtre* (les chunks d'une autre langue sont ignorés, voir `core/languages.py` + `ConfigMessage` + le filtre dans `session.py`). Reste à adapter `SYSTEM_PROMPT` et l'enum de catégories de `claim_extractor.py` à la langue de la session — actuellement figés FR, donc les claims sortent en français même pour un audio non francophone. - [x] **Niveau de vérification réglable** : le message WS `config` porte un champ `verification_level` (`fast` / `thorough`, défaut `thorough`). `fast` n'offre pas l'outil `web_search` (un seul appel, connaissances internes) ; `thorough` le rend disponible (comportement antérieur). Plumbing `session.py` → `extract_and_verify(web_search=…)`. ## Tests (priorité haute) -Aujourd'hui seul `tests/test_claim_extractor.py` existe. Manquent : +Plusieurs suites existent désormais (`test_claim_extractor`, `test_extract_usage`, +`test_stats`, `test_export`, `test_sessions_route`, `test_session_persistence`, +`test_session_config`, `test_auth_route`…). Restent : -- [ ] `session.py` : tester `_make_claim`, `_spawn_claims` (skip si < `MIN_WORDS`), le cycle pending → claim/remove_claim. Mocker `extract_and_verify` et le `WebSocket`. +- [ ] `session.py` : `_make_claim`, `_spawn_claims` (skip si < `MIN_WORDS`), le cycle pending → claim/remove_claim. Mocker `extract_and_verify` et le `WebSocket`. (Partiel : `_ensure_persisted` couvert par `test_session_persistence`.) - [ ] Auth : `/admin/login` (bon mot de passe → JWT, mauvais → 401), expiration du token, `require_admin` qui rejette un token absent/invalide/expiré. -- [ ] Routes admin : un test d'intégration par route via `TestClient`, avec `require_admin` overridé (`app.dependency_overrides`). -- [ ] `extract_and_verify` : le fallback deux-tours (web_search sans `submit_claims` → second appel forcé). Mocker le client Anthropic. -- [ ] `_parse_claims` : statut invalide → `uncertain`, `confidence` clampée 0-10, champ `text` manquant → claim ignoré. +- [ ] Routes admin : un test d'intégration par route via `TestClient`, avec `require_admin` overridé (`app.dependency_overrides`). (Fait pour `/sessions/*` dans `test_sessions_route`.) +- [x] `extract_and_verify` : le fallback deux-tours (web_search sans `submit_claims` → second appel forcé) + accumulation usage/tokens. Couvert par `test_extract_usage` (client Anthropic mocké). +- [x] `_parse_claims` : statut invalide → `uncertain`, `confidence` clampée 0-10, champ `text` manquant → claim ignoré. Couvert par `test_claim_extractor` (`test_unknown_status_falls_back_to_uncertain`, `test_confidence_is_clamped_to_0_10`, `test_entries_without_text_are_dropped`). ## Robustesse & sécurité @@ -38,10 +40,10 @@ Aujourd'hui seul `tests/test_claim_extractor.py` existe. Manquent : ## Architecture & dette -- [ ] `_active_sessions` est un dict global au niveau module : OK pour un process unique, mais à documenter comme limite (ne survit pas à plusieurs workers / un restart). +- [x] `_active_sessions` est un dict global au niveau module : OK pour un process unique. Limite documentée en tête de `session.py` — ce n'est pas un cache de la DB (il porte des `asyncio.Task` vivants + une `deque` de contexte non persistés) mais l'état runtime des connexions ouvertes, lu seulement par `/admin/ws/status`. Non remplaçable par des requêtes DB (la base ne sait pas « qui est connecté maintenant »). Reste mono-process : ne survit pas à plusieurs workers / un restart. - [ ] Transcription : toujours en auto-détection ; la langue par session filtre les chunks (cf. ci-dessus). Le prompt/catégories de fact-checking restent figés FR — voir la ligne « Multilingue (prompt Claude) ». ## Observabilité -- [ ] `/admin/logs` lit un buffer en mémoire (`get_logs`) — vérifier qu'il est borné (pas de fuite mémoire sur longue session). +- [x] `/admin/logs` lit un buffer en mémoire (`get_logs`) — déjà borné : `_log_history = deque(maxlen=300)` dans `core/observability.py` évince les entrées les plus anciennes, donc pas de fuite mémoire sur longue session. - [ ] Exposer des métriques agrégées (claims/min, ratio web_search, latence moyenne transcription + vérification) en plus du statut de session brut. diff --git a/app/api/routers/admin.py b/app/api/routers/admin.py index 40a4dd9..144f2ad 100644 --- a/app/api/routers/admin.py +++ b/app/api/routers/admin.py @@ -58,6 +58,8 @@ def _value_type(value: object) -> ValueType: return "bool" if isinstance(value, int): return "int" + if isinstance(value, float): + return "float" if isinstance(value, list): return "list" return "str" diff --git a/app/api/routers/fact_check.py b/app/api/routers/fact_check.py index 8b1e052..e5958e9 100644 --- a/app/api/routers/fact_check.py +++ b/app/api/routers/fact_check.py @@ -18,9 +18,9 @@ async def fact_check( req: FactCheckRequest, _admin: str = Depends(require_admin) ) -> FactCheckResponse: - results = await extract_and_verify(req.text, web_search=req.web_search) + result = await extract_and_verify(req.text, web_search=req.web_search) # Service dicts carry every Claim field except id/timestamp; the response # model fills those with defaults. This route is diagnostic, not the live # WS path, so stable ids/timestamps aren't needed here. - claims = [Claim(id="", **r) for r in results] + claims = [Claim(id="", **r) for r in result.claims] return FactCheckResponse(text=req.text, claims=claims) diff --git a/app/api/routers/sessions.py b/app/api/routers/sessions.py new file mode 100644 index 0000000..8054739 --- /dev/null +++ b/app/api/routers/sessions.py @@ -0,0 +1,111 @@ +"""Session history & export (read-only, admin-gated). + +Browsing past sessions is an admin capability: there is no per-user notion, so the +whole history is global and sits behind ``require_admin``. A regular user exporting +their *own live* session is a separate, client-side path (built from the in-browser +stores), not these routes. + +Handlers are plain ``def`` so FastAPI runs them in its threadpool — the sync +SQLAlchemy session from ``get_db`` never blocks the event loop. +""" + +from typing import Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from fastapi.responses import JSONResponse, PlainTextResponse +from sqlalchemy import select +from sqlalchemy.orm import Session as DBSession + +from app.config import settings +from app.db.session import get_db +from app.dependencies import require_admin +from app.models import Session +from app.schemas.history import ClaimOut, SegmentOut, SessionDetail, SessionSummary +from app.services.export import session_to_markdown +from app.services.stats import compute_stats + +router = APIRouter( + prefix="/sessions", tags=["sessions"], dependencies=[Depends(require_admin)] +) + + +def _summary(session: Session, model: str) -> SessionSummary: + stats = compute_stats(session, model) + return SessionSummary( + id=session.id, + started_at=session.started_at, + ended_at=session.ended_at, + active=session.ended_at is None, + client_host=session.client_host, + transcripts_count=stats.transcripts_count, + claims_count=stats.claims_count, + false_count=stats.claims_by_status.get("false", 0), + estimated_cost_usd=stats.estimated_cost_usd, + ) + + +def _detail(session: Session, model: str) -> SessionDetail: + return SessionDetail( + id=session.id, + started_at=session.started_at, + ended_at=session.ended_at, + active=session.ended_at is None, + client_host=session.client_host, + chunks_received=session.chunks_received, + stats=compute_stats(session, model), + # Explicit ORM → schema conversion (segments come ordered by seq). + segments=[SegmentOut.model_validate(s) for s in session.segments], + claims=[ + ClaimOut.model_validate(c) + for c in sorted(session.claims, key=lambda c: c.timestamp) + ], + ) + + +def _get_or_404(db: DBSession, session_id: str) -> Session: + session = db.get(Session, session_id) + if session is None: + raise HTTPException(status_code=404, detail="Session introuvable") + return session + + +@router.get("", response_model=list[SessionSummary]) +def list_sessions( + db: DBSession = Depends(get_db), + limit: int = Query(default=100, ge=1, le=1000), +) -> list[SessionSummary]: + sessions = ( + db.execute(select(Session).order_by(Session.started_at.desc()).limit(limit)) + .scalars() + .all() + ) + model = settings.ANTHROPIC_MODEL + return [_summary(s, model) for s in sessions] + + +@router.get("/{session_id}", response_model=SessionDetail) +def get_session(session_id: str, db: DBSession = Depends(get_db)) -> SessionDetail: + return _detail(_get_or_404(db, session_id), settings.ANTHROPIC_MODEL) + + +@router.get("/{session_id}/export", response_model=None) +def export_session( + session_id: str, + db: DBSession = Depends(get_db), + fmt: Literal["json", "md"] = Query(default="json", alias="format"), +) -> JSONResponse | PlainTextResponse: + detail = _detail(_get_or_404(db, session_id), settings.ANTHROPIC_MODEL) + if fmt == "md": + return PlainTextResponse( + session_to_markdown(detail), + media_type="text/markdown", + headers={ + "Content-Disposition": f'attachment; filename="session-{session_id}.md"' + }, + ) + return JSONResponse( + detail.model_dump(mode="json"), + headers={ + "Content-Disposition": f'attachment; filename="session-{session_id}.json"' + }, + ) diff --git a/app/config.py b/app/config.py index 0e70af8..f957667 100644 --- a/app/config.py +++ b/app/config.py @@ -20,11 +20,25 @@ class Settings(BaseSettings): WHISPER_MODEL: str = "medium" WHISPER_DEVICE: Literal["cpu", "cuda"] = "cpu" - # Upper bound on a single received audio blob (bytes), on /ws chunks and the - # /admin/whisper/transcribe upload. ~5 s of WebM/Opus is well under this; the - # cap guards against a malformed/oversized blob saturating memory. Default 10 MiB. + # Upper bound on a single received audio frame (bytes), on /ws frames and the + # /admin/whisper/transcribe upload. A /ws PCM frame (~250 ms of 16 kHz mono + # Int16) is a few KiB; the cap guards against a malformed/oversized blob + # saturating memory. Default 10 MiB. MAX_AUDIO_BYTES: int = 10 * 1024 * 1024 + # Voice-activity endpointing for the live /ws stream. The client streams raw + # PCM continuously; the server cuts it into utterances on natural pauses + # (Silero VAD) instead of fixed client-side chunks. See services/audio_endpointer. + # VAD_THRESHOLD: Silero speech probability above which a frame counts as speech. + VAD_THRESHOLD: float = 0.5 + # Trailing silence (ms) that closes an utterance and flushes it for transcription. + VAD_SILENCE_FLUSH_MS: int = 700 + # Force-flush an utterance once it reaches this length, even without a pause + # (keeps a long monologue from delaying feedback indefinitely). + VAD_MAX_SEGMENT_MS: int = 12000 + # Drop a flushed utterance shorter than this (filters out blips/noise). + VAD_MIN_SEGMENT_MS: int = 400 + LOG_LEVEL: str = "INFO" # CORS: origines autorisées pour le front (format JSON dans .env, ex. @@ -43,6 +57,18 @@ class Settings(BaseSettings): AUTO_START_WHISPER: bool = True + # How many preceding utterances are handed to claim extraction as read-only + # context so a sentence that only makes sense after the previous one + # ("Il en est de même de…") can be resolved instead of dropped as unverifiable. + CONTEXT_TURNS: int = 4 + + # Session persistence. PERSIST_SESSIONS gates *writing* sessions, transcripts + # and verified claims to the DB; the live WS path is unaffected when off. The + # tables are always created at startup so the /sessions read routes work + # regardless. DATABASE_URL is any SQLAlchemy URL (default: a local SQLite file). + PERSIST_SESSIONS: bool = True + DATABASE_URL: str = "sqlite:///./livefactchecker.db" + @field_validator("ANTHROPIC_API_KEY") @classmethod def api_key_must_be_set(cls, v: str) -> str: diff --git a/app/core/config_descriptor.py b/app/core/config_descriptor.py index 8e3c8b6..ed20598 100644 --- a/app/core/config_descriptor.py +++ b/app/core/config_descriptor.py @@ -64,7 +64,27 @@ class ConfigBlock: id="audio", title="Audio", fields=( - ConfigField("MAX_AUDIO_BYTES", "Taille max d'un blob (octets)", "editable"), + ConfigField( + "MAX_AUDIO_BYTES", "Taille max d'une frame (octets)", "editable" + ), + ), + ), + ConfigBlock( + id="vad", + title="Découpage VAD (live)", + fields=( + ConfigField( + "VAD_THRESHOLD", "Seuil de détection de parole (0-1)", "editable" + ), + ConfigField( + "VAD_SILENCE_FLUSH_MS", "Silence de fin d'énoncé (ms)", "editable" + ), + ConfigField( + "VAD_MAX_SEGMENT_MS", "Longueur max d'un énoncé (ms)", "editable" + ), + ConfigField( + "VAD_MIN_SEGMENT_MS", "Longueur min d'un énoncé (ms)", "editable" + ), ), ), ConfigBlock( @@ -87,6 +107,27 @@ class ConfigBlock: title="CORS", fields=(ConfigField("ALLOWED_ORIGINS", "Origines autorisées", "readonly"),), ), + ConfigBlock( + id="extraction", + title="Extraction & vérification", + fields=( + ConfigField( + "CONTEXT_TURNS", + "Fenêtre de contexte (énoncés précédents)", + "editable", + ), + ), + ), + ConfigBlock( + id="persistence", + title="Persistance des sessions", + # Read-only: both take effect at startup (the DB engine is built once), so + # a runtime change wouldn't apply to the live process. + fields=( + ConfigField("PERSIST_SESSIONS", "Persistance activée", "readonly"), + ConfigField("DATABASE_URL", "URL base de données", "readonly"), + ), + ), ConfigBlock( id="logs", title="Logs", diff --git a/app/db/__init__.py b/app/db/__init__.py new file mode 100644 index 0000000..71ac0a3 --- /dev/null +++ b/app/db/__init__.py @@ -0,0 +1,6 @@ +"""Database layer: engine, session factory and the declarative base. + +Cross-cutting infrastructure (see .claude/rules/architecture.md). ORM models live +in ``app/models/``; this package only owns the connection and the ``Base`` they +inherit from. +""" diff --git a/app/db/base.py b/app/db/base.py new file mode 100644 index 0000000..c78bc55 --- /dev/null +++ b/app/db/base.py @@ -0,0 +1,11 @@ +"""Declarative base shared by every ORM model. + +Kept in its own module so ``app.db.session`` (engine/init) and the models can both +import it without a circular dependency. +""" + +from sqlalchemy.orm import DeclarativeBase + + +class Base(DeclarativeBase): + pass diff --git a/app/db/session.py b/app/db/session.py new file mode 100644 index 0000000..2193b7f --- /dev/null +++ b/app/db/session.py @@ -0,0 +1,46 @@ +"""SQLAlchemy engine, session factory and schema bootstrap. + +The app is async but persistence uses the *sync* SQLAlchemy API: writes on the WS +path are offloaded to a thread (``asyncio.to_thread``, like transcription), and the +read routes are plain ``def`` handlers that FastAPI runs in its threadpool. This +keeps the dependency surface minimal (no aiosqlite) and matches the existing +"offload blocking work to a thread" pattern. +""" + +from collections.abc import Iterator + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from app.config import settings +from app.db.base import Base + +# SQLite refuses cross-thread connection reuse by default; we hand connections to +# threadpool workers, so disable that guard. SQLAlchemy still gives each Session +# its own connection from the pool, so this stays safe. +_connect_args = ( + {"check_same_thread": False} if settings.DATABASE_URL.startswith("sqlite") else {} +) + +engine = create_engine(settings.DATABASE_URL, connect_args=_connect_args) + +# expire_on_commit=False lets a committed ORM object still be read (its attributes +# stay populated) after the transaction closes — convenient for short write helpers. +SessionLocal = sessionmaker(bind=engine, autoflush=False, expire_on_commit=False) + + +def init_db() -> None: + """Create any missing tables. Idempotent; called once at startup.""" + # Import the models so they register on Base.metadata before create_all. + from app import models # noqa: F401 + + Base.metadata.create_all(bind=engine) + + +def get_db() -> Iterator[Session]: + """FastAPI dependency yielding a DB session, closed in a finally.""" + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/app/main.py b/app/main.py index 4eb7956..707f5b5 100644 --- a/app/main.py +++ b/app/main.py @@ -9,15 +9,17 @@ from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from app.api.routers import admin, auth, fact_check, health, ws +from app.api.routers import admin, auth, fact_check, health, sessions, ws from app.config import settings from app.core.observability import setup_logging +from app.db.session import init_db from app.services.transcription import preload_model @asynccontextmanager async def lifespan(_app: FastAPI): setup_logging() + init_db() preload_model() yield @@ -36,4 +38,5 @@ async def lifespan(_app: FastAPI): app.include_router(auth.router) app.include_router(fact_check.router) app.include_router(admin.router) +app.include_router(sessions.router) app.include_router(ws.router) diff --git a/app/models/__init__.py b/app/models/__init__.py index f5f2080..6b28905 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -1,6 +1,13 @@ """ORM models (database tables). -Empty for now — the service has no DB yet. ORM models live here when one is -added; they stay distinct from the Pydantic API schemas in ``app/schemas/`` -(see .claude/rules/architecture.md). Convert explicitly across the boundary. +Importing this package registers every model on ``Base.metadata`` (so +``init_db`` can create the tables). Models stay distinct from the Pydantic API +schemas in ``app/schemas/`` (see .claude/rules/architecture.md); convert +explicitly across the boundary. """ + +from app.models.claim import Claim +from app.models.session import Session +from app.models.transcript_segment import TranscriptSegment + +__all__ = ["Claim", "Session", "TranscriptSegment"] diff --git a/app/models/claim.py b/app/models/claim.py new file mode 100644 index 0000000..e526147 --- /dev/null +++ b/app/models/claim.py @@ -0,0 +1,57 @@ +"""ORM model for a persisted, verified claim. + +Mirrors the ``ClaimBase`` contract fields (see ``app/schemas/claim.py``) plus the +DB-side links and ``created_at``. Only the *final* state of a claim is stored — +never the transient ``pending`` placeholder shown live over the WebSocket. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import ( + JSON, + Boolean, + DateTime, + Float, + ForeignKey, + Integer, + String, + Text, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.db.base import Base + +if TYPE_CHECKING: + from app.models.session import Session + from app.models.transcript_segment import TranscriptSegment + + +class Claim(Base): + __tablename__ = "claims" + + id: Mapped[str] = mapped_column(String, primary_key=True) + session_id: Mapped[str] = mapped_column( + ForeignKey("sessions.id", ondelete="CASCADE"), index=True + ) + # A claim outlives the link to its source utterance: SET NULL, not CASCADE. + segment_id: Mapped[str | None] = mapped_column( + ForeignKey("transcript_segments.id", ondelete="SET NULL"), default=None + ) + + # Contract fields (mirror of ClaimBase). + text: Mapped[str] = mapped_column(Text) + status: Mapped[str] = mapped_column(String) # VerificationStatus value + explanation: Mapped[str] = mapped_column(Text, default="") + sources: Mapped[list[str]] = mapped_column(JSON, default=list) + timestamp: Mapped[float] = mapped_column(Float) # ms epoch (contract field) + category: Mapped[str] = mapped_column(String, default="") + confidence: Mapped[int] = mapped_column(Integer, default=0) + counter_claim: Mapped[str] = mapped_column(Text, default="") + web_search_used: Mapped[bool] = mapped_column(Boolean, default=False) + created_at: Mapped[datetime] = mapped_column(DateTime) + + session: Mapped[Session] = relationship(back_populates="claims") + segment: Mapped[TranscriptSegment | None] = relationship(back_populates="claims") diff --git a/app/models/session.py b/app/models/session.py new file mode 100644 index 0000000..5f9253d --- /dev/null +++ b/app/models/session.py @@ -0,0 +1,44 @@ +"""ORM model for a persisted WebSocket session. + +A session row is the parent of its transcript segments and verified claims. It +holds only its identity plus ``chunks_received`` (a raw audio-frame counter that +isn't derivable from anything else and is written once at close). Everything else +— token totals, latencies, claim ratios — is derived at read time by +``app.services.stats`` from the child rows, never denormalised here. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import DateTime, Integer, String +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.db.base import Base + +if TYPE_CHECKING: + from app.models.claim import Claim + from app.models.transcript_segment import TranscriptSegment + + +class Session(Base): + __tablename__ = "sessions" + + id: Mapped[str] = mapped_column(String, primary_key=True) + started_at: Mapped[datetime] = mapped_column(DateTime) + # NULL until the connection closes; an open session has no end yet. + ended_at: Mapped[datetime | None] = mapped_column(DateTime, default=None) + client_host: Mapped[str] = mapped_column(String) + # Raw count of received audio frames — operational, not derivable (we don't + # store the audio). Written once at session close from the in-memory counter. + chunks_received: Mapped[int] = mapped_column(Integer, default=0) + + segments: Mapped[list[TranscriptSegment]] = relationship( + back_populates="session", + cascade="all, delete-orphan", + order_by="TranscriptSegment.seq", + ) + claims: Mapped[list[Claim]] = relationship( + back_populates="session", cascade="all, delete-orphan" + ) diff --git a/app/models/transcript_segment.py b/app/models/transcript_segment.py new file mode 100644 index 0000000..a625d64 --- /dev/null +++ b/app/models/transcript_segment.py @@ -0,0 +1,53 @@ +"""ORM model for one transcribed utterance and the measurements of its pipeline pass. + +A segment maps 1:1 to one pass of the pipeline: one transcription and (if the text +is long enough) one ``extract_and_verify`` call. The per-pass measurements live +here rather than as sums on the session, so session-level totals are plain +aggregates (SUM/AVG) computed at read time. + +The measurement columns are NULL when verification did not run (utterance below +``MIN_WORDS``). A segment that has ``api_calls`` set but no linked claims is a +"reject" — verification ran but found no fact. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import DateTime, Float, ForeignKey, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.db.base import Base + +if TYPE_CHECKING: + from app.models.claim import Claim + from app.models.session import Session + + +class TranscriptSegment(Base): + __tablename__ = "transcript_segments" + + id: Mapped[str] = mapped_column(String, primary_key=True) + session_id: Mapped[str] = mapped_column( + ForeignKey("sessions.id", ondelete="CASCADE"), index=True + ) + seq: Mapped[int] = mapped_column(Integer) # order of appearance in the session + text: Mapped[str] = mapped_column(Text) + detected_language: Mapped[str] = mapped_column(String) + language_probability: Mapped[float] = mapped_column(Float) + created_at: Mapped[datetime] = mapped_column(DateTime) + + # Per-pass measurements. transcribe_ms is always set (transcription always + # runs for a stored segment); the rest are NULL when extraction didn't run. + transcribe_ms: Mapped[float] = mapped_column(Float) + verify_ms: Mapped[float | None] = mapped_column(Float, default=None) + tokens_input: Mapped[int | None] = mapped_column(Integer, default=None) + tokens_output: Mapped[int | None] = mapped_column(Integer, default=None) + tokens_cache_read: Mapped[int | None] = mapped_column(Integer, default=None) + tokens_cache_write: Mapped[int | None] = mapped_column(Integer, default=None) + api_calls: Mapped[int | None] = mapped_column(Integer, default=None) # 1 or 2 + web_search_calls: Mapped[int | None] = mapped_column(Integer, default=None) + + session: Mapped[Session] = relationship(back_populates="segments") + claims: Mapped[list[Claim]] = relationship(back_populates="segment") diff --git a/app/schemas/admin.py b/app/schemas/admin.py index 4123cfc..8b7c3d4 100644 --- a/app/schemas/admin.py +++ b/app/schemas/admin.py @@ -34,7 +34,7 @@ class PromptResponse(BaseModel): FieldKind = Literal["readonly", "editable", "secret_status"] -ValueType = Literal["str", "int", "bool", "list"] +ValueType = Literal["str", "int", "float", "bool", "list"] class ConfigFieldValue(BaseModel): diff --git a/app/schemas/history.py b/app/schemas/history.py new file mode 100644 index 0000000..0f591bb --- /dev/null +++ b/app/schemas/history.py @@ -0,0 +1,111 @@ +"""Read-side schemas for the /sessions history & export routes. + +These serialize the ORM models in ``app/models/`` out to the API. ``from_attributes`` +lets them be built straight from an ORM instance; aggregate stats are computed by +``app.services.stats`` and attached, never stored. +""" + +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class TokenTotals(BaseModel): + input: int = 0 + output: int = 0 + cache_read: int = 0 + cache_write: int = 0 + total: int = 0 + + +class SessionStats(BaseModel): + """Everything derived from a session's segments and claims at read time.""" + + duration_s: float | None = None + transcripts_count: int = 0 + claims_count: int = 0 + claims_by_status: dict[str, int] = {} + dominant_category: str | None = None + avg_confidence: float | None = None + # Segments where verification ran (api_calls set), and those that ran but + # yielded no claim (a "reject" — remove_claim live). + segments_verified: int = 0 + rejects: int = 0 + # web_search reach: segments that searched, total searches, claims that used it. + web_search_segments: int = 0 + web_search_calls_total: int = 0 + claims_with_web_search: int = 0 + tokens: TokenTotals = TokenTotals() + api_calls_total: int = 0 + fallback_count: int = 0 # segments that needed the two-turn fallback + avg_transcribe_ms: float | None = None + avg_verify_ms: float | None = None + # Rough cost estimate using ``pricing_model``'s rates; None if that model + # has no known pricing. The model is the one configured now, which may differ + # from the one used during the session. + estimated_cost_usd: float | None = None + pricing_model: str = "" + + +class ClaimOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + segment_id: str | None = None + text: str + status: str + explanation: str + sources: list[str] + timestamp: float + category: str + confidence: int + counter_claim: str + web_search_used: bool + created_at: datetime + + +class SegmentOut(BaseModel): + model_config = ConfigDict(from_attributes=True) + + id: str + seq: int + text: str + detected_language: str + language_probability: float + created_at: datetime + transcribe_ms: float + verify_ms: float | None = None + tokens_input: int | None = None + tokens_output: int | None = None + tokens_cache_read: int | None = None + tokens_cache_write: int | None = None + api_calls: int | None = None + web_search_calls: int | None = None + + +class SessionSummary(BaseModel): + """Compact row for the session list.""" + + id: str + started_at: datetime + ended_at: datetime | None = None + active: bool + client_host: str + transcripts_count: int + claims_count: int + false_count: int + estimated_cost_usd: float | None = None + + +class SessionDetail(BaseModel): + """Full session: identity, derived stats, transcript and claims.""" + + id: str + started_at: datetime + ended_at: datetime | None = None + active: bool + client_host: str + chunks_received: int + stats: SessionStats + segments: list[SegmentOut] + claims: list[ClaimOut] diff --git a/app/services/audio_endpointer.py b/app/services/audio_endpointer.py new file mode 100644 index 0000000..b9f7535 --- /dev/null +++ b/app/services/audio_endpointer.py @@ -0,0 +1,105 @@ +"""Server-side utterance endpointing for the live /ws PCM stream. + +The client streams raw PCM (16 kHz mono, Int16 frames decoded to float32) with no +fixed chunking. This module buffers the stream and cuts it into utterances on +natural pauses using Silero VAD, replacing the old fixed ~5 s client-side chunks +(which dropped audio at every boundary and split words/sentences). + +The endpointing *policy* (when to flush) is plain array logic, decoupled from the +VAD model via an injectable ``vad_fn`` so it can be unit-tested without ONNX. +""" + +from collections.abc import Callable + +import numpy as np +from faster_whisper.vad import VadOptions, get_speech_timestamps + +# The wire contract fixes this: the client always sends 16 kHz mono PCM. It is a +# constant, not a setting — changing it server-side alone would mis-decode audio. +SAMPLE_RATE = 16000 + +# Padding kept around detected speech when Silero reports spans. Internal detail, +# not exposed as a tunable. +_SPEECH_PAD_MS = 200 + +# A VAD function: PCM float32 @ 16 kHz -> speech spans as ``{"start", "end"}`` dicts +# in sample indices (the shape faster-whisper's get_speech_timestamps returns). +VadFn = Callable[[np.ndarray], list[dict]] + + +def silero_vad(threshold: float) -> VadFn: + """Production VAD: Silero (bundled with faster-whisper, no extra dependency).""" + + options = VadOptions(threshold=threshold, speech_pad_ms=_SPEECH_PAD_MS) + + def run(audio: np.ndarray) -> list[dict]: + return get_speech_timestamps(audio, options, sampling_rate=SAMPLE_RATE) + + return run + + +def _ms_to_samples(ms: int) -> int: + return int(ms / 1000 * SAMPLE_RATE) + + +class Endpointer: + """Buffers a PCM stream and yields one utterance at a time on pause/length. + + Feed frames with :meth:`add`, then call :meth:`pop` repeatedly until it returns + ``None`` to drain every utterance the policy has completed. A single connection + owns one instance and drives it sequentially, so the buffer needs no locking. + """ + + def __init__( + self, + *, + silence_flush_ms: int, + max_segment_ms: int, + min_segment_ms: int, + vad_fn: VadFn, + ) -> None: + self._silence_flush = _ms_to_samples(silence_flush_ms) + self._max_segment = _ms_to_samples(max_segment_ms) + self._min_segment = _ms_to_samples(min_segment_ms) + self._vad_fn = vad_fn + self._buf = np.empty(0, dtype=np.float32) + + def add(self, samples: np.ndarray) -> None: + self._buf = np.concatenate((self._buf, samples)) + + def pop(self) -> np.ndarray | None: + """Return the next ready utterance and trim the buffer, or ``None``. + + An utterance is ready when speech is followed by ``silence_flush_ms`` of + trailing silence, or once the buffer reaches ``max_segment_ms`` (force-flush + a pauseless monologue). Utterances shorter than ``min_segment_ms`` are + dropped. Runs the (CPU-bound) VAD, so call it off the event loop. + """ + if self._buf.size == 0: + return None + + spans = self._vad_fn(self._buf) + if not spans: + # No speech yet: don't let pure silence grow without bound. + if self._buf.size > self._max_segment: + self._buf = np.empty(0, dtype=np.float32) + return None + + first_start = spans[0]["start"] + last_end = spans[-1]["end"] + trailing_silence = self._buf.size - last_end + + ended = trailing_silence >= self._silence_flush + too_long = self._buf.size >= self._max_segment + if not (ended or too_long): + return None + + # Cut at the end of the last detected speech: on a normal flush the rest is + # silence; on a force-flush mid-speech this still avoids slicing past the + # last known speech sample. + segment = self._buf[first_start:last_end] + self._buf = self._buf[last_end:] + + if segment.size < self._min_segment: + return None + return segment diff --git a/app/services/claim_extractor.py b/app/services/claim_extractor.py index a4826c1..a94895e 100644 --- a/app/services/claim_extractor.py +++ b/app/services/claim_extractor.py @@ -1,4 +1,5 @@ import logging +from dataclasses import dataclass, field from typing import Any, cast import anthropic @@ -21,6 +22,21 @@ VALID_STATUSES = {"verified", "false", "uncertain", "unverifiable"} +@dataclass +class ExtractResult: + """Outcome of one extraction pass: the claims plus the call's measurements. + + ``usage`` keys mirror :func:`_usage_dict` (input/output/cache_read/cache_write); + ``api_calls`` is 1, or 2 when the two-turn fallback fired; ``web_search_calls`` + counts the server-side web_search invocations across all turns. + """ + + claims: list[dict[str, Any]] + usage: dict[str, int] = field(default_factory=dict) + api_calls: int = 0 + web_search_calls: int = 0 + + def _log_usage(label: str, usage: Usage) -> None: """Log token usage so cache hits/misses are observable (prefix is small).""" logger.info( @@ -134,6 +150,19 @@ def _log_usage(label: str, usage: Usage) -> None: 'pour trancher de façon fiable, classe-le "uncertain" et explique pourquoi. ' "Si rien à extraire, liste vide.\n" "\n" + "Contexte — le message peut contenir un bloc « Contexte précédent » suivi " + "d'un « Texte à analyser ». N'extrais et ne vérifie QUE les affirmations du " + "« Texte à analyser » ; le contexte sert uniquement à lever les références " + "(« de même », « idem », « donc », pronoms, sujets implicites). Reformule le " + "champ text de chaque claim pour qu'il soit autosuffisant et compréhensible " + "sans le contexte. RÈGLE CRITIQUE : reformule toujours l'affirmation telle que " + "le locuteur la pose, en préservant sa polarité (affirmation ou négation). " + "Ex. positif : contexte « 1 + 1 = 2 », texte « Il en est de même de 1*2 » " + "→ text = « 1 * 2 = 2 » (verified). " + "Ex. négatif : contexte « 1 + 1 = 2 », texte « Ce n'est pas le cas pour 4+4 » " + "→ text = « 4 + 4 != 2 » (verified, car le locuteur a raison). " + "Ne jamais extraire le fait sous-jacent nié comme si le locuteur l'affirmait.\n" + "\n" "Vérification — par défaut, vérifie avec tes connaissances internes SANS " "recherche web. N'utilise web_search QUE si le fait dépend d'informations " "récentes ou changeantes que tu ne peux pas connaître de façon fiable " @@ -149,6 +178,24 @@ def _log_usage(label: str, usage: Usage) -> None: ) +def _build_analysis_prompt(text: str, context: list[str] | None) -> str: + """Build the user message, prepending recent utterances as read-only context. + + Without context the model sees the utterance alone and can't resolve a + back-reference ("Il en est de même de…"). We label the two blocks so the + system prompt can tell the model to extract only from the current text. + """ + if not context: + return f"Analyse ce texte :\n\n{text}" + preceding = "\n".join(context) + return ( + "Contexte précédent (pour comprendre les références ; " + "n'en extrais AUCUN claim) :\n\n" + f"{preceding}\n\n" + f"Texte à analyser :\n\n{text}" + ) + + def _parse_claims(claims_raw: list[Any]) -> list[dict[str, Any]]: return [ { @@ -198,22 +245,42 @@ def _usage_dict(usage: Usage) -> dict[str, int]: } -def _web_search_called(response: Message) -> bool: - return any( +def _add_usage(total: dict[str, int], usage: Usage) -> None: + """Accumulate one response's usage into a running per-call total.""" + for key, value in _usage_dict(usage).items(): + total[key] = total.get(key, 0) + value + + +def _count_web_search(response: Message) -> int: + return sum( block.type == "server_tool_use" and block.name == "web_search" for block in response.content ) +_FORCE_SUBMIT_MSG = ( + "Utilise maintenant submit_claims pour structurer les claims identifiés." +) + + async def extract_and_verify( - text: str, web_search: bool = True -) -> list[dict[str, Any]]: + text: str, context: list[str] | None = None, web_search: bool = True +) -> ExtractResult: + """Extract and verify claims, returning the claims plus the call's measurements. + + The measurements (token usage, number of API calls, web_search count) are what + the persistence layer records on the transcript segment; callers that only need + the claims read ``result.claims``. + """ if len(text.split()) < MIN_WORDS: - return [] + return ExtractResult(claims=[]) messages: list[MessageParam] = [ - {"role": "user", "content": f"Analyse ce texte :\n\n{text}"} + {"role": "user", "content": _build_analysis_prompt(text, context)} ] + usage_total: dict[str, int] = {} + web_search_calls = 0 + api_calls = 0 response = await _client.messages.create( model=settings.ANTHROPIC_MODEL, @@ -223,27 +290,18 @@ async def extract_and_verify( tools=_build_tools(web_search), tool_choice={"type": "auto"}, ) + api_calls += 1 _log_usage("extract", response.usage) + _add_usage(usage_total, response.usage) + web_search_calls += _count_web_search(response) - # Happy path: submit_claims present in first response - # (with or without prior web_search) + # Happy path: submit_claims present in the first response (with or without a + # prior web_search). Otherwise, if Claude searched but stopped before calling + # submit_claims, continue the conversation and force the structured output. claims = _claims_from_response(response) - if claims is not None: - return claims - - # Fallback: Claude did web searches but didn't call submit_claims yet. - # Continue the conversation and force the structured output. - if response.stop_reason == "tool_use": + if claims is None and response.stop_reason == "tool_use": messages.append({"role": "assistant", "content": response.content}) - messages.append( - { - "role": "user", - "content": ( - "Utilise maintenant submit_claims pour structurer " - "les claims identifiés." - ), - } - ) + messages.append({"role": "user", "content": _FORCE_SUBMIT_MSG}) response2 = await _client.messages.create( model=settings.ANTHROPIC_MODEL, max_tokens=1024, @@ -252,80 +310,30 @@ async def extract_and_verify( tools=[WEB_SEARCH_TOOL, CLAIM_TOOL], tool_choice={"type": "tool", "name": "submit_claims"}, ) + api_calls += 1 _log_usage("extract-fallback", response2.usage) + _add_usage(usage_total, response2.usage) + web_search_calls += _count_web_search(response2) claims = _claims_from_response(response2) - if claims is not None: - return claims - - return [] - - -async def debug_extract(text: str, web_search: bool = True) -> dict[str, Any]: - """Like extract_and_verify but also returns token usage and turn count.""" - - def result( - claims: list[dict[str, Any]], - turns: int, - usage: dict[str, int], - web_search_called: bool, - ) -> dict[str, Any]: - return { - "claims": claims, - "turns": turns, - "usage": usage, - "model": settings.ANTHROPIC_MODEL, - "web_search_enabled": web_search, - "web_search_called": web_search_called, - } - - if len(text.split()) < MIN_WORDS: - return result([], 0, {}, False) - messages: list[MessageParam] = [ - {"role": "user", "content": f"Analyse ce texte :\n\n{text}"} - ] - - response = await _client.messages.create( - model=settings.ANTHROPIC_MODEL, - max_tokens=1024, - system=SYSTEM_PROMPT, - messages=messages, - tools=_build_tools(web_search), - tool_choice={"type": "auto"}, + return ExtractResult( + claims=claims or [], + usage=usage_total, + api_calls=api_calls, + web_search_calls=web_search_calls, ) - _log_usage("debug-extract", response.usage) - - web_search_called = _web_search_called(response) - total_usage = _usage_dict(response.usage) - claims = _claims_from_response(response) - if claims is not None: - return result(claims, 1, total_usage, web_search_called) - if response.stop_reason == "tool_use": - messages.append({"role": "assistant", "content": response.content}) - messages.append( - { - "role": "user", - "content": ( - "Utilise maintenant submit_claims pour structurer " - "les claims identifiés." - ), - } - ) - response2 = await _client.messages.create( - model=settings.ANTHROPIC_MODEL, - max_tokens=1024, - system=SYSTEM_PROMPT, - messages=messages, - tools=[WEB_SEARCH_TOOL, CLAIM_TOOL], - tool_choice={"type": "tool", "name": "submit_claims"}, - ) - _log_usage("debug-extract-fallback", response2.usage) - for key, value in _usage_dict(response2.usage).items(): - total_usage[key] += value - claims = _claims_from_response(response2) - if claims is not None: - return result(claims, 2, total_usage, web_search_called) - - return result([], 1, total_usage, web_search_called) +async def debug_extract( + text: str, context: list[str] | None = None, web_search: bool = True +) -> dict[str, Any]: + """Shape ``extract_and_verify`` for the admin model-test panel.""" + result = await extract_and_verify(text, context=context, web_search=web_search) + return { + "claims": result.claims, + "turns": result.api_calls, + "usage": result.usage, + "model": settings.ANTHROPIC_MODEL, + "web_search_enabled": web_search, + "web_search_called": result.web_search_calls > 0, + } diff --git a/app/services/export.py b/app/services/export.py new file mode 100644 index 0000000..990a279 --- /dev/null +++ b/app/services/export.py @@ -0,0 +1,85 @@ +"""Render a session to a shareable Markdown document. + +Pure formatting over the read schemas (no DB, no I/O), so it's trivially testable +and reusable by both the JSON and Markdown export paths. The JSON export is just +the ``SessionDetail`` serialized as-is; this module covers the human-readable form. +""" + +from app.schemas.history import SessionDetail + + +def _format_dt(value: object) -> str: + return str(value) if value is not None else "—" + + +def session_to_markdown(detail: SessionDetail) -> str: + s = detail.stats + lines: list[str] = [] + + lines.append(f"# LiveFactChecker — session {detail.id}") + lines.append("") + lines.append(f"- **Started**: {_format_dt(detail.started_at)}") + lines.append( + f"- **Ended**: {'active' if detail.active else _format_dt(detail.ended_at)}" + ) + if s.duration_s is not None: + lines.append(f"- **Duration**: {s.duration_s} s") + lines.append(f"- **Client**: {detail.client_host}") + lines.append("") + + lines.append("## Statistics") + lines.append("") + lines.append(f"- Transcripts: {s.transcripts_count}") + by_status = ", ".join(f"{k} {v}" for k, v in sorted(s.claims_by_status.items())) + claims_line = f"- Claims: {s.claims_count}" + if by_status: + claims_line += f" ({by_status})" + lines.append(claims_line) + lines.append(f"- Rejects (verified, no claim): {s.rejects}") + if s.dominant_category: + lines.append(f"- Dominant category: {s.dominant_category}") + if s.avg_confidence is not None: + lines.append(f"- Average confidence: {s.avg_confidence}/10") + lines.append( + f"- Web search: {s.web_search_calls_total} calls " + f"over {s.web_search_segments} segments" + ) + lines.append( + f"- Tokens: {s.tokens.total} total " + f"(in {s.tokens.input}, out {s.tokens.output}, " + f"cache_read {s.tokens.cache_read}, cache_write {s.tokens.cache_write})" + ) + if s.estimated_cost_usd is not None: + lines.append(f"- Estimated cost: ${s.estimated_cost_usd} ({s.pricing_model})") + lines.append(f"- API calls: {s.api_calls_total} (fallbacks: {s.fallback_count})") + if s.avg_transcribe_ms is not None: + lines.append(f"- Avg transcription latency: {s.avg_transcribe_ms} ms") + if s.avg_verify_ms is not None: + lines.append(f"- Avg verification latency: {s.avg_verify_ms} ms") + lines.append("") + + lines.append("## Claims") + lines.append("") + if not detail.claims: + lines.append("_No claims._") + for claim in detail.claims: + lines.append(f"### [{claim.status}] {claim.text}") + if claim.explanation: + lines.append(f"- {claim.explanation}") + if claim.counter_claim: + lines.append(f"- **Correction**: {claim.counter_claim}") + meta = f"confidence {claim.confidence}/10" + if claim.category: + meta += f" · {claim.category}" + lines.append(f"- _{meta}_") + for source in claim.sources: + lines.append(f"- Source: {source}") + lines.append("") + + lines.append("## Transcript") + lines.append("") + for segment in detail.segments: + lines.append(f"{segment.seq + 1}. ({segment.detected_language}) {segment.text}") + lines.append("") + + return "\n".join(lines) diff --git a/app/services/session.py b/app/services/session.py index 2b792c9..db5f14f 100644 --- a/app/services/session.py +++ b/app/services/session.py @@ -10,9 +10,12 @@ import logging import time import uuid +from collections import deque +import numpy as np from fastapi import WebSocket from pydantic import ValidationError +from sqlalchemy.exc import SQLAlchemyError from starlette.websockets import WebSocketDisconnect from app.config import settings @@ -23,15 +26,59 @@ VerificationLevel, VerificationStatus, ) +from app.services import session_store +from app.services.audio_endpointer import Endpointer, silero_vad from app.services.claim_extractor import MIN_WORDS, extract_and_verify -from app.services.transcription import transcribe_chunk +from app.services.transcription import transcribe_samples logger = logging.getLogger(__name__) +# Live registry of currently-open WebSocket connections, keyed by session id. +# This is runtime state, NOT a cache of the DB: each value holds live Python +# objects (the set of in-flight asyncio.Tasks, the rolling-context deque) and +# sub-second counters that are never persisted. Its sole reader is +# get_sessions_status() -> GET /admin/ws/status (the live monitor page). +# +# Limitation — single process only: with multiple uvicorn workers each worker +# sees only its own connections, and the registry doesn't survive a restart. +# It cannot be replaced by DB queries: the DB tracks *history* (rows created +# lazily on first transcript, closed on disconnect), not who is connected now. _active_sessions: dict[str, dict] = {} _total_sessions = 0 +async def _persist(fn, *args, **kwargs) -> None: + """Run a session_store write off the event loop, best-effort. + + Persistence must never drop a live session: a DB error is logged and + swallowed, and the whole call is skipped when PERSIST_SESSIONS is off. + """ + if not settings.PERSIST_SESSIONS: + return + try: + await asyncio.to_thread(fn, *args, **kwargs) + except SQLAlchemyError as e: + logger.error("Persistence error in %s: %s", fn.__name__, e) + + +async def _ensure_persisted(session_info: dict) -> None: + """Create the session row lazily, on the first transcript. + + A connection that opens and closes without ever producing a transcript leaves + no row behind — no empty session clutters the history. Called from the + sequential utterance loop, so the create-once guard needs no locking. + """ + if session_info["persisted"]: + return + session_info["persisted"] = True + await _persist( + session_store.create_session, + session_info["id"], + session_info["client"], + session_info["started_at"], + ) + + def get_sessions_status() -> dict: now = time.time() sessions = [ @@ -67,8 +114,19 @@ def _make_claim(result: dict, claim_id: str, timestamp: int) -> Claim: ) -async def _process_claims(ws: WebSocket, transcript: str, web_search: bool): - """Show a pending claim, then replace/remove it with the verified results.""" +async def _process_claims( + ws: WebSocket, + transcript: str, + context: list[str], + web_search: bool, + session_id: str, + segment_id: str, +): + """Show a pending claim, then replace/remove it with the verified results. + + Also records the verification measurements on the segment and persists each + final claim (never the transient pending placeholder). + """ try: pending_id = str(uuid.uuid4()) pending_ts = int(time.time() * 1000) @@ -80,21 +138,34 @@ async def _process_claims(ws: WebSocket, transcript: str, web_search: bool): ) await ws.send_json({"type": "claim", "claim": pending.model_dump()}) - results = await extract_and_verify(transcript, web_search=web_search) - if not results: + started = time.perf_counter() + result = await extract_and_verify( + transcript, context=context, web_search=web_search + ) + verify_ms = (time.perf_counter() - started) * 1000.0 + await _persist( + session_store.set_segment_metrics, + segment_id, + verify_ms=verify_ms, + usage=result.usage, + api_calls=result.api_calls, + web_search_calls=result.web_search_calls, + ) + + if not result.claims: await ws.send_json({"type": "remove_claim", "id": pending_id}) return - first, *rest = results - await ws.send_json( - { - "type": "claim", - "claim": _make_claim(first, pending_id, pending_ts).model_dump(), - } - ) - for result in rest: - claim = _make_claim(result, str(uuid.uuid4()), int(time.time() * 1000)) - await ws.send_json({"type": "claim", "claim": claim.model_dump()}) + first, *rest = result.claims + first_claim = _make_claim(first, pending_id, pending_ts).model_dump() + await ws.send_json({"type": "claim", "claim": first_claim}) + await _persist(session_store.add_claim, first_claim, session_id, segment_id) + for extra in rest: + claim = _make_claim( + extra, str(uuid.uuid4()), int(time.time() * 1000) + ).model_dump() + await ws.send_json({"type": "claim", "claim": claim}) + await _persist(session_store.add_claim, claim, session_id, segment_id) except Exception as e: logger.error("Claim processing error: %s", e) @@ -103,8 +174,10 @@ async def _process_claims(ws: WebSocket, transcript: str, web_search: bool): def _spawn_claims( ws: WebSocket, transcript: str, + context: list[str], background_tasks: set[asyncio.Task], session_info: dict, + segment_id: str, ): """Fire claim extraction for a transcribed chunk, tracking the task.""" if len(transcript.split()) < MIN_WORDS: @@ -113,11 +186,103 @@ def _spawn_claims( # THOROUGH offers the web_search tool; FAST keeps verification to internal # knowledge for a single, faster API call. web_search = session_info["verification_level"] == VerificationLevel.THOROUGH - task = asyncio.create_task(_process_claims(ws, transcript, web_search)) + task = asyncio.create_task( + _process_claims( + ws, transcript, context, web_search, session_info["id"], segment_id + ) + ) background_tasks.add(task) task.add_done_callback(background_tasks.discard) +async def _emit_utterance( + ws: WebSocket, + segment: np.ndarray, + session_info: dict, + background_tasks: set[asyncio.Task], + loop: asyncio.AbstractEventLoop, +) -> None: + """Transcribe one endpointed utterance and emit its transcript + claims.""" + # Always transcribe in auto-detect: forcing a language would make Whisper + # translate a mismatched utterance instead of transcribing it. + started = time.perf_counter() + try: + transcript, detected_lang, detected_prob = await loop.run_in_executor( + None, transcribe_samples, segment + ) + except Exception as e: + logger.error("Transcription error: %s", e) + return + transcribe_ms = (time.perf_counter() - started) * 1000.0 + + if not transcript: + return + + # The chosen language is a filter, not a forced transcription target: drop an + # utterance whose detected language doesn't match (None = accept all). + language = session_info["language"] + if language is not None and detected_lang != language: + logger.info( + "Skipping utterance: detected %s, session filter is %s", + detected_lang, + language, + ) + return + + session_info["transcripts"] += 1 + session_info["last_transcript"] = transcript[:120] + session_info["last_activity"] = time.time() + + logger.info(transcript) + await ws.send_json( + { + "type": "transcript", + "text": transcript, + "language": detected_lang, + "language_probability": detected_prob, + } + ) + + # Create the session row on the first transcript, then persist the segment + # before spawning verification, so the background task's metrics update and + # any linked claims have a parent row to reference. + await _ensure_persisted(session_info) + segment_id = str(uuid.uuid4()) + seq = session_info["seq"] + session_info["seq"] += 1 + await _persist( + session_store.add_segment, + segment_id=segment_id, + session_id=session_info["id"], + seq=seq, + text=transcript, + detected_language=detected_lang, + language_probability=detected_prob, + transcribe_ms=transcribe_ms, + ) + + # Snapshot the preceding utterances *before* adding the current one, so a + # back-reference ("Il en est de même de…") can be resolved against them. The + # list() copy is what the background task reads, immune to later mutations. + context = list(session_info["context"]) + _spawn_claims(ws, transcript, context, background_tasks, session_info, segment_id) + session_info["context"].append(transcript) + + +def _make_endpointer() -> Endpointer: + """Build a per-session endpointer from the current VAD settings. + + Reads settings at session start, so a runtime ``/admin/config`` change applies + to new sessions (an existing live session keeps the config it opened with). + """ + return Endpointer( + silence_flush_ms=settings.VAD_SILENCE_FLUSH_MS, + max_segment_ms=settings.VAD_MAX_SEGMENT_MS, + min_segment_ms=settings.VAD_MIN_SEGMENT_MS, + vad_fn=silero_vad(settings.VAD_THRESHOLD), + ) + + def parse_config(raw: str) -> ConfigMessage: """Parse and validate a client config frame. @@ -140,17 +305,19 @@ def parse_config(raw: str) -> ConfigMessage: async def run_session(ws: WebSocket): - """Drive one WebSocket connection: receive chunks, transcribe, fact-check. + """Drive one WebSocket connection: receive PCM, endpoint, transcribe, fact-check. - The client records ~5 s slices with MediaRecorder and sends each as a - complete WebM/Opus blob (binary frame). We transcribe the blob in one pass - and fire claim extraction on the resulting text. Sentences may be split - across chunk boundaries — this is the simple baseline. + The client streams raw PCM continuously (16 kHz mono Int16 frames, binary). We + buffer it and cut utterances on natural pauses with a VAD endpointer instead of + fixed client-side chunks — so words/sentences are no longer split every 5 s and + no audio is dropped at chunk boundaries. Each completed utterance is transcribed + and fed to claim extraction. """ global _total_sessions await ws.accept() loop = asyncio.get_event_loop() background_tasks: set[asyncio.Task] = set() + endpointer = _make_endpointer() session_id = str(uuid.uuid4()) _total_sessions += 1 @@ -164,6 +331,10 @@ async def run_session(ws: WebSocket): "claims_spawned": 0, "_tasks": background_tasks, "last_transcript": "", + # Rolling window of recent transcripts, handed to claim extraction as + # read-only context so an utterance that references the previous one can + # be resolved instead of dropped as unverifiable. + "context": deque(maxlen=settings.CONTEXT_TURNS), # Transcription language: None = auto-detect (the default until the # client sends a config frame), or a forced ISO code. "language": None, @@ -171,6 +342,13 @@ async def run_session(ws: WebSocket): # until the client says otherwise, matching the prior default. "verification_level": VerificationLevel.THOROUGH, "last_activity": time.time(), + # Monotonic sequence number for persisted transcript segments. + "seq": 0, + # The session row is created lazily on the first transcript (see + # _ensure_persisted), so an empty connection leaves no row. started_at is + # captured now so the row reflects the real connect time, not first speech. + "started_at": session_store.utcnow(), + "persisted": False, } _active_sessions[session_id] = session_info logger.info("WS session %s opened from %s", session_id[:8], client_host) @@ -203,57 +381,35 @@ async def run_session(ws: WebSocket): if not audio: continue - # Drop an oversized blob instead of killing the session: one bad - # chunk shouldn't end a legitimate live stream. + # Drop an oversized frame instead of killing the session: one bad + # frame shouldn't end a legitimate live stream. if len(audio) > settings.MAX_AUDIO_BYTES: logger.warning( - "Dropping oversized audio chunk: %d bytes (max %d)", + "Dropping oversized audio frame: %d bytes (max %d)", len(audio), settings.MAX_AUDIO_BYTES, ) continue - session_info["chunks_received"] += 1 - session_info["last_activity"] = time.time() - - # Always transcribe in auto-detect: forcing a language would make - # Whisper translate a mismatched chunk instead of transcribing it. - try: - transcript, detected_lang, detected_prob = await loop.run_in_executor( - None, transcribe_chunk, audio - ) - except Exception as e: - logger.error("Transcription error: %s", e) - continue - - if not transcript: + # Each binary frame is raw Int16 LE PCM @ 16 kHz mono. An odd byte + # count means a torn frame — skip it rather than mis-decode/crash. + if len(audio) % 2 != 0: + logger.warning("Dropping PCM frame with odd byte count: %d", len(audio)) continue - # The chosen language is a filter, not a forced transcription target: - # drop a chunk whose detected language doesn't match (None = accept all). - language = session_info["language"] - if language is not None and detected_lang != language: - logger.info( - "Skipping chunk: detected %s, session filter is %s", - detected_lang, - language, - ) - continue - - session_info["transcripts"] += 1 - session_info["last_transcript"] = transcript[:120] + session_info["chunks_received"] += 1 session_info["last_activity"] = time.time() - logger.info(transcript) - await ws.send_json( - { - "type": "transcript", - "text": transcript, - "language": detected_lang, - "language_probability": detected_prob, - } - ) - _spawn_claims(ws, transcript, background_tasks, session_info) + samples = np.frombuffer(audio, dtype=" datetime: + """Naive UTC timestamp, so all stored datetimes are comparable.""" + return datetime.now(UTC).replace(tzinfo=None) + + +def create_session(session_id: str, client_host: str, started_at: datetime) -> None: + with SessionLocal() as db: + db.add( + Session( + id=session_id, + started_at=started_at, + client_host=client_host, + ) + ) + db.commit() + + +def end_session(session_id: str, ended_at: datetime, chunks_received: int) -> None: + with SessionLocal() as db: + session = db.get(Session, session_id) + if session is None: + return + session.ended_at = ended_at + session.chunks_received = chunks_received + db.commit() + + +def add_segment( + *, + segment_id: str, + session_id: str, + seq: int, + text: str, + detected_language: str, + language_probability: float, + transcribe_ms: float, +) -> None: + with SessionLocal() as db: + db.add( + TranscriptSegment( + id=segment_id, + session_id=session_id, + seq=seq, + text=text, + detected_language=detected_language, + language_probability=language_probability, + created_at=utcnow(), + transcribe_ms=transcribe_ms, + ) + ) + db.commit() + + +def set_segment_metrics( + segment_id: str, + *, + verify_ms: float, + usage: dict[str, int], + api_calls: int, + web_search_calls: int, +) -> None: + """Fill in the verification measurements once the background task completes.""" + with SessionLocal() as db: + segment = db.get(TranscriptSegment, segment_id) + if segment is None: + return + segment.verify_ms = verify_ms + segment.tokens_input = usage.get("input_tokens", 0) + segment.tokens_output = usage.get("output_tokens", 0) + segment.tokens_cache_read = usage.get("cache_read", 0) + segment.tokens_cache_write = usage.get("cache_write", 0) + segment.api_calls = api_calls + segment.web_search_calls = web_search_calls + db.commit() + + +def add_claim(claim: dict, session_id: str, segment_id: str | None) -> None: + """Persist a verified claim from its WS ``model_dump`` dict (final state only).""" + with SessionLocal() as db: + db.add( + Claim( + id=claim["id"], + session_id=session_id, + segment_id=segment_id, + text=claim["text"], + status=claim["status"], + explanation=claim["explanation"], + sources=claim["sources"], + timestamp=claim["timestamp"], + category=claim["category"], + confidence=claim["confidence"], + counter_claim=claim["counter_claim"], + web_search_used=claim["web_search_used"], + created_at=utcnow(), + ) + ) + db.commit() diff --git a/app/services/stats.py b/app/services/stats.py new file mode 100644 index 0000000..6aa8207 --- /dev/null +++ b/app/services/stats.py @@ -0,0 +1,93 @@ +"""Derived per-session statistics, computed at read time from the child rows. + +Nothing here is stored: a session row holds only identity + ``chunks_received``; +everything else (token totals, latencies, claim ratios, cost) is aggregated from +its segments and claims when a ``/sessions`` route is served. +""" + +from collections import Counter + +from app.models import Session +from app.schemas.history import SessionStats, TokenTotals + +# USD per million tokens. VERIFY against current Anthropic pricing before relying +# on the cost estimate — these are indicative and easy to get stale. A model +# absent from this map yields estimated_cost_usd = None rather than a wrong number. +PRICING: dict[str, dict[str, float]] = { + "claude-haiku-4-5": { + "input": 1.0, + "output": 5.0, + "cache_write": 1.25, + "cache_read": 0.1, + }, + "claude-haiku-4-5-20251001": { + "input": 1.0, + "output": 5.0, + "cache_write": 1.25, + "cache_read": 0.1, + }, +} + + +def _mean(values: list[float]) -> float | None: + return round(sum(values) / len(values), 2) if values else None + + +def _estimate_cost(model: str, tokens: TokenTotals) -> float | None: + rates = PRICING.get(model) + if rates is None: + return None + cost = ( + tokens.input * rates["input"] + + tokens.output * rates["output"] + + tokens.cache_write * rates["cache_write"] + + tokens.cache_read * rates["cache_read"] + ) / 1_000_000 + return round(cost, 6) + + +def compute_stats(session: Session, model: str) -> SessionStats: + segments = session.segments + claims = session.claims + + duration_s: float | None = None + if session.ended_at is not None: + duration_s = round((session.ended_at - session.started_at).total_seconds(), 2) + + tokens = TokenTotals( + input=sum(s.tokens_input or 0 for s in segments), + output=sum(s.tokens_output or 0 for s in segments), + cache_read=sum(s.tokens_cache_read or 0 for s in segments), + cache_write=sum(s.tokens_cache_write or 0 for s in segments), + ) + tokens.total = tokens.input + tokens.output + tokens.cache_read + tokens.cache_write + + verified_segments = [s for s in segments if s.api_calls is not None] + segments_with_claims = {c.segment_id for c in claims if c.segment_id is not None} + rejects = sum(1 for s in verified_segments if s.id not in segments_with_claims) + + categories = [c.category for c in claims if c.category] + confidences = [c.confidence for c in claims] + + return SessionStats( + duration_s=duration_s, + transcripts_count=len(segments), + claims_count=len(claims), + claims_by_status=dict(Counter(c.status for c in claims)), + dominant_category=Counter(categories).most_common(1)[0][0] + if categories + else None, + avg_confidence=_mean([float(c) for c in confidences]), + segments_verified=len(verified_segments), + rejects=rejects, + web_search_segments=sum(1 for s in segments if (s.web_search_calls or 0) > 0), + web_search_calls_total=sum(s.web_search_calls or 0 for s in segments), + claims_with_web_search=sum(1 for c in claims if c.web_search_used), + tokens=tokens, + api_calls_total=sum(s.api_calls or 0 for s in segments), + fallback_count=sum(1 for s in segments if (s.api_calls or 0) >= 2), + avg_transcribe_ms=_mean([s.transcribe_ms for s in segments]), + avg_verify_ms=_mean([s.verify_ms for s in segments if s.verify_ms is not None]), + estimated_cost_usd=_estimate_cost(model, tokens), + pricing_model=model, + ) diff --git a/app/services/transcription.py b/app/services/transcription.py index ffe6916..9e7621a 100644 --- a/app/services/transcription.py +++ b/app/services/transcription.py @@ -6,6 +6,7 @@ import time from pathlib import Path +import numpy as np from faster_whisper import WhisperModel from app.config import settings @@ -67,7 +68,11 @@ def is_model_loaded() -> bool: def transcribe_with_detail(audio: bytes) -> dict: - """Like transcribe_chunk but returns segments, language info and timing.""" + """Transcribe encoded audio bytes (ffmpeg-decoded), returning segments + timing. + + Used by the /admin upload probe, which accepts arbitrary files — so unlike the + live ``transcribe_samples`` path it still decodes via ffmpeg. + """ model = _get_model() source = io.BytesIO(audio) t0 = time.time() @@ -101,11 +106,12 @@ def transcribe_with_detail(audio: bytes) -> dict: return {"error": str(e), "text": "", "segments": []} -def transcribe_chunk(audio: bytes) -> tuple[str, str, float]: - """Transcribe a self-contained audio chunk into plain text. +def transcribe_samples(audio: np.ndarray) -> tuple[str, str, float]: + """Transcribe an endpointed utterance (float32 PCM @ 16 kHz) into plain text. - Accepts raw encoded audio bytes (e.g. a complete WebM/Opus blob, - decoded via ffmpeg). Synchronous and CPU-bound — call it from a thread pool. + Accepts the decoded waveform directly — the live /ws path streams raw PCM and + cuts utterances server-side (see services.audio_endpointer), so no ffmpeg decode + is needed here. Synchronous and CPU-bound — call it from a thread pool. Always auto-detects the language: forcing a non-matching language makes Whisper translate/hallucinate into that language rather than transcribe @@ -117,7 +123,7 @@ def transcribe_chunk(audio: bytes) -> tuple[str, str, float]: """ try: segments, info = _get_model().transcribe( - io.BytesIO(audio), + audio, language=None, vad_filter=True, ) diff --git a/pyproject.toml b/pyproject.toml index b594c4d..5871303 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -25,6 +25,7 @@ dependencies = [ "numpy>=2.4.6", "pyjwt>=2.8.0", "psutil>=7.2.2", + "sqlalchemy>=2.0.0", ] [project.optional-dependencies] diff --git a/tests/test_audio_endpointer.py b/tests/test_audio_endpointer.py new file mode 100644 index 0000000..a214ba5 --- /dev/null +++ b/tests/test_audio_endpointer.py @@ -0,0 +1,106 @@ +"""Tests for the VAD endpointing policy, with a fake VAD (no Silero/ONNX). + +The flush logic is plain array arithmetic, so we inject a deterministic ``vad_fn`` +that marks every contiguous run of non-zero samples as speech. ``1.0`` = speech, +``0.0`` = silence. Durations are kept tiny so the arrays stay small. +""" + +import numpy as np + +from app.services.audio_endpointer import SAMPLE_RATE, Endpointer + + +def _nonzero_vad(audio: np.ndarray) -> list[dict]: + """Mark each contiguous run of non-zero samples as a speech span.""" + spans: list[dict] = [] + start: int | None = None + for i, v in enumerate(audio): + if v != 0 and start is None: + start = i + elif v == 0 and start is not None: + spans.append({"start": start, "end": i}) + start = None + if start is not None: + spans.append({"start": start, "end": len(audio)}) + return spans + + +def _ms(ms: int) -> int: + return int(ms / 1000 * SAMPLE_RATE) + + +def _speech(ms: int) -> np.ndarray: + return np.ones(_ms(ms), dtype=np.float32) + + +def _silence(ms: int) -> np.ndarray: + return np.zeros(_ms(ms), dtype=np.float32) + + +def _endpointer( + *, + silence_flush_ms: int = 100, + max_segment_ms: int = 10_000, + min_segment_ms: int = 20, +) -> Endpointer: + return Endpointer( + silence_flush_ms=silence_flush_ms, + max_segment_ms=max_segment_ms, + min_segment_ms=min_segment_ms, + vad_fn=_nonzero_vad, + ) + + +def test_flushes_after_trailing_silence() -> None: + ep = _endpointer(silence_flush_ms=100) + ep.add(np.concatenate((_speech(500), _silence(150)))) + + segment = ep.pop() + assert segment is not None + assert segment.size == _ms(500) # only the speech, trailing silence trimmed + assert ep.pop() is None # nothing left but silence + + +def test_no_flush_until_silence_long_enough() -> None: + ep = _endpointer(silence_flush_ms=100) + ep.add(np.concatenate((_speech(500), _silence(50)))) # 50 ms < 100 ms + assert ep.pop() is None + + +def test_force_flush_on_max_length_without_pause() -> None: + ep = _endpointer(silence_flush_ms=100, max_segment_ms=200) + ep.add(_speech(250)) # pure speech, no pause, over the 200 ms cap + + segment = ep.pop() + assert segment is not None + assert segment.size == _ms(250) + + +def test_too_short_utterance_is_dropped() -> None: + ep = _endpointer(silence_flush_ms=100, min_segment_ms=500) + ep.add(np.concatenate((_speech(100), _silence(150)))) # speech < 500 ms + + assert ep.pop() is None # dropped + assert ep.pop() is None # and the buffer was trimmed past it + + +def test_merges_speech_across_short_gap() -> None: + ep = _endpointer(silence_flush_ms=100) + ep.add( + np.concatenate( + (_speech(200), _silence(50), _speech(200), _silence(150)) + ) # 50 ms gap stays inside one utterance; 150 ms gap ends it + ) + + segment = ep.pop() + assert segment is not None + # From first speech start to last speech end, inner gap included. + assert segment.size == _ms(200) + _ms(50) + _ms(200) + + +def test_pure_silence_buffer_is_bounded() -> None: + ep = _endpointer(max_segment_ms=200) + ep.add(_silence(300)) # over the cap, no speech + + assert ep.pop() is None + assert ep._buf.size == 0 # silence was discarded, not kept growing diff --git a/tests/test_claim_extractor.py b/tests/test_claim_extractor.py index 3120e8d..08fba02 100644 --- a/tests/test_claim_extractor.py +++ b/tests/test_claim_extractor.py @@ -5,7 +5,7 @@ ``extract_and_verify`` is left for a later integration test with a mocked client. """ -from app.services.claim_extractor import _parse_claims +from app.services.claim_extractor import _build_analysis_prompt, _parse_claims def test_keeps_valid_claim_and_normalises_fields() -> None: @@ -71,3 +71,26 @@ def test_web_search_used_is_strict_boolean() -> None: [{"text": "a", "status": "verified", "web_search_used": "yes"}] ) assert truthy["web_search_used"] is False + + +def test_prompt_without_context_is_the_bare_text() -> None: + prompt = _build_analysis_prompt("Il en est de même de 1*2", None) + assert "Il en est de même de 1*2" in prompt + assert "Contexte précédent" not in prompt + + +def test_prompt_with_context_includes_preceding_utterances() -> None: + # Regression: without the preceding "1 + 1 = 2", the model can't resolve the + # back-reference and drops the second utterance as unverifiable. + prompt = _build_analysis_prompt( + "Il en est de même de 1*2", ["1 + 1 = 2.", "Voici un exemple."] + ) + assert "1 + 1 = 2." in prompt + assert "Voici un exemple." in prompt + assert "Il en est de même de 1*2" in prompt + # The guard that stops the model from re-extracting claims from the context. + assert "AUCUN claim" in prompt + + +def test_empty_context_list_is_treated_as_no_context() -> None: + assert _build_analysis_prompt("x", []) == _build_analysis_prompt("x", None) diff --git a/tests/test_export.py b/tests/test_export.py new file mode 100644 index 0000000..b8d6c57 --- /dev/null +++ b/tests/test_export.py @@ -0,0 +1,89 @@ +"""Unit tests for the Markdown session export (pure formatting).""" + +from datetime import datetime + +from app.schemas.history import ( + ClaimOut, + SegmentOut, + SessionDetail, + SessionStats, + TokenTotals, +) +from app.services.export import session_to_markdown + + +def _detail() -> SessionDetail: + return SessionDetail( + id="abc", + started_at=datetime(2026, 1, 1, 10, 0, 0), + ended_at=datetime(2026, 1, 1, 10, 0, 30), + active=False, + client_host="127.0.0.1", + chunks_received=10, + stats=SessionStats( + duration_s=30.0, + transcripts_count=2, + claims_count=1, + claims_by_status={"false": 1}, + dominant_category="science", + avg_confidence=6.0, + tokens=TokenTotals(input=100, output=50, total=150), + estimated_cost_usd=0.0001, + pricing_model="claude-haiku-4-5", + ), + segments=[ + SegmentOut( + id="seg1", + seq=0, + text="La Terre est plate.", + detected_language="fr", + language_probability=0.99, + created_at=datetime(2026, 1, 1, 10, 0, 1), + transcribe_ms=12.0, + ) + ], + claims=[ + ClaimOut( + id="c1", + segment_id="seg1", + text="La Terre est plate.", + status="false", + explanation="La Terre est un géoïde.", + sources=["https://example.com"], + timestamp=0.0, + category="science", + confidence=6, + counter_claim="La Terre est sphérique.", + web_search_used=True, + created_at=datetime(2026, 1, 1, 10, 0, 2), + ) + ], + ) + + +def test_markdown_includes_header_and_stats() -> None: + md = session_to_markdown(_detail()) + assert "# LiveFactChecker — session abc" in md + assert "Duration**: 30.0 s" in md + assert "Claims: 1 (false 1)" in md + assert "Estimated cost: $0.0001 (claude-haiku-4-5)" in md + + +def test_markdown_includes_claim_and_correction() -> None: + md = session_to_markdown(_detail()) + assert "### [false] La Terre est plate." in md + assert "**Correction**: La Terre est sphérique." in md + assert "Source: https://example.com" in md + + +def test_markdown_includes_transcript() -> None: + md = session_to_markdown(_detail()) + assert "## Transcript" in md + assert "1. (fr) La Terre est plate." in md + + +def test_markdown_handles_empty_claims() -> None: + detail = _detail() + detail.claims = [] + md = session_to_markdown(detail) + assert "_No claims._" in md diff --git a/tests/test_extract_usage.py b/tests/test_extract_usage.py new file mode 100644 index 0000000..2072a7b --- /dev/null +++ b/tests/test_extract_usage.py @@ -0,0 +1,114 @@ +"""Tests for the usage/metrics plumbing in extract_and_verify (mocked client). + +These lock the behaviour the persistence layer depends on: accumulated token +usage, the API-call count (including the two-turn fallback) and the web_search +count — without any real Anthropic call. +""" + +import asyncio + +from app.services import claim_extractor +from app.services.claim_extractor import extract_and_verify + + +class _FakeUsage: + def __init__(self, i: int = 0, o: int = 0, cw: int = 0, cr: int = 0) -> None: + self.input_tokens = i + self.output_tokens = o + self.cache_creation_input_tokens = cw + self.cache_read_input_tokens = cr + + +class _Block: + def __init__(self, type: str, name: str = "", input: dict | None = None) -> None: + self.type = type + self.name = name + self.input = input or {} + + +class _FakeResponse: + def __init__(self, content: list, usage: _FakeUsage, stop_reason: str) -> None: + self.content = content + self.usage = usage + self.stop_reason = stop_reason + + +_VALID_CLAIM = { + "text": "La Tour Eiffel mesure 330 m.", + "status": "verified", + "explanation": "ok", + "sources": [], + "category": "histoire", + "confidence": 9, + "counter_claim": "", + "web_search_used": False, +} + + +def _patch_responses(monkeypatch, responses: list[_FakeResponse]) -> None: + queue = list(responses) + + async def fake_create(*_args, **_kwargs) -> _FakeResponse: + return queue.pop(0) + + monkeypatch.setattr(claim_extractor._client.messages, "create", fake_create) + + +def test_happy_path_records_usage_and_one_call(monkeypatch) -> None: + response = _FakeResponse( + content=[ + _Block("server_tool_use", "web_search"), + _Block("tool_use", "submit_claims", {"claims": [_VALID_CLAIM]}), + ], + usage=_FakeUsage(i=100, o=50, cw=5, cr=10), + stop_reason="tool_use", + ) + _patch_responses(monkeypatch, [response]) + + result = asyncio.run(extract_and_verify("trois mots ici", web_search=True)) + + assert len(result.claims) == 1 + assert result.api_calls == 1 + assert result.web_search_calls == 1 + assert result.usage == { + "input_tokens": 100, + "output_tokens": 50, + "cache_write": 5, + "cache_read": 10, + } + + +def test_two_turn_fallback_sums_usage_and_calls(monkeypatch) -> None: + first = _FakeResponse( + content=[_Block("server_tool_use", "web_search")], # searched, no submit + usage=_FakeUsage(i=200, o=20, cw=0, cr=0), + stop_reason="tool_use", + ) + second = _FakeResponse( + content=[_Block("tool_use", "submit_claims", {"claims": [_VALID_CLAIM]})], + usage=_FakeUsage(i=80, o=40, cw=0, cr=5), + stop_reason="tool_use", + ) + _patch_responses(monkeypatch, [first, second]) + + result = asyncio.run(extract_and_verify("trois mots ici", web_search=True)) + + assert result.api_calls == 2 + assert result.web_search_calls == 1 # only the first turn searched + assert result.usage["input_tokens"] == 280 + assert result.usage["output_tokens"] == 60 + assert result.usage["cache_read"] == 5 + assert len(result.claims) == 1 + + +def test_below_min_words_makes_no_call(monkeypatch) -> None: + def _boom(*_args, **_kwargs): + raise AssertionError("the API must not be called for a too-short utterance") + + monkeypatch.setattr(claim_extractor._client.messages, "create", _boom) + + result = asyncio.run(extract_and_verify("hi")) + + assert result.claims == [] + assert result.api_calls == 0 + assert result.usage == {} diff --git a/tests/test_session_persistence.py b/tests/test_session_persistence.py new file mode 100644 index 0000000..0e572af --- /dev/null +++ b/tests/test_session_persistence.py @@ -0,0 +1,49 @@ +"""Tests for the lazy session-row creation in session.py. + +A connection only gets persisted once it produces a transcript, so an empty +connection (open → close, no speech) leaves no row behind. +""" + +import asyncio + +from app.services import session +from app.services.session_store import utcnow + + +def _info() -> dict: + return { + "id": "sess-1", + "client": "127.0.0.1", + "started_at": utcnow(), + "persisted": False, + } + + +def test_creates_row_once(monkeypatch) -> None: + calls: list[str] = [] + monkeypatch.setattr( + session.session_store, + "create_session", + lambda session_id, client, started_at: calls.append(session_id), + ) + + info = _info() + asyncio.run(session._ensure_persisted(info)) + asyncio.run(session._ensure_persisted(info)) # second call is a no-op + + assert calls == ["sess-1"] + assert info["persisted"] is True + + +def test_no_write_when_persistence_disabled(monkeypatch) -> None: + calls: list[str] = [] + monkeypatch.setattr(session.settings, "PERSIST_SESSIONS", False) + monkeypatch.setattr( + session.session_store, + "create_session", + lambda *args: calls.append("called"), + ) + + asyncio.run(session._ensure_persisted(_info())) + + assert calls == [] diff --git a/tests/test_sessions_route.py b/tests/test_sessions_route.py new file mode 100644 index 0000000..8862e43 --- /dev/null +++ b/tests/test_sessions_route.py @@ -0,0 +1,144 @@ +"""Integration tests for the /sessions read & export routes. + +Overrides ``get_db`` with a shared in-memory SQLite so the routes hit a seeded +test database instead of the real file. +""" + +from datetime import datetime + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import sessionmaker +from sqlalchemy.pool import StaticPool + +from app.db.base import Base +from app.db.session import get_db +from app.dependencies import require_admin +from app.main import app +from app.models import Claim, Session, TranscriptSegment + +_engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, # one shared connection so the in-memory DB persists +) +_TestSession = sessionmaker(bind=_engine, expire_on_commit=False) +Base.metadata.create_all(_engine) + + +def _override_get_db(): + db = _TestSession() + try: + yield db + finally: + db.close() + + +app.dependency_overrides[get_db] = _override_get_db +client = TestClient(app) + + +@pytest.fixture(autouse=True) +def _seed() -> None: + # These routes are admin-gated; bypass the JWT check per test (another test + # module pops this override in its teardown, so set it fresh each time). + app.dependency_overrides[require_admin] = lambda: "test-admin" + with _TestSession() as db: + db.query(Claim).delete() + db.query(TranscriptSegment).delete() + db.query(Session).delete() + session = Session( + id="sess-1", + started_at=datetime(2026, 1, 1, 10, 0, 0), + ended_at=datetime(2026, 1, 1, 10, 0, 20), + client_host="127.0.0.1", + chunks_received=5, + ) + session.segments = [ + TranscriptSegment( + id="seg-1", + session_id="sess-1", + seq=0, + text="La Terre est plate.", + detected_language="fr", + language_probability=0.99, + created_at=datetime(2026, 1, 1, 10, 0, 1), + transcribe_ms=12.0, + verify_ms=120.0, + tokens_input=100, + tokens_output=50, + tokens_cache_read=0, + tokens_cache_write=0, + api_calls=1, + web_search_calls=0, + ) + ] + session.claims = [ + Claim( + id="claim-1", + session_id="sess-1", + segment_id="seg-1", + text="La Terre est plate.", + status="false", + explanation="La Terre est un géoïde.", + sources=["https://example.com"], + timestamp=1.0, + category="science", + confidence=6, + counter_claim="La Terre est sphérique.", + web_search_used=False, + created_at=datetime(2026, 1, 1, 10, 0, 2), + ) + ] + db.add(session) + db.commit() + + +def test_routes_require_admin() -> None: + # Drop the auth override to confirm the routes are actually gated. The autouse + # fixture re-adds it before the next test. + app.dependency_overrides.pop(require_admin, None) + assert client.get("/sessions").status_code == 401 + assert client.get("/sessions/sess-1").status_code == 401 + assert client.get("/sessions/sess-1/export").status_code == 401 + + +def test_list_sessions() -> None: + resp = client.get("/sessions") + assert resp.status_code == 200 + body = resp.json() + assert len(body) == 1 + assert body[0]["id"] == "sess-1" + assert body[0]["claims_count"] == 1 + assert body[0]["false_count"] == 1 + assert body[0]["active"] is False + + +def test_get_session_detail() -> None: + resp = client.get("/sessions/sess-1") + assert resp.status_code == 200 + body = resp.json() + assert body["stats"]["claims_count"] == 1 + assert body["stats"]["tokens"]["total"] == 150 + assert len(body["segments"]) == 1 + assert len(body["claims"]) == 1 + + +def test_get_unknown_session_is_404() -> None: + assert client.get("/sessions/nope").status_code == 404 + + +def test_export_markdown() -> None: + resp = client.get("/sessions/sess-1/export", params={"format": "md"}) + assert resp.status_code == 200 + assert resp.headers["content-type"].startswith("text/markdown") + assert "session-sess-1.md" in resp.headers["content-disposition"] + assert "### [false] La Terre est plate." in resp.text + + +def test_export_json_is_default() -> None: + resp = client.get("/sessions/sess-1/export") + assert resp.status_code == 200 + assert "session-sess-1.json" in resp.headers["content-disposition"] + assert resp.json()["id"] == "sess-1" diff --git a/tests/test_stats.py b/tests/test_stats.py new file mode 100644 index 0000000..bea779f --- /dev/null +++ b/tests/test_stats.py @@ -0,0 +1,148 @@ +"""Unit tests for the derived per-session statistics (pure, no DB). + +Builds transient ORM instances in memory and checks that ``compute_stats`` +aggregates them correctly — token sums, latencies, reject detection, cost. +""" + +from datetime import datetime + +from app.models import Claim, Session, TranscriptSegment +from app.services.stats import compute_stats + + +def _seg( + seg_id: str, + seq: int, + *, + transcribe_ms: float = 10.0, + verify_ms: float | None = None, + ti: int | None = None, + to: int | None = None, + cr: int | None = None, + cw: int | None = None, + api: int | None = None, + web: int | None = None, +) -> TranscriptSegment: + return TranscriptSegment( + id=seg_id, + session_id="s", + seq=seq, + text="t", + detected_language="fr", + language_probability=0.9, + created_at=datetime(2026, 1, 1), + transcribe_ms=transcribe_ms, + verify_ms=verify_ms, + tokens_input=ti, + tokens_output=to, + tokens_cache_read=cr, + tokens_cache_write=cw, + api_calls=api, + web_search_calls=web, + ) + + +def _claim( + claim_id: str, + *, + status: str = "verified", + confidence: int = 8, + segment_id: str = "seg1", + web: bool = False, +) -> Claim: + return Claim( + id=claim_id, + session_id="s", + segment_id=segment_id, + text="c", + status=status, + explanation="", + sources=[], + timestamp=0.0, + category="science", + confidence=confidence, + counter_claim="", + web_search_used=web, + created_at=datetime(2026, 1, 1), + ) + + +def _session() -> Session: + session = Session( + id="s", + started_at=datetime(2026, 1, 1, 10, 0, 0), + ended_at=datetime(2026, 1, 1, 10, 0, 30), + client_host="127.0.0.1", + chunks_received=42, + ) + session.segments = [ + _seg("seg1", 0, verify_ms=100, ti=100, to=50, cr=10, cw=5, api=1, web=0), + _seg("seg2", 1, verify_ms=200, ti=200, to=20, cr=0, cw=0, api=2, web=1), + _seg("seg3", 2, verify_ms=50, ti=10, to=5, cr=0, cw=0, api=1, web=0), + _seg("seg4", 3), # too short: extraction never ran (metrics NULL) + ] + session.claims = [ + _claim("c1", status="verified", confidence=8, segment_id="seg1"), + _claim("c2", status="false", confidence=6, segment_id="seg2", web=True), + ] + return session + + +def test_counts_and_ratios() -> None: + stats = compute_stats(_session(), "claude-haiku-4-5") + + assert stats.duration_s == 30.0 + assert stats.transcripts_count == 4 + assert stats.claims_count == 2 + assert stats.claims_by_status == {"verified": 1, "false": 1} + assert stats.dominant_category == "science" + assert stats.avg_confidence == 7.0 + + +def test_verification_and_reject_detection() -> None: + stats = compute_stats(_session(), "claude-haiku-4-5") + + # seg1, seg2, seg3 ran verification; seg3 produced no claim → one reject. + assert stats.segments_verified == 3 + assert stats.rejects == 1 + assert stats.api_calls_total == 4 + assert stats.fallback_count == 1 # seg2 needed two turns + + +def test_web_search_and_latency() -> None: + stats = compute_stats(_session(), "claude-haiku-4-5") + + assert stats.web_search_segments == 1 + assert stats.web_search_calls_total == 1 + assert stats.claims_with_web_search == 1 + assert stats.avg_transcribe_ms == 10.0 + assert stats.avg_verify_ms == round((100 + 200 + 50) / 3, 2) + + +def test_token_totals() -> None: + stats = compute_stats(_session(), "claude-haiku-4-5") + + assert stats.tokens.input == 310 + assert stats.tokens.output == 75 + assert stats.tokens.cache_read == 10 + assert stats.tokens.cache_write == 5 + assert stats.tokens.total == 400 + + +def test_cost_estimate_for_known_model() -> None: + stats = compute_stats(_session(), "claude-haiku-4-5") + # 310*1 + 75*5 + 5*1.25 + 10*0.1 = 692.25 USD per million tokens. + assert stats.estimated_cost_usd == round(692.25 / 1_000_000, 6) + assert stats.pricing_model == "claude-haiku-4-5" + + +def test_cost_is_none_for_unknown_model() -> None: + stats = compute_stats(_session(), "some-unpriced-model") + assert stats.estimated_cost_usd is None + + +def test_active_session_has_no_duration() -> None: + session = _session() + session.ended_at = None + stats = compute_stats(session, "claude-haiku-4-5") + assert stats.duration_s is None diff --git a/tests/test_transcription.py b/tests/test_transcription.py index d02694c..b60a5cd 100644 --- a/tests/test_transcription.py +++ b/tests/test_transcription.py @@ -1,4 +1,4 @@ -"""Tests for transcribe_chunk, with the Whisper model mocked. +"""Tests for transcribe_samples, with the Whisper model mocked. No real audio or model: we patch ``_get_model`` so the test stays offline and fast, and assert on what gets passed to ``.transcribe`` and what comes back. @@ -7,6 +7,7 @@ from types import SimpleNamespace from unittest.mock import MagicMock +import numpy as np import pytest from app.services import transcription @@ -24,11 +25,14 @@ def mock_model(monkeypatch: pytest.MonkeyPatch) -> MagicMock: def test_always_auto_detects_and_reports(mock_model: MagicMock) -> None: - text, lang, prob = transcription.transcribe_chunk(b"audio") + audio = np.zeros(16000, dtype=np.float32) + text, lang, prob = transcription.transcribe_samples(audio) # Transcription always runs in auto mode (language=None) so a mismatched - # chunk is never translated; the detected language is surfaced for filtering. + # utterance is never translated; the detected language is surfaced for filtering. assert mock_model.transcribe.call_args.kwargs["language"] is None + # The decoded waveform is forwarded as-is (no ffmpeg decode on the live path). + assert mock_model.transcribe.call_args.args[0] is audio assert text == "Hello world." assert lang == "en" assert prob == 0.987 @@ -39,4 +43,5 @@ def test_error_returns_empty_triple(monkeypatch: pytest.MonkeyPatch) -> None: model.transcribe.side_effect = RuntimeError("boom") monkeypatch.setattr(transcription, "_get_model", lambda: model) - assert transcription.transcribe_chunk(b"audio") == ("", "", 0.0) + audio = np.zeros(16000, dtype=np.float32) + assert transcription.transcribe_samples(audio) == ("", "", 0.0)