diff --git a/.env.example b/.env.example index c1dd568..fada88b 100644 --- a/.env.example +++ b/.env.example @@ -1,6 +1,20 @@ LINNET_LICENSE_KEY= -CF_VOICE_MOCK=1 +CF_VOICE_MOCK=0 CF_VOICE_WHISPER_MODEL=small +CF_VOICE_DIARIZE=0 HF_TOKEN= LINNET_PORT=8522 LINNET_FRONTEND_PORT=8521 + +# cf-orch coordinator URL — enables real cf-voice sidecar allocation per session. +# Leave empty to fall back to in-process mock (also set CF_VOICE_MOCK=1 for mock audio). +CF_ORCH_URL= + +# cf-orch agent URL — local service lifecycle (stop/start). +# Coordinator (:7700) = read/scheduling; Agent (:7701) = write operations. +CF_ORCH_AGENT_URL= + +# Vite dev server base paths — set when serving under a sub-path (e.g. /linnet/ via Caddy). +# Leave empty for root-path local dev (http://localhost:8521). +VITE_BASE_URL= +VITE_API_BASE= diff --git a/.nfs0000000000bc6a8c00021df0 b/.nfs0000000000bc6a8c00021df0 new file mode 100755 index 0000000..37ec41d --- /dev/null +++ b/.nfs0000000000bc6a8c00021df0 @@ -0,0 +1,169 @@ +#!/usr/bin/env bash +# manage.sh — Linnet dev/demo/cloud management script +set -euo pipefail + +CMD=${1:-help} +PROFILE=${2:-dev} # dev | demo | cloud +API_PORT=${LINNET_PORT:-8522} +FE_PORT=${LINNET_FRONTEND_PORT:-8521} + +# ── Helpers ─────────────────────────────────────────────────────────────────── +GREEN='\033[0;32m'; YELLOW='\033[1;33m'; RED='\033[0;31m'; NC='\033[0m' +info() { echo -e "${GREEN}[linnet]${NC} $*"; } +warn() { echo -e "${YELLOW}[linnet]${NC} $*"; } +error() { echo -e "${RED}[linnet]${NC} $*" >&2; exit 1; } + +_check_env() { + if [[ ! -f .env ]]; then + warn "No .env found — copying .env.example" + cp .env.example .env + fi +} + +_compose_cmd() { + # Returns the correct compose file flags for a given profile + case "$1" in + demo) echo "-f compose.demo.yml -p linnet-demo" ;; + cloud) echo "-f compose.cloud.yml -p linnet-cloud" ;; + test) echo "-f compose.test.yml" ;; + dev) echo "-f compose.yml" ;; + *) error "Unknown profile: $1. Use dev|demo|cloud|test." ;; + esac +} + +# ── Commands ────────────────────────────────────────────────────────────────── +case "$CMD" in + + start) + _check_env + if [[ "$PROFILE" == "dev" ]]; then + # Load .env into the current shell so uvicorn and Vite inherit all settings + # (CF_VOICE_MOCK, CF_ORCH_URL, VITE_BASE_URL, VITE_API_BASE, etc.) + set -a; source .env; set +a + info "Starting Linnet API on :${API_PORT}…" + conda run -n cf uvicorn app.main:app \ + --host 0.0.0.0 --port "$API_PORT" --reload & + info "Starting Linnet frontend on :${FE_PORT}…" + cd frontend && npm install --silent && npm run dev & + info "API: http://localhost:${API_PORT}" + info "UI: http://localhost:${FE_PORT}" + wait + else + FLAGS=$(_compose_cmd "$PROFILE") + info "Starting Linnet ($PROFILE profile)…" + # shellcheck disable=SC2086 + docker compose $FLAGS up -d --build + case "$PROFILE" in + demo) info "Frontend: http://localhost:8524" ;; + cloud) info "Frontend: http://localhost:8527 → menagerie.circuitforge.tech/linnet" ;; + esac + fi + ;; + + stop) + if [[ "$PROFILE" == "dev" ]]; then + info "Stopping Linnet dev processes…" + pkill -f "uvicorn app.main:app" 2>/dev/null || true + pkill -f "vite" 2>/dev/null || true + else + FLAGS=$(_compose_cmd "$PROFILE") + # shellcheck disable=SC2086 + docker compose $FLAGS down + fi + info "Stopped ($PROFILE)." + ;; + + restart) + "$0" stop "$PROFILE" + "$0" start "$PROFILE" + ;; + + status) + case "$PROFILE" in + dev) + echo -n "API: " + curl -sf "http://localhost:${API_PORT}/health" 2>/dev/null \ + | python3 -c "import sys,json; d=json.load(sys.stdin); print(d['status'], '|', d.get('mode','dev'))" \ + || echo "not running" + echo -n "Frontend: " + curl -sf "http://localhost:${FE_PORT}" -o /dev/null && echo "running" || echo "not running" + ;; + demo) + echo -n "Demo API: "; curl -sf "http://localhost:8523/health" 2>/dev/null | python3 -c "import sys,json; d=json.load(sys.stdin); print(d['status'])" || echo "not running" + echo -n "Demo Web: "; curl -sf "http://localhost:8524" -o /dev/null && echo "running" || echo "not running" + ;; + cloud) + echo -n "Cloud API: "; curl -sf "http://localhost:8522/health" 2>/dev/null | python3 -c "import sys,json; d=json.load(sys.stdin); print(d['status'])" || echo "not running" + echo -n "Cloud Web: "; curl -sf "http://localhost:8527" -o /dev/null && echo "running" || echo "not running" + ;; + esac + ;; + + test) + _check_env + if [[ "$PROFILE" == "dev" ]]; then + info "Running test suite (local)…" + CF_VOICE_MOCK=1 conda run -n cf python -m pytest tests/ -v "${@:3}" + else + info "Running test suite (Docker)…" + docker compose -f compose.test.yml run --rm linnet-test "${@:3}" + fi + ;; + + logs) + if [[ "$PROFILE" == "dev" ]]; then + warn "Dev mode: check terminal output." + else + FLAGS=$(_compose_cmd "$PROFILE") + SERVICE=${3:-} + # shellcheck disable=SC2086 + docker compose $FLAGS logs -f $SERVICE + fi + ;; + + build) + info "Building Docker images ($PROFILE)…" + FLAGS=$(_compose_cmd "$PROFILE") + # shellcheck disable=SC2086 + docker compose $FLAGS build + info "Build complete." + ;; + + open) + case "$PROFILE" in + demo) URL="http://localhost:8524" ;; + cloud) URL="http://localhost:8527" ;; + *) URL="http://localhost:${FE_PORT}" ;; + esac + xdg-open "$URL" 2>/dev/null || open "$URL" 2>/dev/null || info "Open $URL in your browser" + ;; + + help|*) + echo "" + echo " Usage: $0 [profile]" + echo "" + echo " Commands:" + echo -e " ${GREEN}start${NC} [profile] Start API + frontend" + echo -e " ${GREEN}stop${NC} [profile] Stop services" + echo -e " ${GREEN}restart${NC} [profile] Stop then start" + echo -e " ${GREEN}status${NC} [profile] Check running services" + echo -e " ${GREEN}test${NC} [profile] Run pytest suite" + echo -e " ${GREEN}logs${NC} [profile] Tail logs (Docker profiles only)" + echo -e " ${GREEN}build${NC} [profile] Build Docker images" + echo -e " ${GREEN}open${NC} [profile] Open UI in browser" + echo "" + echo " Profiles:" + echo " dev (default) Local uvicorn + Vite dev server, mock mode" + echo " demo Docker: DEMO_MODE=true, port 8524" + echo " cloud Docker: CLOUD_MODE=true, port 8527 → menagerie.circuitforge.tech/linnet" + echo " test Docker: pytest suite, hermetic" + echo "" + echo " Examples:" + echo " $0 start # dev mode" + echo " $0 start demo # demo stack" + echo " $0 test # local pytest" + echo " $0 test docker # pytest in Docker" + echo " $0 logs demo # demo logs" + echo "" + ;; +esac diff --git a/README.md b/README.md index a3565f4..965b4d5 100644 --- a/README.md +++ b/README.md @@ -148,3 +148,7 @@ Linnet is developed and maintained on Forgejo at [git.opensourcesolarpunk.com/Ci --- *Linnet is a product of [CircuitForge LLC](https://circuitforge.tech) — privacy-first, safety-first, accessible AI tools for the tasks the system made hard on purpose.* + +--- + +Humans own design, architecture, code review, testing, and verification. LLMs are part of our development workflow. [Our positions on LLM use →](https://circuitforge.tech/positions) diff --git a/app/api/dev.py b/app/api/dev.py new file mode 100644 index 0000000..47a2bba --- /dev/null +++ b/app/api/dev.py @@ -0,0 +1,79 @@ +# app/api/dev.py — dev-only endpoints for local service management +# +# Proxies stop/start calls to the cf-orch agent so the frontend dev panel +# doesn't need to know the agent URL directly. +# Not registered in cloud or demo mode. +from __future__ import annotations + +import logging + +import httpx +from fastapi import APIRouter, HTTPException + +from app.config import settings + +router = APIRouter(prefix="/dev", tags=["dev"]) +logger = logging.getLogger(__name__) + + +def _agent_url() -> str: + url = settings.cf_orch_agent_url + if not url: + raise HTTPException(503, detail="CF_ORCH_AGENT_URL not configured") + return url.rstrip("/") + + +@router.post("/voice/stop") +async def voice_stop() -> dict: + """Stop the cf-voice sidecar via the cf-orch agent.""" + agent = _agent_url() + async with httpx.AsyncClient(timeout=10) as client: + try: + resp = await client.post(f"{agent}/services/cf-voice/stop", json={}) + except httpx.RequestError as exc: + raise HTTPException(502, detail=f"Agent unreachable: {exc}") from exc + if resp.status_code >= 400: + raise HTTPException(resp.status_code, detail=resp.text) + logger.info("dev: cf-voice stop → %s", resp.status_code) + return {"ok": True, "action": "stop"} + + +@router.post("/voice/start") +async def voice_start() -> dict: + """Start the cf-voice sidecar via the cf-orch agent.""" + agent = _agent_url() + async with httpx.AsyncClient(timeout=10) as client: + try: + resp = await client.post(f"{agent}/services/cf-voice/start", json={}) + except httpx.RequestError as exc: + raise HTTPException(502, detail=f"Agent unreachable: {exc}") from exc + if resp.status_code >= 400: + raise HTTPException(resp.status_code, detail=resp.text) + logger.info("dev: cf-voice start → %s", resp.status_code) + return {"ok": True, "action": "start"} + + +@router.post("/voice/restart") +async def voice_restart() -> dict: + """Stop then start cf-voice via the cf-orch agent.""" + import asyncio + + agent = _agent_url() + async with httpx.AsyncClient(timeout=10) as client: + try: + await client.post(f"{agent}/services/cf-voice/stop", json={}) + except httpx.RequestError as exc: + raise HTTPException(502, detail=f"Agent unreachable on stop: {exc}") from exc + + # Give the process time to fully exit before start fires + await asyncio.sleep(2.0) + + async with httpx.AsyncClient(timeout=10) as client: + try: + resp = await client.post(f"{agent}/services/cf-voice/start", json={}) + except httpx.RequestError as exc: + raise HTTPException(502, detail=f"Agent unreachable on start: {exc}") from exc + if resp.status_code >= 400: + raise HTTPException(resp.status_code, detail=resp.text) + logger.info("dev: cf-voice restart complete") + return {"ok": True, "action": "restart"} diff --git a/app/api/events.py b/app/api/events.py index 287e7cc..719b800 100644 --- a/app/api/events.py +++ b/app/api/events.py @@ -3,11 +3,15 @@ from __future__ import annotations import asyncio import logging +from dataclasses import replace from fastapi import APIRouter, HTTPException, Request from fastapi.responses import StreamingResponse +from app.models.service_event import ServiceEvent +from app.models.tone_event import ToneEvent from app.services import session_store +from app.services.translator import Translator logger = logging.getLogger(__name__) router = APIRouter(prefix="/session", tags=["events"]) @@ -22,6 +26,10 @@ async def stream_events(session_id: str, request: Request) -> StreamingResponse: const es = new EventSource(`/session/${sessionId}/stream`) es.addEventListener('tone-event', e => { ... }) + If the session has a target_lang set, ToneEvent labels are translated via + DeepL before emission. Labels are cached per-session — each unique label + is translated once regardless of how often it appears. + The stream runs until the client disconnects or the session ends. """ session = session_store.get_session(session_id) @@ -29,16 +37,32 @@ async def stream_events(session_id: str, request: Request) -> StreamingResponse: raise HTTPException(status_code=404, detail=f"Session {session_id} not found") queue = session.subscribe() + translator = Translator.for_session(session) async def generator(): try: # Heartbeat comment every 15s to keep connection alive through proxies yield ": heartbeat\n\n" + + # If the voice probe already completed before this subscriber connected, + # emit the result immediately rather than waiting for a queue event + # that was already broadcast to an empty subscriber list. + if session.voice_ready: + yield ServiceEvent(session_id=session_id, status="ready").to_sse() + elif session.voice_error: + yield ServiceEvent( + session_id=session_id, status="error", detail=session.voice_error, + ).to_sse() while True: if await request.is_disconnected(): break try: event = await asyncio.wait_for(queue.get(), timeout=15.0) + # Translate tone label if a translator is configured for this session + if translator is not None and isinstance(event, ToneEvent): + translated = translator.translate(event.label) + if translated != event.label: + event = replace(event, label=translated) yield event.to_sse() except asyncio.TimeoutError: yield ": heartbeat\n\n" diff --git a/app/api/sessions.py b/app/api/sessions.py index 3a2523f..89b8832 100644 --- a/app/api/sessions.py +++ b/app/api/sessions.py @@ -1,9 +1,11 @@ # app/api/sessions.py — session lifecycle endpoints from __future__ import annotations -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, HTTPException, Request from pydantic import BaseModel +from app import tiers +from app.config import settings from app.services import session_store router = APIRouter(prefix="/session", tags=["sessions"]) @@ -11,22 +13,61 @@ router = APIRouter(prefix="/session", tags=["sessions"]) class StartRequest(BaseModel): elcor: bool = False # enable Elcor subtext format (easter egg) + # Self-hosted installs may pass their license key per-request instead of via env var. + # Cloud mode ignores this field — tier is resolved from the authenticated user_id. + license_key: str | None = None + # Translation: BCP-47 target language tag (e.g. "es", "fr", "de"). + # Empty string or omitted = no translation. + target_lang: str = "" + # BYOK DeepL API key — user's own DeepL Free key. + # Bypasses the Paid tier gate for translation only (see tiers.py BYOK_UNLOCKABLE). + byok_deepl_key: str = "" + # Audio classification tuning. + # window_ms: PCM accumulation window before each /classify call (multiple of 100). + # Longer = more Whisper context = better accuracy, higher latency. Default 1000ms. + # transcribe_lang: BCP-47 hint for Whisper ("en", "es", …). Empty = auto-detect. + # num_speakers: hint for pyannote diarization. 0 = auto-detect; 1–8 = fixed count. + # Fixed count (e.g. 2) improves boundary accuracy for known speaker setups. + window_ms: int = 1000 + transcribe_lang: str = "" + num_speakers: int = 0 class SessionResponse(BaseModel): session_id: str state: str elcor: bool + tier: str # "free" | "paid" | "premium" @router.post("/start", response_model=SessionResponse) -async def start_session(req: StartRequest = StartRequest()) -> SessionResponse: +async def start_session( + request: Request, + req: StartRequest = StartRequest(), +) -> SessionResponse: """Start a new annotation session and begin streaming VoiceFrames.""" - session = await session_store.create_session(elcor=req.elcor) + # In cloud mode, user_id was set by CloudAuthMiddleware. + # In self-hosted mode, fall back to the per-request key or env var. + user_id: str | None = getattr(request.state, "cf_user", None) if settings.cloud_mode else None + license_key: str | None = None if settings.cloud_mode else req.license_key + + tier = tiers.get_tier(license_key=license_key, user_id=user_id) + + session = await session_store.create_session( + elcor=req.elcor, + tier=tier, + user_id=user_id or "", + target_lang=req.target_lang, + byok_deepl_key=req.byok_deepl_key, + window_ms=max(100, (req.window_ms // 100) * 100), # round down to nearest 100ms + transcribe_lang=req.transcribe_lang, + num_speakers=max(0, min(8, req.num_speakers)), # clamp 0–8 + ) return SessionResponse( session_id=session.session_id, state=session.state, elcor=session.elcor, + tier=tier, ) @@ -40,12 +81,16 @@ async def end_session(session_id: str) -> dict: @router.get("/{session_id}") -def get_session(session_id: str) -> SessionResponse: +def get_session(request: Request, session_id: str) -> SessionResponse: session = session_store.get_session(session_id) if session is None: raise HTTPException(status_code=404, detail=f"Session {session_id} not found") + user_id: str | None = getattr(request.state, "cf_user", None) if settings.cloud_mode else None + license_key: str | None = None if settings.cloud_mode else request.headers.get("X-CF-License-Key") + tier = tiers.get_tier(license_key=license_key, user_id=user_id) return SessionResponse( session_id=session.session_id, state=session.state, elcor=session.elcor, + tier=tier, ) diff --git a/app/api/snapshots.py b/app/api/snapshots.py new file mode 100644 index 0000000..ff0ff68 --- /dev/null +++ b/app/api/snapshots.py @@ -0,0 +1,156 @@ +# app/api/snapshots.py — session snapshot list and resume (Paid tier) +# +# Snapshots are saved automatically when a Paid session ends with history. +# Users can list their past sessions and resume them (imports history into a new session). +from __future__ import annotations + +import json + +from fastapi import APIRouter, Depends, HTTPException, Request +from pydantic import BaseModel + +from app import tiers +from app.config import settings +from app.db import get_db +from app.services import session_store + +router = APIRouter(tags=["snapshots"]) + + +def _resolve_user_id(request: Request) -> str: + """Return the cloud user_id from request state, or empty string for self-hosted.""" + return getattr(request.state, "cf_user", "") if settings.cloud_mode else "" + + +# ── List snapshots ──────────────────────────────────────────────────────────── + +class SnapshotSummary(BaseModel): + id: str + session_id: str + created_at: float + ended_at: float + elcor: bool + event_count: int + + +@router.get("/sessions/history", response_model=list[SnapshotSummary]) +def list_snapshots( + request: Request, + limit: int = 20, + conn=Depends(get_db), +) -> list[SnapshotSummary]: + """ + Return the user's last N saved sessions (Paid tier only). + + Self-hosted: scoped to the instance (user_id = ''). + Cloud: scoped to the authenticated user. + """ + user_id = _resolve_user_id(request) + license_key = None if settings.cloud_mode else request.headers.get("X-CF-License-Key") + tiers.require_paid(license_key=license_key, user_id=user_id or None) + + rows = conn.execute( + """ + SELECT id, session_id, created_at, ended_at, elcor, event_count + FROM session_snapshots + WHERE user_id = ? + ORDER BY ended_at DESC + LIMIT ? + """, + (user_id, min(limit, 100)), + ).fetchall() + + return [ + SnapshotSummary( + id=row[0], + session_id=row[1], + created_at=row[2], + ended_at=row[3], + elcor=bool(row[4]), + event_count=row[5], + ) + for row in rows + ] + + +# ── Resume a snapshot ───────────────────────────────────────────────────────── + +class ResumeRequest(BaseModel): + snapshot_id: str + elcor: bool | None = None # override Elcor mode; None = use snapshot's setting + target_lang: str = "" + byok_deepl_key: str = "" + license_key: str | None = None # self-hosted key + + +class ResumeResponse(BaseModel): + session_id: str + state: str + elcor: bool + tier: str + resumed_from: str # snapshot_id + event_count: int # number of historical events injected + + +@router.post("/session/resume", response_model=ResumeResponse) +async def resume_session( + request: Request, + req: ResumeRequest, + conn=Depends(get_db), +) -> ResumeResponse: + """ + Resume a previous session by restoring its annotation history into a new session. + + The new session starts fresh (new session_id, new classifier) but is pre-populated + with the historical ToneEvents from the snapshot so the frontend can display them. + + Paid tier only. The snapshot must belong to the authenticated user. + """ + user_id = _resolve_user_id(request) + license_key = None if settings.cloud_mode else req.license_key + tiers.require_paid(license_key=license_key, user_id=user_id or None) + + tier = tiers.get_tier(license_key=license_key, user_id=user_id or None) + + row = conn.execute( + "SELECT session_id, elcor, events_json FROM session_snapshots WHERE id = ? AND user_id = ?", + (req.snapshot_id, user_id), + ).fetchone() + if row is None: + raise HTTPException(status_code=404, detail="Snapshot not found") + + orig_session_id, orig_elcor, events_json = row + use_elcor = req.elcor if req.elcor is not None else bool(orig_elcor) + + session = await session_store.create_session( + elcor=use_elcor, + tier=tier, + user_id=user_id, + target_lang=req.target_lang, + byok_deepl_key=req.byok_deepl_key, + ) + + # Inject historical events so the frontend can render the prior session + from app.models.tone_event import ToneEvent + events_data = json.loads(events_json) + for ev in events_data: + tone = ToneEvent( + session_id=session.session_id, + label=ev["label"], + confidence=ev["confidence"], + speaker_id=ev.get("speaker_id", "speaker_a"), + shift_magnitude=ev.get("shift_magnitude", 0.0), + timestamp=ev["timestamp"], + subtext=ev.get("subtext", ""), + affect=ev.get("affect", ""), + ) + session.history.append(tone) + + return ResumeResponse( + session_id=session.session_id, + state=session.state, + elcor=session.elcor, + tier=tier, + resumed_from=req.snapshot_id, + event_count=len(events_data), + ) diff --git a/app/config.py b/app/config.py index 32366ab..82149fd 100644 --- a/app/config.py +++ b/app/config.py @@ -15,17 +15,27 @@ class Settings: # DEMO: auto-kill sessions after this many seconds of inactivity demo_session_ttl_s: int = int(os.getenv("DEMO_SESSION_TTL_S", "300")) # 5 min - # All modes: kill a session this many seconds after its last SSE subscriber - # disconnects. Covers mobile tab timeout / screen lock / crash. - # Set generously enough to survive a brief screen-off without losing the session, - # but short enough that zombie sessions don't accumulate. + # Free tier: kill session this many seconds after the last SSE subscriber leaves. + # Covers mobile tab timeout / screen lock / crash. session_idle_ttl_s: int = int(os.getenv("SESSION_IDLE_TTL_S", "90")) + # Paid tier: hard idle timeout — kill session this many seconds after the last + # annotation event, regardless of subscriber state. 30 minutes default. + session_paid_ttl_s: int = int(os.getenv("SESSION_PAID_TTL_S", "1800")) + + # DeepL API key for CF-managed cloud translation (Paid tier). + # BYOK users supply their own key per-session in StartRequest.byok_deepl_key. + deepl_api_key: str = os.getenv("DEEPL_API_KEY", "") # CLOUD: where Caddy injects the cf_session user token cloud_session_header: str = os.getenv("CLOUD_SESSION_HEADER", "X-CF-Session") cloud_data_root: str = os.getenv("CLOUD_DATA_ROOT", "/devl/linnet-cloud-data") heimdall_url: str = os.getenv("HEIMDALL_URL", "https://license.circuitforge.tech") + # Admin token for cloud-mode user tier resolution (/admin/cloud/resolve). + # Not required for self-hosted installs; required when CLOUD_MODE=true. + heimdall_admin_token: str = os.getenv("HEIMDALL_ADMIN_TOKEN", "") + # Self-hosted license key. Cloud mode resolves tier via user_id instead. + linnet_license_key: str = os.getenv("LINNET_LICENSE_KEY", "") linnet_port: int = int(os.getenv("LINNET_PORT", "8522")) linnet_frontend_port: int = int(os.getenv("LINNET_FRONTEND_PORT", "8521")) @@ -35,6 +45,10 @@ class Settings: # When set, each session allocates a cf-voice sidecar via the coordinator REST API. cf_orch_url: str = os.getenv("CF_ORCH_URL", "") + # cf-orch agent URL — local service lifecycle (start/stop). Separate from the + # coordinator: coordinator is read/scheduling (:7700), agent is write (:7701). + cf_orch_agent_url: str = os.getenv("CF_ORCH_AGENT_URL", "") + # Static cf-voice sidecar URL — legacy override (bypasses cf-orch, points directly # at a running cf-voice process). Takes precedence over cf-orch allocation when set. cf_voice_url: str = os.getenv("CF_VOICE_URL", "") diff --git a/app/main.py b/app/main.py index 0b14613..d51fd82 100644 --- a/app/main.py +++ b/app/main.py @@ -8,7 +8,7 @@ from contextlib import asynccontextmanager from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from app.api import audio, corrections, events, export, history, samples, sessions +from app.api import audio, corrections, dev, events, export, history, samples, sessions, snapshots from app.config import settings from app.services import session_store @@ -68,11 +68,14 @@ app.add_middleware( # ── Routers ─────────────────────────────────────────────────────────────────── app.include_router(sessions.router) app.include_router(events.router) +if not settings.demo_mode and not settings.cloud_mode: + app.include_router(dev.router) app.include_router(history.router) app.include_router(audio.router) app.include_router(export.router) app.include_router(samples.router) app.include_router(corrections.router, prefix="/corrections", tags=["corrections"]) +app.include_router(snapshots.router) @app.get("/health") diff --git a/app/migrations/002_session_snapshots.sql b/app/migrations/002_session_snapshots.sql new file mode 100644 index 0000000..8ce0f59 --- /dev/null +++ b/app/migrations/002_session_snapshots.sql @@ -0,0 +1,21 @@ +-- Migration 002: Session snapshots for Paid tier session pinning. +-- When a Paid session ends, its annotation history is saved here so the user +-- can resume it in a later session. Snapshots contain annotation metadata only — +-- no audio, no raw transcription text unless the user has opted in to STT retention. +-- +-- Privacy: user_id scopes snapshots to the authenticated user. For self-hosted +-- installs with no cloud auth, user_id is an empty string (instance-shared). + +CREATE TABLE IF NOT EXISTS session_snapshots ( + id TEXT PRIMARY KEY, -- UUID + user_id TEXT NOT NULL DEFAULT '', -- cf_user from cloud mode; '' for self-hosted + session_id TEXT NOT NULL, -- original session_id (short 8-char) + created_at REAL NOT NULL, -- Unix epoch (time.time()) + ended_at REAL NOT NULL, -- Unix epoch + elcor INTEGER NOT NULL DEFAULT 0, -- 1 = Elcor subtext mode was active + event_count INTEGER NOT NULL DEFAULT 0, + events_json TEXT NOT NULL DEFAULT '[]' -- JSON array of ToneEvent dicts +); + +CREATE INDEX IF NOT EXISTS idx_snapshots_user_time + ON session_snapshots (user_id, ended_at DESC); diff --git a/app/models/scene_event.py b/app/models/scene_event.py new file mode 100644 index 0000000..f3e00ab --- /dev/null +++ b/app/models/scene_event.py @@ -0,0 +1,58 @@ +# app/models/scene_event.py — acoustic scene and accent SSE event models +from __future__ import annotations + +import json +from dataclasses import dataclass + + +@dataclass +class SceneEvent: + """ + Acoustic scene classification from cf-voice. + + Broad scene label (e.g. "indoor_quiet", "outdoor_urban") derived from the + AST model. Primary input to the privacy risk indicator in the frontend. + Broadcasts via SSE as `event: scene-event`. + Not stored in session history. + """ + session_id: str + label: str # SCENE_LABELS: indoor_quiet, outdoor_urban, etc. + confidence: float + timestamp: float + privacy_risk: str = "low" # "low" | "moderate" | "high" + + def to_sse(self) -> str: + payload = { + "session_id": self.session_id, + "label": self.label, + "confidence": self.confidence, + "timestamp": self.timestamp, + "privacy_risk": self.privacy_risk, + } + return f"event: scene-event\ndata: {json.dumps(payload)}\n\n" + + +@dataclass +class AccentEvent: + """ + Regional accent / language identification from cf-voice. + + Gated by CF_VOICE_ACCENT=1 — only emitted when that flag is set. + Broadcasts via SSE as `event: accent-event`. + Not stored in session history or corrections DB. + """ + session_id: str + region: str # ACCENT_LABELS: en_gb, en_us, fr, etc. + language: str # raw language tag from the model + confidence: float + timestamp: float + + def to_sse(self) -> str: + payload = { + "session_id": self.session_id, + "region": self.region, + "language": self.language, + "confidence": self.confidence, + "timestamp": self.timestamp, + } + return f"event: accent-event\ndata: {json.dumps(payload)}\n\n" diff --git a/app/models/service_event.py b/app/models/service_event.py new file mode 100644 index 0000000..13eee7e --- /dev/null +++ b/app/models/service_event.py @@ -0,0 +1,30 @@ +# app/models/service_event.py — system/service lifecycle events for the SSE stream +from __future__ import annotations + +import json +from dataclasses import dataclass + + +@dataclass +class ServiceEvent: + """ + Lifecycle signal broadcast over SSE to frontend subscribers. + + Currently used to signal when the cf-voice backend is ready to accept audio. + The frontend holds the mic button in a disabled "loading" state until it + receives status="ready". + + SSE event name: service-event + """ + session_id: str + status: str # "ready" | "loading" | "error" + detail: str = "" # human-readable message shown in the UI + + def to_sse(self) -> str: + payload = { + "event_type": "service", + "session_id": self.session_id, + "status": self.status, + "detail": self.detail, + } + return f"event: service-event\ndata: {json.dumps(payload)}\n\n" diff --git a/app/models/session.py b/app/models/session.py index 7476cbf..e125985 100644 --- a/app/models/session.py +++ b/app/models/session.py @@ -32,19 +32,51 @@ class Session: created_at: float = field(default_factory=time.monotonic) elcor: bool = False + # Tier and user identity — set at creation from the license/auth resolution. + # Used by the reaper for tier-aware TTL and by snapshot logic for user scoping. + tier: str = "free" + user_id: str = "" # cf_user in cloud mode; empty string for self-hosted + # cf-voice sidecar — populated by session_store._allocate_voice() when cf-orch # is configured. Empty string means in-process fallback is active. cf_voice_url: str = "" cf_voice_allocation_id: str = "" - # History: last 50 events retained for GET /session/{id}/history + # Translation settings — set at session start from StartRequest. + # target_lang: BCP-47 tag (e.g. "es", "fr", "de") or empty string = no translation. + # byok_deepl_key: user-supplied DeepL Free API key; bypasses Paid gate for translation. + target_lang: str = "" + byok_deepl_key: str = "" + + # Audio classification settings — set at session start, cannot be changed mid-session. + # window_ms: how many ms of PCM to accumulate before calling cf-voice /classify. + # AudioWorklet sends 100ms chunks; must be a multiple of 100. Default 1000ms. + # transcribe_lang: BCP-47 language hint for Whisper (e.g. "en", "es"). + # Empty = Whisper auto-detects (slower, less accurate on short clips). + # num_speakers: hint for pyannote diarization. 0 = auto-detect; 1–8 = fixed count + # passed as min_speakers=max_speakers. Auto is slower; fixed count improves accuracy. + window_ms: int = 1000 + transcribe_lang: str = "" + num_speakers: int = 0 + + # History: last 200 events retained for GET /session/{id}/history and snapshots. history: list[ToneEvent] = field(default_factory=list) _subscribers: list[asyncio.Queue] = field(default_factory=list, repr=False) + # Voice readiness: set by _probe_voice_ready() once cf-voice health check passes. + # Used to immediately send a service-event to late SSE subscribers who connect + # after the probe has already completed (avoids the broadcast-before-subscribe race). + voice_ready: bool = False + voice_error: str = "" # non-empty when probe timed out + + # Activity tracking: updated on every broadcast() so the paid-tier reaper + # can measure 30-min hard idle (activity-based, not subscriber-based). + last_activity_at: float = field(default_factory=time.monotonic, repr=False) + # Idle tracking: monotonic timestamp of when the last subscriber left. # None means at least one subscriber is currently active, or the session - # has never had one yet (just started). The reaper uses this to detect - # abandoned sessions after a mobile tab timeout. + # has never had one yet (just started). The free-tier reaper uses this to + # detect abandoned sessions after a mobile tab timeout. last_subscriber_left_at: float | None = field(default=None, repr=False) def subscriber_count(self) -> int: @@ -69,8 +101,10 @@ class Session: def broadcast(self, event: SessionEvent) -> None: """Fan out any SessionEvent (tone, speaker, etc.) to all SSE subscribers.""" + self.last_activity_at = time.monotonic() + if isinstance(event, ToneEvent): - if len(self.history) >= 50: + if len(self.history) >= 200: self.history.pop(0) self.history.append(event) diff --git a/app/services/session_store.py b/app/services/session_store.py index 4be3ad7..6f2b406 100644 --- a/app/services/session_store.py +++ b/app/services/session_store.py @@ -6,6 +6,7 @@ import logging import time from app.config import settings +from app.models.service_event import ServiceEvent from app.models.session import Session from app.models.tone_event import ToneEvent from app.services.annotator import annotate @@ -26,13 +27,31 @@ _CHUNKS_PER_WINDOW = _CLASSIFY_WINDOW_MS // _CHUNK_MS # 10 chunks _audio_buffers: dict[str, list[bytes]] = {} -async def create_session(elcor: bool = False) -> Session: +async def create_session( + elcor: bool = False, + tier: str = "free", + user_id: str = "", + target_lang: str = "", + byok_deepl_key: str = "", + window_ms: int = 1000, + transcribe_lang: str = "", + num_speakers: int = 0, +) -> Session: """Create a new session and start its ContextClassifier background task. If CF_ORCH_URL is configured, requests a managed cf-voice instance before starting the classifier. Falls back to in-process mock if allocation fails. """ - session = Session(elcor=elcor) + session = Session( + elcor=elcor, + tier=tier, + user_id=user_id, + target_lang=target_lang, + byok_deepl_key=byok_deepl_key, + window_ms=window_ms, + transcribe_lang=transcribe_lang, + num_speakers=num_speakers, + ) _sessions[session.session_id] = session await _allocate_voice(session) task = asyncio.create_task( @@ -40,6 +59,10 @@ async def create_session(elcor: bool = False) -> Session: name=f"classifier-{session.session_id}", ) _tasks[session.session_id] = task + asyncio.create_task( + _probe_voice_ready(session), + name=f"voice-ready-{session.session_id}", + ) session.state = "running" logger.info( "Session %s started (voice=%s)", @@ -59,7 +82,11 @@ def active_session_count() -> int: async def end_session(session_id: str) -> bool: - """Stop and remove a session. Returns True if it existed.""" + """Stop and remove a session. Returns True if it existed. + + For Paid tier sessions with annotation history, saves a snapshot to SQLite + so the user can resume the session later. + """ session = _sessions.pop(session_id, None) if session is None: return False @@ -69,10 +96,194 @@ async def end_session(session_id: str) -> bool: task.cancel() _audio_buffers.pop(session_id, None) await _release_voice(session) + + if session.tier in ("paid", "premium") and session.history: + _save_snapshot(session) + logger.info("Session %s ended", session_id) return True +def _save_snapshot(session: Session) -> None: + """Persist a session's annotation history to the snapshots table.""" + import json + import time as _time + import uuid + + try: + from app.db import get_connection + conn = get_connection() + snapshot_id = str(uuid.uuid4()) + events_json = json.dumps([ + { + "label": e.label, + "confidence": e.confidence, + "speaker_id": e.speaker_id, + "shift_magnitude": e.shift_magnitude, + "timestamp": e.timestamp, + "subtext": e.subtext, + "affect": e.affect, + } + for e in session.history + ]) + conn.execute( + """ + INSERT INTO session_snapshots + (id, user_id, session_id, created_at, ended_at, elcor, event_count, events_json) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + snapshot_id, + session.user_id, + session.session_id, + session.created_at, + _time.time(), + 1 if session.elcor else 0, + len(session.history), + events_json, + ), + ) + conn.commit() + conn.close() + logger.info( + "Snapshot %s saved for session %s (%d events)", + snapshot_id, session.session_id, len(session.history), + ) + except Exception as exc: + logger.warning("Failed to save snapshot for session %s: %s", session.session_id, exc) + + +async def _probe_voice_ready(session: Session) -> None: + """Poll the cf-voice health endpoint and broadcast a service-event when ready. + + Retries every second for up to 30s. On success, broadcasts status="ready" + so the frontend can unlock the mic button. On timeout, broadcasts + status="error" so the UI can surface a warning. + + No-op for in-process mode (no cf_voice_url) — broadcasts ready immediately. + """ + def _mark_ready() -> None: + session.voice_ready = True + session.broadcast(ServiceEvent(session_id=session.session_id, status="ready")) + + def _mark_error(detail: str) -> None: + session.voice_error = detail + session.broadcast(ServiceEvent( + session_id=session.session_id, status="error", detail=detail, + )) + + if not session.cf_voice_url: + # In-process mock or no managed voice: ready immediately + _mark_ready() + return + + asyncio.create_task( + _poll_model_status(session), + name=f"model-status-{session.session_id}", + ) + + import httpx + + health_url = session.cf_voice_url.rstrip("/") + "/health" + deadline = time.monotonic() + 30.0 + + while time.monotonic() < deadline: + try: + async with httpx.AsyncClient(timeout=3.0) as client: + resp = await client.get(health_url) + if resp.status_code == 200: + _mark_ready() + logger.debug("cf-voice ready for session %s", session.session_id) + # Surface any soft warnings (e.g. diarize configured but HF_TOKEN missing) + try: + data = resp.json() + for msg in data.get("warnings", []): + session.broadcast(ServiceEvent( + session_id=session.session_id, + status="warning", + detail=msg, + )) + except Exception: + pass + return + except Exception: + pass + await asyncio.sleep(1.0) + + _mark_error("Voice service did not become ready in time. Try reloading.") + logger.warning("cf-voice readiness probe timed out for session %s", session.session_id) + + +_MODEL_LABELS = { + "stt": "speech-to-text (Whisper)", + "diarizer": "speaker diarization (pyannote)", + "dimensional": "emotion dimensions (audeering)", + "prosody": "prosody (openSMILE)", +} + + +async def _poll_model_status(session: Session) -> None: + """Poll cf-voice /health for per-model download progress and broadcast + loading/ready events to the frontend. Stops when all models are stable.""" + if not session.cf_voice_url: + return + + import httpx + + health_url = session.cf_voice_url.rstrip("/") + "/health" + prev: dict[str, str] = {} + + # Poll for up to 10 minutes (large model downloads can take a while) + deadline = time.monotonic() + 600.0 + + while time.monotonic() < deadline and session.state == "running": + try: + async with httpx.AsyncClient(timeout=3.0) as client: + resp = await client.get(health_url) + if resp.status_code == 200: + data = resp.json() + models: dict[str, str] = data.get("models", {}) + + for key, status in models.items(): + if models.get(key) == prev.get(key): + continue + label = _MODEL_LABELS.get(key, key) + if status == "loading": + session.broadcast(ServiceEvent( + session_id=session.session_id, + status="loading", + detail=f"Downloading {label}…", + )) + elif status == "ready" and prev.get(key) == "loading": + session.broadcast(ServiceEvent( + session_id=session.session_id, + status="loading", + detail=f"{label.capitalize()} ready.", + )) + elif status == "error": + session.broadcast(ServiceEvent( + session_id=session.session_id, + status="warning", + detail=f"Failed to load {label}.", + )) + + prev = dict(models) + + # Done when nothing is still loading + if models and all(s != "loading" for s in models.values()): + # Clear any loading detail + session.broadcast(ServiceEvent( + session_id=session.session_id, + status="loading", + detail="", + )) + return + except Exception: + pass + + await asyncio.sleep(2.0) + + async def _allocate_voice(session: Session) -> None: """Request a managed cf-voice instance from cf-orch. No-op if CF_ORCH_URL is unset.""" # Static override takes precedence — skip cf-orch allocation @@ -207,7 +418,8 @@ async def forward_audio_chunk( buf = _audio_buffers.setdefault(session.session_id, []) buf.append(raw) - if len(buf) < _CHUNKS_PER_WINDOW: + chunks_needed = max(1, session.window_ms // _CHUNK_MS) + if len(buf) < chunks_needed: return # not enough audio yet — wait for more chunks # Flush: concatenate window, reset buffer @@ -221,6 +433,8 @@ async def forward_audio_chunk( "timestamp": timestamp, "elcor": session.elcor, "session_id": session.session_id, + "language": session.transcribe_lang or None, # None = Whisper auto-detect + "num_speakers": session.num_speakers or None, # None = pyannote auto-detect } try: async with httpx.AsyncClient(timeout=5.0) as client: @@ -233,6 +447,7 @@ async def forward_audio_chunk( from app.models.queue_event import QueueEvent from app.models.transcript_event import TranscriptEvent + from app.models.scene_event import SceneEvent, AccentEvent for ev in data.get("events", []): etype = ev.get("event_type") @@ -277,63 +492,89 @@ async def forward_audio_chunk( ) session.broadcast(queue_ev) + elif etype == "scene": + scene_ev = SceneEvent( + session_id=session.session_id, + label=ev["label"], + confidence=ev.get("confidence", 1.0), + timestamp=ev["timestamp"], + privacy_risk=ev.get("privacy_risk", "low"), + ) + session.broadcast(scene_ev) + + elif etype == "accent": + accent_ev = AccentEvent( + session_id=session.session_id, + region=ev["label"], + language=ev.get("language", ""), + confidence=ev.get("confidence", 1.0), + timestamp=ev["timestamp"], + ) + session.broadcast(accent_ev) + # ── Idle session reaper ─────────────────────────────────────────────────────── +def _should_reap(session: Session, now: float) -> bool: + """ + Return True if a session should be reaped. + + Free tier: reap when last subscriber left > SESSION_IDLE_TTL_S ago. + Paid/premium: reap when last activity (broadcast) > SESSION_PAID_TTL_S ago, + regardless of subscriber state. This lets paid sessions survive reconnects + but enforces a hard 30-min activity timeout. + """ + if session.state != "running": + return False + + if session.tier in ("paid", "premium"): + return (now - session.last_activity_at) > settings.session_paid_ttl_s + + # Free tier: only reap when there are no subscribers and the idle clock is running + return ( + session.subscriber_count() == 0 + and session.last_subscriber_left_at is not None + and (now - session.last_subscriber_left_at) > settings.session_idle_ttl_s + ) + + async def _reaper_loop() -> None: """ - Periodically kill sessions with no active SSE subscribers. + Periodically kill sessions that have exceeded their tier TTL. - A session becomes eligible for reaping when: - - Its last SSE subscriber disconnected more than SESSION_IDLE_TTL_S seconds ago - - It has not been explicitly ended (state != "stopped") + Free tier (90s): reaps when last SSE subscriber left > 90s ago. + Covers mobile tab timeout / screen lock / crash. - This covers the common mobile pattern: screen locks → browser suspends tab → - EventSource closes → SSE subscriber count drops to zero. If the user doesn't - return within the TTL window, the session is cleaned up automatically. + Paid tier (30 min): reaps when last annotation event > 30 min ago, + regardless of subscriber state. Allows paid users to reconnect within + the window without losing their session. - The reaper runs every REAP_INTERVAL_S seconds (half the TTL, so the worst-case - overshoot is TTL + REAP_INTERVAL_S). + The reaper runs every REAP_INTERVAL_S seconds (half the free TTL, capped + at 60s, so worst-case overshoot is small). """ ttl = settings.session_idle_ttl_s - interval = max(15, ttl // 2) - logger.info("Session reaper started (TTL=%ds, check every %ds)", ttl, interval) + interval = max(15, min(60, ttl // 2)) + logger.info("Session reaper started (free TTL=%ds, paid TTL=%ds, check every %ds)", + ttl, settings.session_paid_ttl_s, interval) while True: await asyncio.sleep(interval) now = time.monotonic() - to_reap = [ - sid - for sid, session in list(_sessions.items()) - if ( - session.state == "running" - and session.subscriber_count() == 0 - and session.last_subscriber_left_at is not None - and (now - session.last_subscriber_left_at) > ttl - ) - ] + to_reap = [sid for sid, s in list(_sessions.items()) if _should_reap(s, now)] for sid in to_reap: - logger.info( - "Reaping idle session %s (no subscribers for >%ds)", sid, ttl - ) + session = _sessions.get(sid) + tier = session.tier if session else "free" + logger.info("Reaping idle %s session %s", tier, sid) await end_session(sid) async def _reaper_loop_once() -> None: """Single reaper pass — used by tests to avoid sleeping.""" - ttl = settings.session_idle_ttl_s now = time.monotonic() - to_reap = [ - sid - for sid, session in list(_sessions.items()) - if ( - session.state == "running" - and session.subscriber_count() == 0 - and session.last_subscriber_left_at is not None - and (now - session.last_subscriber_left_at) > ttl - ) - ] + to_reap = [sid for sid, s in list(_sessions.items()) if _should_reap(s, now)] for sid in to_reap: - logger.info("Reaping idle session %s (no subscribers for >%ds)", sid, ttl) + session = _sessions.get(sid) + tier = session.tier if session else "free" + logger.info("Reaping idle %s session %s", tier, sid) await end_session(sid) diff --git a/app/services/translator.py b/app/services/translator.py new file mode 100644 index 0000000..7acae02 --- /dev/null +++ b/app/services/translator.py @@ -0,0 +1,113 @@ +# app/services/translator.py — DeepL-backed tone label translation +# +# Translates short tone labels (e.g. "Neutral", "Stressed", "Apologetic") +# into the user's target language via the DeepL API. +# +# Two API key paths: +# CF-managed (Paid tier): DEEPL_API_KEY env var (DeepL Pro endpoint) +# BYOK (any tier): per-session byok_deepl_key from StartRequest (DeepL Free endpoint) +# +# Per-session label cache: each unique label is translated once per session. +# Cache is discarded when the session ends — it lives on the Translator instance, +# which is created per-session and GC'd with the session. +# +# Rate limiting: translation runs only on ToneEvent emit, not on raw audio frames. +# The SSE generator calls translate() once per event; the cache makes repeat labels free. +from __future__ import annotations + +import logging +import time + +import requests + +from app.config import settings + +logger = logging.getLogger(__name__) + +_DEEPL_FREE_URL = "https://api-free.deepl.com/v2/translate" +_DEEPL_PRO_URL = "https://api.deepl.com/v2/translate" + + +class Translator: + """ + Per-session DeepL translator with label cache. + + Usage: + translator = Translator.for_session(session) + translated = translator.translate("Stressed") # "Stressé" for target_lang="fr" + translated = translator.translate("Stressed") # cache hit — no API call + """ + + def __init__( + self, + target_lang: str, + api_key: str, + pro: bool = True, + ) -> None: + self._target_lang = target_lang.upper() # DeepL uses uppercase codes + self._api_key = api_key + self._url = _DEEPL_PRO_URL if pro else _DEEPL_FREE_URL + self._cache: dict[str, str] = {} + + @classmethod + def for_session(cls, session) -> "Translator | None": + """ + Create a Translator for a session, or return None if translation is disabled. + + Returns None when: + - session.target_lang is empty + - No API key is available (neither BYOK nor CF-managed) + """ + target_lang = session.target_lang + if not target_lang: + return None + + byok = session.byok_deepl_key.strip() + cf_key = settings.deepl_api_key.strip() + + if byok: + return cls(target_lang=target_lang, api_key=byok, pro=False) + if cf_key: + return cls(target_lang=target_lang, api_key=cf_key, pro=True) + + logger.warning( + "[translator] target_lang=%s set but no DeepL key available — skipping translation", + target_lang, + ) + return None + + def translate(self, label: str) -> str: + """ + Translate a tone label. Returns the original label on any error. + + Results are cached per-session so the same label is only translated once. + """ + if not label: + return label + + if label in self._cache: + return self._cache[label] + + translated = self._call_deepl(label) + self._cache[label] = translated + return translated + + def _call_deepl(self, text: str) -> str: + try: + resp = requests.post( + self._url, + headers={"Authorization": f"DeepL-Auth-Key {self._api_key}"}, + json={ + "text": [text], + "target_lang": self._target_lang, + }, + timeout=5, + ) + if not resp.ok: + logger.warning("[translator] DeepL returned %s for label %r", resp.status_code, text) + return text + data = resp.json() + return data["translations"][0]["text"] + except Exception as exc: + logger.warning("[translator] DeepL call failed for label %r: %s", text, exc) + return text diff --git a/app/tiers.py b/app/tiers.py index 2302964..3e204ed 100644 --- a/app/tiers.py +++ b/app/tiers.py @@ -1,31 +1,159 @@ # app/tiers.py — tier gate checks # -# Free tier: local inference only, no license key required. -# Paid tier: cloud STT/TTS fallback, session pinning (v1.0). +# Two validation paths: +# Self-hosted: LINNET_LICENSE_KEY env var (or key passed per-request) +# → cf-core validate_license() → Heimdall /v1/licenses/verify +# Cloud mode: user_id from X-CF-Session header +# → Heimdall /admin/cloud/resolve (admin token required) +# +# Both paths cache results in-process for 30 minutes so a brief Heimdall +# outage does not interrupt active sessions. +# +# BYOK: user-supplied DeepL key bypasses the Paid gate for translation only. +# Pass byok_deepl=True to require_paid() to enable this exception. from __future__ import annotations -import os +import logging +import time -BYOK_UNLOCKABLE = ["cloud_stt", "cloud_tts", "session_pinning"] +import requests + +from app.config import settings + +logger = logging.getLogger(__name__) + +_PRODUCT = "LNNT" +_CACHE_TTL = 1800 # 30 minutes — matches Heimdall offline grace window + +# Cache: cache_key (str) -> (tier: str, expires_at: float) +_cache: dict[str, tuple[str, float]] = {} -def is_paid(license_key: str | None = None) -> bool: - """Return True if the request has a valid Paid+ license key.""" - key = license_key or os.environ.get("LINNET_LICENSE_KEY", "") - # Paid keys start with CFG-LNNT- (format: CFG-LNNT-XXXX-XXXX-XXXX) - return bool(key) and key.upper().startswith("CFG-LNNT-") +# ── Internal helpers ────────────────────────────────────────────────────────── + +def _cached(cache_key: str) -> str | None: + entry = _cache.get(cache_key) + if entry and time.monotonic() < entry[1]: + return entry[0] + return None -def require_free() -> None: - """No-op. All users get Free tier features.""" +def _store(cache_key: str, tier: str) -> str: + _cache[cache_key] = (tier, time.monotonic() + _CACHE_TTL) + return tier -def require_paid(license_key: str | None = None) -> None: - """Raise if caller doesn't have a Paid license.""" - if not is_paid(license_key): +def _resolve_by_key(license_key: str) -> str: + """Hit Heimdall /v1/licenses/verify for a raw key. Returns tier string.""" + cache_key = f"key:{license_key}" + cached = _cached(cache_key) + if cached is not None: + return cached + + try: + resp = requests.post( + f"{settings.heimdall_url}/v1/licenses/verify", + json={"key": license_key, "min_tier": "free"}, + timeout=5, + ) + if not resp.ok: + logger.warning("[tiers] Heimdall /verify returned %s", resp.status_code) + return _store(cache_key, "free") + data = resp.json() + if not data.get("valid", False): + return _store(cache_key, "free") + tier = data.get("tier", "free") or "free" + return _store(cache_key, tier) + except Exception as exc: + logger.warning("[tiers] Heimdall key validation failed: %s", exc) + # Do NOT cache on network failure — allow retry after next request + return "free" + + +def _resolve_by_user(user_id: str) -> str: + """Hit Heimdall /admin/cloud/resolve for a cloud user. Returns tier string.""" + if not settings.heimdall_admin_token: + logger.warning("[tiers] HEIMDALL_ADMIN_TOKEN not set — defaulting to free") + return "free" + + cache_key = f"user:{user_id}" + cached = _cached(cache_key) + if cached is not None: + return cached + + try: + resp = requests.post( + f"{settings.heimdall_url}/admin/cloud/resolve", + json={"user_id": user_id, "product": _PRODUCT}, + headers={"Authorization": f"Bearer {settings.heimdall_admin_token}"}, + timeout=5, + ) + if not resp.ok: + logger.warning("[tiers] Heimdall /cloud/resolve returned %s", resp.status_code) + return _store(cache_key, "free") + data = resp.json() + tier = data.get("tier", "free") or "free" + return _store(cache_key, tier) + except Exception as exc: + logger.warning("[tiers] Heimdall cloud resolve failed: %s", exc) + return "free" + + +# ── Public API ──────────────────────────────────────────────────────────────── + +def get_tier( + license_key: str | None = None, + user_id: str | None = None, +) -> str: + """ + Return the active tier for this request context. + + Precedence: + 1. user_id (cloud mode — set by CloudAuthMiddleware on request.state.cf_user) + 2. license_key argument (self-hosted, passed per-request) + 3. LINNET_LICENSE_KEY env var (self-hosted, instance-wide) + 4. "free" (no key configured) + """ + if user_id: + return _resolve_by_user(user_id) + + key = license_key or settings.linnet_license_key + if not key: + return "free" + + return _resolve_by_key(key) + + +def is_paid( + license_key: str | None = None, + user_id: str | None = None, +) -> bool: + """Return True if the resolved tier is paid or premium.""" + return get_tier(license_key=license_key, user_id=user_id) in ("paid", "premium") + + +def require_paid( + license_key: str | None = None, + user_id: str | None = None, + byok_deepl: bool = False, +) -> None: + """ + Raise HTTP 402 if caller doesn't have a Paid+ license. + + byok_deepl=True skips the check — the caller has provided their own + DeepL API key and bypasses the Paid gate for translation only. + """ + if byok_deepl: + return + if not is_paid(license_key=license_key, user_id=user_id): from fastapi import HTTPException raise HTTPException( status_code=402, detail="This feature requires a Linnet Paid license. " "Get one at circuitforge.tech.", ) + + +# ── Backwards-compat shim (old call sites that pass key as positional) ──────── + +BYOK_UNLOCKABLE = ["cloud_stt", "cloud_tts", "session_pinning"] diff --git a/compose.cloud.yml b/compose.cloud.yml index edcc6d6..d153951 100644 --- a/compose.cloud.yml +++ b/compose.cloud.yml @@ -6,11 +6,14 @@ # # Requires in .env: # CLOUD_MODE=true -# LINNET_LICENSE_KEY=CFG-LNNT-... (or per-request key via API) # HEIMDALL_URL=https://license.circuitforge.tech +# HEIMDALL_ADMIN_TOKEN=... (admin token for /admin/cloud/resolve — resolves tier from user_id) # DIRECTUS_JWT_SECRET=... (Caddy injects X-CF-Session from cf_session cookie) # HF_TOKEN=... (needed for real cf-voice inference; omit for mock) # CF_VOICE_MOCK=0 (set to 1 during staged rollout) +# +# Note: LINNET_LICENSE_KEY is for self-hosted installs only. +# Cloud mode resolves tier via HEIMDALL_ADMIN_TOKEN + /admin/cloud/resolve. services: linnet-api: diff --git a/frontend/src/App.vue b/frontend/src/App.vue index 4a215b0..98e1c04 100644 --- a/frontend/src/App.vue +++ b/frontend/src/App.vue @@ -7,18 +7,30 @@
-
-

Now

- -
+ + + + +
@@ -26,11 +38,15 @@ diff --git a/frontend/src/components/DevPanel.vue b/frontend/src/components/DevPanel.vue new file mode 100644 index 0000000..195e388 --- /dev/null +++ b/frontend/src/components/DevPanel.vue @@ -0,0 +1,137 @@ + + + + + + + + + diff --git a/frontend/src/components/HistoryStrip.vue b/frontend/src/components/HistoryStrip.vue index 1cf88d1..b11a9f4 100644 --- a/frontend/src/components/HistoryStrip.vue +++ b/frontend/src/components/HistoryStrip.vue @@ -5,8 +5,9 @@ :key="i" class="history-item" :data-affect="evt.affect" - :title="`${evt.label} (${(evt.confidence * 100).toFixed(0)}%)`" + :title="`${speakerShort(evt.speaker_id)} · ${evt.label} (${(evt.confidence * 100).toFixed(0)}%)`" > + {{ speakerShort(evt.speaker_id) }} {{ evt.label }} {{ (evt.confidence * 100).toFixed(0) }}% @@ -25,6 +26,22 @@ const DISPLAY_COUNT = 8; const store = useSessionStore(); // Most recent N events, newest last (scroll right) const recent = computed(() => store.events.slice(-DISPLAY_COUNT)); + +// Returns single-letter label for a speaker_id, or empty for silence/unknown +function speakerShort(id: string | undefined): string { + if (!id || id === "speaker_a") return ""; + if (id === "Multiple") return "M"; + const match = id.match(/Speaker\s+([A-Z]+)/i); + return match ? match[1] : ""; +} + +// CSS slot key for per-speaker colour +function speakerSlot(id: string | undefined): string { + if (!id || id === "speaker_a") return ""; + if (id === "Multiple") return "multi"; + const match = id.match(/Speaker\s+([A-Z]+)/i); + return match ? match[1].toLowerCase() : "a"; +} @@ -156,10 +215,12 @@ export default { name: "NowPanel" }; min-height: 1.4rem; } -/* Shared pill base for speaker / queue / environ badges */ +/* Shared pill base for speaker / queue / environ / scene / privacy badges */ .now-speaker, .now-queue-badge, -.now-environ-badge { +.now-environ-badge, +.now-scene-badge, +.now-privacy-badge { display: inline-block; font-size: 0.65rem; font-family: var(--font-mono, monospace); @@ -173,6 +234,24 @@ export default { name: "NowPanel" }; transition: background 0.3s, color 0.3s; } +/* Diarization identity badge — per-speaker colours */ +.now-diar-id { + display: inline-block; + font-size: 0.65rem; + font-family: var(--font-mono, monospace); + font-weight: 700; + letter-spacing: 0.06em; + padding: 0.15em 0.6em; + border-radius: 999px; + border: 1px solid transparent; + transition: background 0.3s, color 0.3s; +} +.now-diar-id[data-diar="a"] { background: #1e3050; color: #60a5fa; border-color: #60a5fa44; } +.now-diar-id[data-diar="b"] { background: #2a1e3a; color: #c084fc; border-color: #c084fc44; } +.now-diar-id[data-diar="c"] { background: #1e3a2f; color: #34d399; border-color: #34d39944; } +.now-diar-id[data-diar="d"] { background: #2a2a1e; color: #fbbf24; border-color: #fbbf2444; } +.now-diar-id[data-diar="multi"] { background: #2a1e1e; color: #f87171; border-color: #f8717144; } + /* Speaker type colours */ .now-speaker[data-speaker="human_single"] { background: #1e3a2f; color: #34d399; border-color: #34d39933; } .now-speaker[data-speaker="human_multi"] { background: #1e3a2f; color: #34d399; border-color: #34d39933; } @@ -186,11 +265,37 @@ export default { name: "NowPanel" }; .now-queue-badge[data-queue="dtmf_tone"] { background: #1e2a3a; color: #60a5fa; border-color: #60a5fa33; } .now-queue-badge[data-queue="dead_air"] { background: #1e1e1e; color: #4b5563; border-color: #4b556333; } -/* Environment colours */ +/* Environment colours — telephony */ .now-environ-badge[data-environ="call_center"] { background: #1e2a3a; color: #60a5fa; border-color: #60a5fa22; } .now-environ-badge[data-environ="music"] { background: #2a1e3a; color: #a78bfa; border-color: #a78bfa22; } .now-environ-badge[data-environ="background_shift"] { background: #2a2a1e; color: #fbbf24; border-color: #fbbf2422; } .now-environ-badge[data-environ="noise_floor_change"] { background: #2a1e1e; color: #f87171; border-color: #f8717122; } +/* Environment colours — nature */ +.now-environ-badge[data-environ="birdsong"] { background: #1e2e1e; color: #86efac; border-color: #86efac22; } +.now-environ-badge[data-environ="wind"] { background: #1e2a2e; color: #7dd3fc; border-color: #7dd3fc22; } +.now-environ-badge[data-environ="rain"] { background: #1e222e; color: #93c5fd; border-color: #93c5fd22; } +.now-environ-badge[data-environ="water"] { background: #1e2a2e; color: #67e8f9; border-color: #67e8f922; } +/* Environment colours — urban */ +.now-environ-badge[data-environ="traffic"] { background: #2a201e; color: #fdba74; border-color: #fdba7422; } +.now-environ-badge[data-environ="crowd_chatter"]{ background: #2a1e28; color: #d8b4fe; border-color: #d8b4fe22; } +.now-environ-badge[data-environ="construction"] { background: #2a201e; color: #fb923c; border-color: #fb923c22; } +/* Environment colours — indoor */ +.now-environ-badge[data-environ="hvac"] { background: #1e2226; color: #94a3b8; border-color: #94a3b822; } +.now-environ-badge[data-environ="keyboard_typing"]{ background: #1e2226; color: #94a3b8; border-color: #94a3b822; } +.now-environ-badge[data-environ="restaurant"] { background: #2a1e22; color: #fca5a5; border-color: #fca5a522; } + +/* Acoustic scene badges — informational, cool tones */ +.now-scene-badge[data-scene="indoor_quiet"] { background: #1e2226; color: #94a3b8; border-color: #94a3b822; } +.now-scene-badge[data-scene="indoor_crowd"] { background: #2a1e28; color: #c084fc; border-color: #c084fc22; } +.now-scene-badge[data-scene="outdoor_urban"] { background: #2a201e; color: #fdba74; border-color: #fdba7422; } +.now-scene-badge[data-scene="outdoor_nature"] { background: #1e2e1e; color: #86efac; border-color: #86efac22; } +.now-scene-badge[data-scene="vehicle"] { background: #1e2226; color: #7dd3fc; border-color: #7dd3fc22; } +.now-scene-badge[data-scene="public_transit"] { background: #1e2226; color: #7dd3fc; border-color: #7dd3fc22; } + +/* Privacy risk indicator — calm, never alarming. Teal for both levels. */ +.now-privacy-badge { cursor: default; } +.now-privacy-badge[data-privacy="moderate"]{ background: #1e2a2e; color: #67e8f9; border-color: #67e8f933; } +.now-privacy-badge[data-privacy="high"] { background: #1e2a2e; color: #22d3ee; border-color: #22d3ee44; } /* Live transcript strip */ .now-transcript { diff --git a/frontend/src/composables/useToneStream.ts b/frontend/src/composables/useToneStream.ts index 3c8cfa6..cbef353 100644 --- a/frontend/src/composables/useToneStream.ts +++ b/frontend/src/composables/useToneStream.ts @@ -11,9 +11,12 @@ import { ref, onUnmounted } from "vue"; import { useSessionStore, + type ServiceEvent, type SpeakerEvent, type TranscriptEvent, type QueueEvent, + type SceneEvent, + type AccentEvent, } from "../stores/session"; import { useWakeLock } from "./useWakeLock"; @@ -45,6 +48,15 @@ export function useToneStream() { source.addEventListener("environ-event", (e: MessageEvent) => { try { store.updateQueue(JSON.parse(e.data) as QueueEvent); } catch { /* */ } }); + source.addEventListener("scene-event", (e: MessageEvent) => { + try { store.updateScene(JSON.parse(e.data) as SceneEvent); } catch { /* */ } + }); + source.addEventListener("accent-event", (e: MessageEvent) => { + try { store.updateAccent(JSON.parse(e.data) as AccentEvent); } catch { /* */ } + }); + source.addEventListener("service-event", (e: MessageEvent) => { + try { store.updateService(JSON.parse(e.data) as ServiceEvent); } catch { /* */ } + }); source.onopen = () => { connected.value = true; }; source.onerror = () => { connected.value = false; }; diff --git a/frontend/src/stores/session.ts b/frontend/src/stores/session.ts index af3d34a..530f170 100644 --- a/frontend/src/stores/session.ts +++ b/frontend/src/stores/session.ts @@ -13,6 +13,22 @@ export interface ToneEvent { affect: string; shift_direction: string; prosody_flags: string[]; + // Dimensional emotion (audeering model — Navigation v0.2.x) + valence: number | null; + arousal: number | null; + dominance: number | null; + // Prosodic signals (openSMILE — Navigation v0.2.x) + sarcasm_risk: number | null; + flat_f0_score: number | null; + // Trajectory signals — null until baseline established (min 3 frames per speaker) + arousal_delta: number | null; + valence_delta: number | null; + trend: string | null; + // Coherence signals (SER vs VAD cross-comparison) + coherence_score: number | null; + suppression_flag: boolean | null; + reframe_type: string | null; + affect_divergence: number | null; } export interface SpeakerEvent { @@ -39,6 +55,31 @@ export interface QueueEvent { timestamp: number; } +export interface ServiceEvent { + event_type: "service"; + session_id: string; + status: "ready" | "loading" | "error"; + detail: string; +} + +export interface SceneEvent { + event_type: "scene"; + session_id: string; + label: string; + confidence: number; + timestamp: number; + privacy_risk: "low" | "moderate" | "high"; +} + +export interface AccentEvent { + event_type: "accent"; + session_id: string; + region: string; + language: string; + confidence: number; + timestamp: number; +} + export const useSessionStore = defineStore("session", () => { const sessionId = ref(null); const elcor = ref(false); @@ -49,14 +90,35 @@ export const useSessionStore = defineStore("session", () => { const currentTranscript = ref(null); const currentQueue = ref(null); const currentEnviron = ref(null); + const currentScene = ref(null); + const currentAccent = ref(null); + + // voiceReady: false until the backend probes cf-voice and emits service-event status="ready". + // voiceError: non-null when the probe times out — surfaced as a warning in the UI. + // voiceWarning: non-null for soft/non-fatal warnings (e.g. diarizer misconfigured). + // voiceLoadingDetail: non-null while models are downloading (cleared when all stable). + const voiceReady = ref(false); + const voiceError = ref(null); + const voiceWarning = ref(null); + const voiceLoadingDetail = ref(null); const apiBase = import.meta.env.VITE_API_BASE ?? ""; - async function startSession(withElcor = false) { + async function startSession(opts: { + elcor?: boolean; + windowMs?: number; + transcribeLang?: string; + numSpeakers?: number; + } = {}) { const resp = await fetch(`${apiBase}/session/start`, { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify({ elcor: withElcor }), + body: JSON.stringify({ + elcor: opts.elcor ?? false, + window_ms: opts.windowMs ?? 1000, + transcribe_lang: opts.transcribeLang ?? "", + num_speakers: opts.numSpeakers ?? 0, + }), }); if (!resp.ok) throw new Error(`Failed to start session: ${resp.status}`); const data = await resp.json(); @@ -64,6 +126,8 @@ export const useSessionStore = defineStore("session", () => { elcor.value = data.elcor; state.value = "running"; events.value = []; + voiceReady.value = false; + voiceError.value = null; } async function endSession() { @@ -86,11 +150,45 @@ export const useSessionStore = defineStore("session", () => { currentTranscript.value = evt; } + function updateScene(evt: SceneEvent) { + currentScene.value = evt; + } + + function updateAccent(evt: AccentEvent) { + currentAccent.value = evt; + } + + // TTL timers — clear environ/queue badges when the signal drops out. + // AST classifier emits nothing when confidence falls below threshold, so without + // a TTL the badge stays stuck indefinitely (e.g. music badge after music stops). + let _environTimer: ReturnType | null = null; + let _queueTimer: ReturnType | null = null; + const ENVIRON_TTL_MS = 5000; + const QUEUE_TTL_MS = 4000; + function updateQueue(evt: QueueEvent) { if (evt.event_type === "queue") { currentQueue.value = evt; + if (_queueTimer) clearTimeout(_queueTimer); + _queueTimer = setTimeout(() => { currentQueue.value = null; }, QUEUE_TTL_MS); } else { currentEnviron.value = evt; + if (_environTimer) clearTimeout(_environTimer); + _environTimer = setTimeout(() => { currentEnviron.value = null; }, ENVIRON_TTL_MS); + } + } + + function updateService(evt: ServiceEvent) { + if (evt.status === "ready") { + voiceReady.value = true; + voiceError.value = null; + } else if (evt.status === "error") { + voiceError.value = evt.detail || "Voice service failed to start."; + } else if (evt.status === "warning") { + voiceWarning.value = evt.detail || null; + } else if (evt.status === "loading") { + // Empty detail = all models done loading; clear the indicator + voiceLoadingDetail.value = evt.detail || null; } } @@ -103,13 +201,22 @@ export const useSessionStore = defineStore("session", () => { currentTranscript.value = null; currentQueue.value = null; currentEnviron.value = null; + currentScene.value = null; + currentAccent.value = null; + voiceReady.value = false; + voiceError.value = null; + voiceWarning.value = null; + voiceLoadingDetail.value = null; } return { sessionId, elcor, state, events, latest, currentSpeaker, currentTranscript, currentQueue, currentEnviron, + currentScene, currentAccent, + voiceReady, voiceError, voiceWarning, voiceLoadingDetail, startSession, endSession, pushEvent, - updateSpeaker, updateTranscript, updateQueue, + updateSpeaker, updateTranscript, updateQueue, updateService, + updateScene, updateAccent, reset, }; }); diff --git a/frontend/src/stores/settings.ts b/frontend/src/stores/settings.ts new file mode 100644 index 0000000..9c80942 --- /dev/null +++ b/frontend/src/stores/settings.ts @@ -0,0 +1,72 @@ +// stores/settings.ts — persistent user preferences, survives page reload via localStorage. +// +// All settings here apply to the NEXT session start. Mid-session changes take +// effect on the following session (the ComposeBar shows current-session values +// as read-only badges while a session is running). + +import { defineStore } from "pinia"; +import { ref, watch } from "vue"; + +const STORAGE_KEY = "linnet:settings:v1"; + +interface PersistedSettings { + windowMs: number; + transcribeLang: string; + elcor: boolean; + numSpeakers: number; // 0 = auto-detect + threadView: boolean; // dev-only: preview v0.2.x thread UI (toggled from DevPanel) +} + +const DEFAULTS: PersistedSettings = { + windowMs: 1000, + transcribeLang: "", + elcor: false, + numSpeakers: 0, + threadView: false, +}; + +function loadFromStorage(): PersistedSettings { + try { + const raw = localStorage.getItem(STORAGE_KEY); + if (raw) { + // Merge with defaults so new fields added after initial save get their defaults + return { ...DEFAULTS, ...(JSON.parse(raw) as Partial) }; + } + } catch { + // corrupt or missing — fall through to defaults + } + return { ...DEFAULTS }; +} + +export const useSettingsStore = defineStore("settings", () => { + const saved = loadFromStorage(); + + const windowMs = ref(saved.windowMs); + const transcribeLang = ref(saved.transcribeLang); + const elcor = ref(saved.elcor); + // 0 = auto; 1–8 = fixed count passed to pyannote as min_speakers=max_speakers + const numSpeakers = ref(saved.numSpeakers); + // Dev-only: show v0.2.x thread UI instead of v0.1.x NowPanel/HistoryStrip. + // Toggled from DevPanel; not exposed in user-facing settings. + const threadView = ref(saved.threadView); + + // Persist to localStorage whenever any setting changes + watch( + [windowMs, transcribeLang, elcor, numSpeakers, threadView], + () => { + localStorage.setItem( + STORAGE_KEY, + JSON.stringify({ + windowMs: windowMs.value, + transcribeLang: transcribeLang.value, + elcor: elcor.value, + numSpeakers: numSpeakers.value, + threadView: threadView.value, + }), + ); + }, + { deep: false }, + ); + + return { windowMs, transcribeLang, elcor, numSpeakers, threadView }; +}); diff --git a/frontend/vite.config.ts b/frontend/vite.config.ts index c8e2159..bae0b0e 100644 --- a/frontend/vite.config.ts +++ b/frontend/vite.config.ts @@ -8,6 +8,7 @@ export default defineConfig({ server: { host: "0.0.0.0", port: 8521, + allowedHosts: ["menagerie.circuitforge.tech"], proxy: { "/session": { target: "http://localhost:8522", @@ -22,6 +23,10 @@ export default defineConfig({ target: "http://localhost:8522", changeOrigin: true, }, + "/dev": { + target: "http://localhost:8522", + changeOrigin: true, + }, }, }, }); diff --git a/manage.sh b/manage.sh index cd3beda..d42d007 100755 --- a/manage.sh +++ b/manage.sh @@ -2,6 +2,12 @@ # manage.sh — Linnet dev/demo/cloud management script set -euo pipefail +# Source .env early so LINNET_PORT, LINNET_FRONTEND_PORT, VITE_*, and all +# other vars are available before any assignments below read them. +if [[ -f .env ]]; then + set -a; source .env; set +a +fi + CMD=${1:-help} PROFILE=${2:-dev} # dev | demo | cloud API_PORT=${LINNET_PORT:-8522} @@ -36,9 +42,14 @@ case "$CMD" in start) _check_env + # Re-source after _check_env in case .env was just created from .env.example + set -a; source .env; set +a + # Recompute ports in case .env was just created with different values + API_PORT=${LINNET_PORT:-8522} + FE_PORT=${LINNET_FRONTEND_PORT:-8521} if [[ "$PROFILE" == "dev" ]]; then - info "Starting Linnet API on :${API_PORT} (mock mode, dev)…" - CF_VOICE_MOCK=1 conda run -n cf uvicorn app.main:app \ + info "Starting Linnet API on :${API_PORT}…" + conda run -n cf uvicorn app.main:app \ --host 0.0.0.0 --port "$API_PORT" --reload & info "Starting Linnet frontend on :${FE_PORT}…" cd frontend && npm install --silent && npm run dev & diff --git a/tests/test_pinning.py b/tests/test_pinning.py new file mode 100644 index 0000000..87d68bf --- /dev/null +++ b/tests/test_pinning.py @@ -0,0 +1,214 @@ +# tests/test_pinning.py — session pinning (Paid tier TTL) and translator +from __future__ import annotations + +import time +from unittest.mock import MagicMock, patch + +import pytest + +from app.models.session import Session +from app.services import session_store +from app.services.translator import Translator + + +@pytest.fixture(autouse=True) +def clean_sessions(): + session_store._sessions.clear() + session_store._tasks.clear() + session_store._audio_buffers.clear() + yield + session_store._sessions.clear() + session_store._tasks.clear() + session_store._audio_buffers.clear() + + +# ── Tier-aware reaper (_should_reap) ───────────────────────────────────────── + +def test_free_session_reaped_after_subscriber_leaves(monkeypatch): + monkeypatch.setattr("app.services.session_store.settings.session_idle_ttl_s", 0) + s = Session(tier="free") + s.state = "running" + q = s.subscribe() + s.unsubscribe(q) # triggers last_subscriber_left_at + now = time.monotonic() + 1 + assert session_store._should_reap(s, now) is True + + +def test_free_session_spared_while_subscriber_active(monkeypatch): + monkeypatch.setattr("app.services.session_store.settings.session_idle_ttl_s", 0) + s = Session(tier="free") + s.state = "running" + s.subscribe() + now = time.monotonic() + 9999 + assert session_store._should_reap(s, now) is False + + +def test_paid_session_reaped_on_activity_timeout(monkeypatch): + monkeypatch.setattr("app.services.session_store.settings.session_paid_ttl_s", 0) + s = Session(tier="paid") + s.state = "running" + now = time.monotonic() + 1 + assert session_store._should_reap(s, now) is True + + +def test_paid_session_spared_within_activity_window(monkeypatch): + monkeypatch.setattr("app.services.session_store.settings.session_paid_ttl_s", 9999) + s = Session(tier="paid") + s.state = "running" + now = time.monotonic() + 1 + assert session_store._should_reap(s, now) is False + + +def test_paid_session_not_reaped_despite_no_subscribers(monkeypatch): + """Paid tier ignores subscriber state — activity window is what matters.""" + monkeypatch.setattr("app.services.session_store.settings.session_paid_ttl_s", 9999) + s = Session(tier="paid") + s.state = "running" + q = s.subscribe() + s.unsubscribe(q) # no subscribers now + now = time.monotonic() + 200 # past free-tier TTL but within paid TTL + assert session_store._should_reap(s, now) is False + + +def test_stopped_session_never_reaped(): + s = Session(tier="free") + s.state = "stopped" + assert session_store._should_reap(s, time.monotonic() + 9999) is False + + +def test_broadcast_updates_last_activity_at(): + from app.models.tone_event import ToneEvent + s = Session(tier="paid") + before = s.last_activity_at + time.sleep(0.01) + tone = ToneEvent( + session_id=s.session_id, label="Neutral", confidence=0.9, + speaker_id="speaker_a", shift_magnitude=0.0, timestamp=0.0, + subtext="", affect="neutral", + ) + s.broadcast(tone) + assert s.last_activity_at > before + + +# ── Snapshot save on end ────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_snapshot_saved_for_paid_session_with_history(): + from app.models.tone_event import ToneEvent + session = await session_store.create_session(tier="paid", user_id="u1") + tone = ToneEvent( + session_id=session.session_id, label="Stressed", confidence=0.85, + speaker_id="Speaker A", shift_magnitude=0.1, timestamp=1.0, + subtext="", affect="stressed", + ) + session.history.append(tone) + + with patch("app.services.session_store._save_snapshot") as mock_save: + await session_store.end_session(session.session_id) + mock_save.assert_called_once_with(session) + + +@pytest.mark.asyncio +async def test_snapshot_not_saved_for_free_session(): + session = await session_store.create_session(tier="free", user_id="u1") + from app.models.tone_event import ToneEvent + tone = ToneEvent( + session_id=session.session_id, label="Neutral", confidence=0.9, + speaker_id="speaker_a", shift_magnitude=0.0, timestamp=0.0, + subtext="", affect="neutral", + ) + session.history.append(tone) + + with patch("app.services.session_store._save_snapshot") as mock_save: + await session_store.end_session(session.session_id) + mock_save.assert_not_called() + + +@pytest.mark.asyncio +async def test_snapshot_not_saved_for_empty_paid_session(): + session = await session_store.create_session(tier="paid", user_id="u1") + # No events in history + with patch("app.services.session_store._save_snapshot") as mock_save: + await session_store.end_session(session.session_id) + mock_save.assert_not_called() + + +# ── Translator ──────────────────────────────────────────────────────────────── + +def _make_session(target_lang="", byok="", tier="free"): + s = Session(tier=tier, target_lang=target_lang, byok_deepl_key=byok) + return s + + +def _mock_deepl_resp(translated: str): + resp = MagicMock() + resp.ok = True + resp.json.return_value = {"translations": [{"text": translated}]} + return resp + + +def test_translator_none_when_no_target_lang(): + s = _make_session(target_lang="") + assert Translator.for_session(s) is None + + +def test_translator_none_when_no_key(monkeypatch): + monkeypatch.setattr("app.services.translator.settings.deepl_api_key", "") + s = _make_session(target_lang="fr") + assert Translator.for_session(s) is None + + +def test_translator_uses_byok_key(monkeypatch): + monkeypatch.setattr("app.services.translator.settings.deepl_api_key", "") + s = _make_session(target_lang="fr", byok="my-deepl-key") + t = Translator.for_session(s) + assert t is not None + assert t._url.endswith("api-free.deepl.com/v2/translate") + + +def test_translator_uses_cf_key(monkeypatch): + monkeypatch.setattr("app.services.translator.settings.deepl_api_key", "cf-pro-key") + s = _make_session(target_lang="es") + t = Translator.for_session(s) + assert t is not None + assert t._url.endswith("api.deepl.com/v2/translate") + + +def test_translator_byok_takes_precedence_over_cf_key(monkeypatch): + monkeypatch.setattr("app.services.translator.settings.deepl_api_key", "cf-pro-key") + s = _make_session(target_lang="de", byok="user-free-key") + t = Translator.for_session(s) + assert t._url.endswith("api-free.deepl.com/v2/translate") + assert t._api_key == "user-free-key" + + +def test_translate_calls_deepl(): + t = Translator(target_lang="fr", api_key="k", pro=False) + with patch("app.services.translator.requests.post", return_value=_mock_deepl_resp("Stressé")): + result = t.translate("Stressed") + assert result == "Stressé" + + +def test_translate_caches_result(): + t = Translator(target_lang="fr", api_key="k", pro=False) + with patch("app.services.translator.requests.post", return_value=_mock_deepl_resp("Stressé")) as mock_post: + t.translate("Stressed") + t.translate("Stressed") + assert mock_post.call_count == 1 + + +def test_translate_returns_original_on_error(): + t = Translator(target_lang="fr", api_key="k", pro=False) + with patch("app.services.translator.requests.post", side_effect=Exception("timeout")): + result = t.translate("Stressed") + assert result == "Stressed" + + +def test_translate_target_lang_uppercased(): + t = Translator(target_lang="fr", api_key="k", pro=False) + assert t._target_lang == "FR" + + +def test_translate_empty_label_returns_empty(): + t = Translator(target_lang="fr", api_key="k", pro=False) + assert t.translate("") == "" diff --git a/tests/test_tiers.py b/tests/test_tiers.py index 5ed74cf..cb8730a 100644 --- a/tests/test_tiers.py +++ b/tests/test_tiers.py @@ -1,34 +1,156 @@ -# tests/test_tiers.py +# tests/test_tiers.py — tier gate and Heimdall validation logic from __future__ import annotations +from unittest.mock import MagicMock, patch + import pytest from fastapi import HTTPException -from app.tiers import is_paid, require_paid +from app import tiers -def test_valid_linnet_key(): - assert is_paid("CFG-LNNT-AAAA-BBBB-CCCC") is True +@pytest.fixture(autouse=True) +def clear_tier_cache(): + """Wipe the in-process tier cache between tests.""" + tiers._cache.clear() + yield + tiers._cache.clear() -def test_invalid_prefix(): - assert is_paid("CFG-PRNG-AAAA-BBBB-CCCC") is False +# ── Key-based resolution ────────────────────────────────────────────────────── + +def _mock_verify(tier: str, valid: bool = True): + """Return a mock requests.Response for /v1/licenses/verify.""" + resp = MagicMock() + resp.ok = valid + resp.json.return_value = {"valid": valid, "tier": tier, "user_id": "u123"} + return resp -def test_empty_key(): - assert is_paid("") is False +def test_get_tier_free_when_no_key(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + assert tiers.get_tier() == "free" -def test_none_key(): - assert is_paid(None) is False # type: ignore[arg-type] +def test_get_tier_paid_via_key(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + with patch("app.tiers.requests.post", return_value=_mock_verify("paid")) as mock_post: + result = tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + assert result == "paid" + mock_post.assert_called_once() + assert "/v1/licenses/verify" in mock_post.call_args[0][0] -def test_require_paid_raises_402(): - with pytest.raises(HTTPException) as exc_info: - require_paid(license_key="bad-key") +def test_get_tier_free_on_invalid_key(monkeypatch): + with patch("app.tiers.requests.post", return_value=_mock_verify("free", valid=False)): + result = tiers.get_tier(license_key="CFG-LNNT-DEAD-BEEF-0000") + assert result == "free" + + +def test_get_tier_free_on_heimdall_error(monkeypatch): + with patch("app.tiers.requests.post", side_effect=Exception("timeout")): + result = tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + assert result == "free" + + +def test_get_tier_uses_env_key(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "CFG-LNNT-ENV0-ENV0-ENV0") + with patch("app.tiers.requests.post", return_value=_mock_verify("paid")) as mock_post: + result = tiers.get_tier() + assert result == "paid" + assert mock_post.call_args[1]["json"]["key"] == "CFG-LNNT-ENV0-ENV0-ENV0" + + +def test_key_result_is_cached(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + with patch("app.tiers.requests.post", return_value=_mock_verify("paid")) as mock_post: + tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + # Second call must hit cache, not Heimdall + assert mock_post.call_count == 1 + + +def test_network_error_does_not_cache(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + with patch("app.tiers.requests.post", side_effect=Exception("timeout")) as mock_post: + tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC") + # Network errors are NOT cached — both calls hit Heimdall + assert mock_post.call_count == 2 + + +# ── Cloud user resolution ───────────────────────────────────────────────────── + +def _mock_cloud_resolve(tier: str): + resp = MagicMock() + resp.ok = True + resp.json.return_value = {"tier": tier, "user_id": "u456"} + return resp + + +def test_get_tier_cloud_paid(monkeypatch): + monkeypatch.setattr("app.tiers.settings.heimdall_admin_token", "tok-admin") + with patch("app.tiers.requests.post", return_value=_mock_cloud_resolve("paid")) as mock_post: + result = tiers.get_tier(user_id="user-uuid-123") + assert result == "paid" + assert "/admin/cloud/resolve" in mock_post.call_args[0][0] + + +def test_get_tier_cloud_free_when_no_admin_token(monkeypatch): + monkeypatch.setattr("app.tiers.settings.heimdall_admin_token", "") + result = tiers.get_tier(user_id="user-uuid-123") + assert result == "free" + + +def test_user_id_takes_precedence_over_key(monkeypatch): + monkeypatch.setattr("app.tiers.settings.heimdall_admin_token", "tok-admin") + with patch("app.tiers.requests.post", return_value=_mock_cloud_resolve("premium")) as mock_post: + result = tiers.get_tier(license_key="CFG-LNNT-AAAA-BBBB-CCCC", user_id="u999") + assert result == "premium" + assert "/admin/cloud/resolve" in mock_post.call_args[0][0] + + +def test_cloud_result_is_cached(monkeypatch): + monkeypatch.setattr("app.tiers.settings.heimdall_admin_token", "tok-admin") + with patch("app.tiers.requests.post", return_value=_mock_cloud_resolve("paid")) as mock_post: + tiers.get_tier(user_id="u-cached") + tiers.get_tier(user_id="u-cached") + assert mock_post.call_count == 1 + + +# ── is_paid() and require_paid() ───────────────────────────────────────────── + +def test_is_paid_true_for_paid_tier(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + with patch("app.tiers.requests.post", return_value=_mock_verify("paid")): + assert tiers.is_paid(license_key="CFG-LNNT-AAAA-BBBB-CCCC") is True + + +def test_is_paid_true_for_premium_tier(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + with patch("app.tiers.requests.post", return_value=_mock_verify("premium")): + assert tiers.is_paid(license_key="CFG-LNNT-PREM-PREM-PREM") is True + + +def test_is_paid_false_for_free(monkeypatch): + monkeypatch.setattr("app.tiers.settings.linnet_license_key", "") + assert tiers.is_paid() is False + + +def test_require_paid_raises_402_for_free(): + with patch("app.tiers.get_tier", return_value="free"): + with pytest.raises(HTTPException) as exc_info: + tiers.require_paid() assert exc_info.value.status_code == 402 -def test_require_paid_passes(): - # Should not raise - require_paid(license_key="CFG-LNNT-AAAA-BBBB-CCCC") +def test_require_paid_passes_for_paid(): + with patch("app.tiers.get_tier", return_value="paid"): + tiers.require_paid() # must not raise + + +def test_require_paid_byok_skips_check(): + # byok_deepl=True bypasses the tier gate regardless of tier + with patch("app.tiers.get_tier", return_value="free") as mock_get: + tiers.require_paid(byok_deepl=True) # must not raise + mock_get.assert_not_called()