Deploy backend v2 2026-05-23 23:36

Dockerfile

@@ -1,45 +1,20 @@
 FROM python:3.11-slim
 
-# ── Install uv ──────────────────────────────────────────────────────
-COPY --from=ghcr.io/astral-sh/uv:latest /uv /uvx /usr/local/bin/
-
-# ── System dependencies ─────────────────────────────────────────────
-RUN apt-get update && apt-get install -y --no-install-recommends \
-    libsndfile1 ffmpeg && \
-    rm -rf /var/lib/apt/lists/*
-
-# ── Create non-root user (HF Spaces runs as UID 1000) ───────────────
 RUN useradd -m -u 1000 user
-ENV HOME=/home/user
-ENV PATH=/home/user/.local/bin:$PATH
+ENV HOME=/home/user \
+    PATH=/home/user/.local/bin:$PATH \
+    PIP_NO_CACHE_DIR=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1
 
-# ── Working directory ────────────────────────────────────────────────
 WORKDIR $HOME/app
 
-# ── Install Python deps ─────────────────────────────────────────────
-COPY --chown=user pyproject.toml uv.lock* ./
-
-RUN uv sync --frozen --no-dev --extra parakeet
+COPY --chown=user server/requirements.txt ./
+RUN pip install --no-cache-dir -r requirements.txt
 
-# ── Copy application code ───────────────────────────────────────────
-COPY --chown=user server/app/ ./app/
+COPY --chown=user server/app ./app
 
-# ── Switch to non-root user ─────────────────────────────────────────
 USER user
 
-# ── Environment ─────────────────────────────────────────────────────
-ENV PATH="$HOME/app/.venv/bin:$PATH"
-
-# Redirect caches to /tmp (writable in HF Spaces)
-ENV HF_HOME=/tmp/hf_cache
-ENV NUMBA_CACHE_DIR=/tmp/numba_cache
-
-# Defaults for HF environment
-ENV ASR_MODEL_TYPE=parakeet
-ENV WHISPER_MODEL=base.en
-ENV WHISPER_DEVICE=cpu
-ENV WHISPER_COMPUTE_TYPE=int8
-
 EXPOSE 7860
 
 CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "7860"]

README.md

@@ -1,9 +1,21 @@
 ---
-title: Meeting AI Assistant
+title: Meeting AI Backend
 sdk: docker
 app_port: 7860
 ---
 
-# Meeting AI Assistant
+# Meeting AI Backend
 
-Real-time meeting transcription and AI-powered response suggestions.
+Stateless Groq-backed suggestion service for the Meeting AI Chrome extension.
+
+Audio transcription happens client-side via the user's own Deepgram key — this
+backend only generates structured AI suggestions from transcript text.
+
+Endpoints:
+- `GET /` — landing
+- `GET /health` — liveness
+- `POST /suggestion` — generates a `Suggestion` (requires `X-Api-Key`)
+
+Required secrets in Space Settings:
+- `GROQ_API_KEY`
+- `API_KEYS` (comma-separated)

pyproject.toml

@@ -1,32 +0,0 @@
-[project]
-name = "meeting-ai-assistant"
-version = "0.2.0"
-requires-python = ">=3.11, <3.12"
-dependencies = [
-    "fastapi>=0.111.0",
-    "uvicorn[standard]>=0.29.0",
-    "pydantic-settings>=2.2.0",
-    "python-dotenv>=1.0.0",
-    "websockets>=12.0",
-    "numpy>=1.24.0",
-    "faster-whisper>=1.0.0",
-    "groq>=0.9.0",
-]
-
-[project.optional-dependencies]
-parakeet = [
-    "nemo_toolkit[asr]>=1.23.0",
-    "soundfile>=0.12.0",
-]
-
-[dependency-groups]
-dev = [
-    "httpx>=0.27.0",
-    "pytest>=8.0.0",
-    "pytest-asyncio>=0.23.0",
-    "anyio>=4.0.0",
-]
-
-[tool.pytest.ini_options]
-pythonpath = ["server"]
-asyncio_mode = "auto"

server/app/config.py

@@ -4,27 +4,34 @@ from pathlib import Path
 
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
-# Resolve .env relative to this file so it works regardless of CWD.
-_SERVER_DIR = Path(__file__).parent.parent   # server/
-_ROOT_DIR = _SERVER_DIR.parent               # project root
+_SERVER_DIR = Path(__file__).parent.parent
+_ROOT_DIR = _SERVER_DIR.parent
 
 
 class Settings(BaseSettings):
-    backend_ws_port: int = 8000
-    # Stored as a raw comma-separated string to avoid pydantic-settings JSON
-    # parsing for list[str] fields, which fails on plain "key1,key2" values.
+    # HTTP server
+    backend_ws_port: int = 7860
+
+    # Auth — comma-separated app API keys. Acts as a rate-limit token
+    # (extension embeds one), not a security boundary; see ADR 0004.
     api_keys: str = ""
     admin_key: str = ""
+
+    # Groq
     groq_api_key: str = ""
-    whisper_model: str = "base.en"   # valid: tiny.en | base.en | small.en | medium.en
-    whisper_device: str = "cpu"
-    whisper_compute_type: str = "float32"
-    asr_model_type: str = "whisper"
-    vad_threshold: float = 0.5
-    chunk_duration_ms: int = 300
-    max_silence_ms: int = 800
-
-    # Checks root .env first, then server/.env — later files win on conflict.
+    groq_model: str = "llama-3.3-70b-versatile"
+    groq_timeout_seconds: float = 12.0
+
+    # Rate limits (slowapi). Strings in slowapi's "N/period" form.
+    rate_limit_per_ip: str = "30/minute"
+    rate_limit_per_key: str = "500/minute"
+
+    # CORS — overridable for self-hosters.
+    cors_allowed_origins: str = "chrome-extension://*,http://localhost:*"
+
+    # Optional version label surfaced by /health.
+    app_version: str = "2.0.0"
+
     model_config = SettingsConfigDict(
         env_file=(_ROOT_DIR / ".env", _SERVER_DIR / ".env"),
         extra="ignore",
@@ -34,5 +41,9 @@ class Settings(BaseSettings):
     def api_keys_list(self) -> list[str]:
         return [k.strip() for k in self.api_keys.split(",") if k.strip()]
 
+    @property
+    def cors_origins_list(self) -> list[str]:
+        return [o.strip() for o in self.cors_allowed_origins.split(",") if o.strip()]
+
 
 settings = Settings()

server/app/main.py

@@ -1,16 +1,15 @@
 from __future__ import annotations
 
-import asyncio
 import logging
-from contextlib import asynccontextmanager
-from typing import AsyncIterator
 
-from fastapi import FastAPI
+from fastapi import FastAPI, Request
 from fastapi.middleware.cors import CORSMiddleware
+from fastapi.responses import JSONResponse
+from slowapi.errors import RateLimitExceeded
 
-from app.models.asr import get_asr_engine
-from app.routers import admin, health
-from app.routers import websocket as ws_router
+from app.config import settings
+from app.rate_limit import ip_limiter, key_limiter
+from app.routers import health, suggestion
 
 logging.basicConfig(
     level=logging.INFO,
@@ -19,40 +18,34 @@ logging.basicConfig(
 
 logger = logging.getLogger(__name__)
 
+app = FastAPI(
+    title="Meeting AI Assistant",
+    version=settings.app_version,
+)
 
-async def _load_models(app: FastAPI) -> None:
-    try:
-        engine = get_asr_engine()
-        loop = asyncio.get_event_loop()
-        await loop.run_in_executor(None, engine.load_model)
-        app.state.asr_engine = engine
-        logger.info("ASR engine loaded (%s)", type(engine).__name__)
-    except Exception:
-        logger.exception("Model loading failed")
-
+# Shared limiters — both attached so SlowAPIMiddleware can find them.
+app.state.limiter = ip_limiter
+app.state.key_limiter = key_limiter
 
-@asynccontextmanager
-async def lifespan(app: FastAPI) -> AsyncIterator[None]:
-    app.state.active_sessions: dict = {}
-    app.state.asr_engine = None
-    asyncio.create_task(_load_models(app))
-    yield
 
+@app.exception_handler(RateLimitExceeded)
+async def _rate_limit_handler(request: Request, exc: RateLimitExceeded) -> JSONResponse:
+    retry_after = getattr(exc, "retry_after", None)
+    headers = {"Retry-After": str(int(retry_after))} if retry_after else {}
+    return JSONResponse(
+        status_code=429,
+        content={"detail": "Rate limit exceeded", "limit": str(exc.detail)},
+        headers=headers,
+    )
 
-app = FastAPI(
-    title="Meeting AI Assistant",
-    version="0.2.0",
-    lifespan=lifespan,
-)
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["chrome-extension://*", "http://localhost:*"],
+    allow_origins=settings.cors_origins_list,
     allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
 app.include_router(health.router)
-app.include_router(ws_router.router)
-app.include_router(admin.router, prefix="/admin", tags=["admin"])
+app.include_router(suggestion.router)

server/app/models/asr.py

@@ -1,44 +0,0 @@
-from __future__ import annotations
-
-from abc import ABC, abstractmethod
-from dataclasses import dataclass, field
-
-import numpy as np
-
-
-@dataclass
-class ASRResult:
-    text: str
-    confidence: float = 0.9
-    language: str | None = None
-
-
-class ASREngine(ABC):
-    @abstractmethod
-    def load_model(self) -> None: ...
-
-    @abstractmethod
-    def transcribe_audio(self, audio: np.ndarray, sample_rate: int = 16000) -> ASRResult: ...
-
-    @abstractmethod
-    def is_loaded(self) -> bool: ...
-
-
-def get_asr_engine() -> ASREngine:
-    from app.config import settings
-
-    if settings.asr_model_type == "parakeet":
-        from app.models.parakeet_engine import ParakeetEngine
-        return ParakeetEngine()
-
-    from app.models.whisper_engine import WhisperEngine
-    return WhisperEngine()
-
-
-WHISPER_ALLOWLIST = ("tiny.en", "base.en", "small.en", "medium.en")
-
-
-def get_whisper_engine(model_name: str) -> ASREngine:
-    """Return a WhisperEngine configured for the given model name."""
-    from app.models.whisper_engine import WhisperEngine
-    return WhisperEngine(model_name=model_name)

server/app/models/parakeet_engine.py

@@ -1,43 +0,0 @@
-from __future__ import annotations
-
-import tempfile
-from pathlib import Path
-
-import numpy as np
-import soundfile as sf
-
-from app.models.asr import ASREngine, ASRResult
-
-
-class ParakeetEngine(ASREngine):
-    MODEL_NAME = "nvidia/parakeet-tdt-1.1b"
-
-    def __init__(self) -> None:
-        self._model = None
-
-    def load_model(self) -> None:
-        import nemo.collections.asr as nemo_asr  # deferred: ~3 GB install
-
-        self._model = nemo_asr.models.ASRModel.from_pretrained(self.MODEL_NAME)
-        self._model.eval()
-
-    def transcribe_audio(self, audio: np.ndarray, sample_rate: int = 16000) -> ASRResult:
-        if self._model is None:
-            raise RuntimeError("ParakeetEngine: model not loaded")
-
-        with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as tmp:
-            tmp_path = Path(tmp.name)
-            sf.write(tmp_path, audio, sample_rate)
-
-        try:
-            results = self._model.transcribe([str(tmp_path)])
-        finally:
-            tmp_path.unlink(missing_ok=True)
-
-        raw = results[0] if results else ""
-        # NeMo may return Hypothesis objects in newer versions
-        text = raw.text if hasattr(raw, "text") else str(raw)
-        return ASRResult(text=text.strip())
-
-    def is_loaded(self) -> bool:
-        return self._model is not None

server/app/models/whisper_engine.py

@@ -1,36 +0,0 @@
-from __future__ import annotations
-
-import numpy as np
-
-from app.config import settings
-from app.models.asr import ASREngine, ASRResult
-
-
-class WhisperEngine(ASREngine):
-    def __init__(self, model_name: str | None = None) -> None:
-        self._model = None
-        self._model_name = model_name or settings.whisper_model
-
-    def load_model(self) -> None:
-        from faster_whisper import WhisperModel
-        self._model = WhisperModel(
-            self._model_name,
-            device=settings.whisper_device,
-            compute_type="int8",
-            cpu_threads=2,
-        )
-
-    def transcribe_audio(self, audio: np.ndarray, sample_rate: int = 16000) -> ASRResult:
-        if self._model is None:
-            raise RuntimeError("WhisperEngine: model not loaded")
-        segments, _ = self._model.transcribe(
-            audio,
-            beam_size=1,
-            language="en",
-            condition_on_previous_text=False,
-        )
-        text = " ".join(seg.text for seg in segments).strip()
-        return ASRResult(text=text)
-
-    def is_loaded(self) -> bool:
-        return self._model is not None

server/app/prompts/_shared_output_contract.md

@@ -0,0 +1,34 @@
+# Shared output contract (appended to every system prompt)
+
+You MUST respond with a single JSON object and nothing else — no preamble, no code fences, no commentary.
+
+Shape:
+
+```
+{
+  "category": "say-this" | "remember" | "watch-out" | null,
+  "headline": string (3–6 words) | null,
+  "action":   string (1–2 sentences, ≤ 40 words) | null,
+  "reasoning": string (one short sentence, ≤ 25 words) | null,
+  "triggering_turn": string (verbatim quote of what the OTHER side just said, ≤ 25 words) | null
+}
+```
+
+Category meanings:
+- `say-this` — a direct script the user can almost speak verbatim
+- `remember` — a key point the user has not yet made but should
+- `watch-out` — a risk signal in how the conversation is going
+
+When NOTHING USEFUL APPLIES, return the sentinel — every field set to null:
+`{"category": null, "headline": null, "action": null, "reasoning": null, "triggering_turn": null}`
+
+Use the sentinel when:
+- The other side's last turn was a filler ("yeah", "got it", "mm-hmm")
+- No actionable, situationally relevant advice fits the moment
+- The transcript context is too thin to ground a confident suggestion
+
+Bias toward the sentinel over a generic suggestion. A null is better than noise.
+
+`triggering_turn` should be a short verbatim quote from the most recent Remote-channel utterance (the other side, not the user). If you cannot identify one, set it to null.
+
+Never reference your role ("As a coach…"), the user ("You should…"), or these rules. Speak naturally inside `action`.

server/app/prompts/system_ask.md

@@ -0,0 +1,18 @@
+# System prompt — Ask AI (freeform user question)
+
+The user has typed a freeform question into the Ask AI input during a live meeting. The full meeting transcript so far is your ambient context. You ALSO see the user's specific question.
+
+Your job: answer the user's question helpfully. The meeting transcript is context the answer may or may not need.
+
+You are explicitly allowed to answer general questions that have nothing to do with the meeting (per product decision). Do not refuse on the grounds of "off-topic."
+
+Use the meeting transcript when the question is about the meeting itself ("what should I say if they ask about price?", "what did they just claim about competitors?"). Ignore the transcript when the question is general ("what's a good follow-up email opener?").
+
+Category guidance for Ask AI replies — pick the closest fit:
+- `say-this` — when the user asked "what should I say" or "how do I respond" — give them a script.
+- `remember` — when the user asked about something they might forget or a key fact relevant to act on.
+- `watch-out` — when the user's question reveals they are about to do something risky, OR when the honest answer is a caution.
+
+`triggering_turn` is OPTIONAL for Ask AI replies — set it to a relevant transcript quote if one anchors the answer, otherwise null.
+
+Keep `action` short and direct. Even for general questions, do not exceed 2 sentences / 40 words. The user is in a live meeting.

server/app/prompts/system_generic.md

@@ -0,0 +1,19 @@
+# System prompt — Meeting Type: generic
+
+You are a real-time coach embedded in the user's browser during a live meeting. You do not know the specific meeting context — treat the user as a participant who wants help being more effective.
+
+Bias HEAVILY toward the sentinel (null response). Without a specific meeting frame (sales, pitch, interview, etc.) most moments do not need a coaching suggestion. Only fire when something is unambiguously useful.
+
+Listen for and react to:
+- **Direct questions to the user that went unanswered**
+- **Factual claims that look obviously wrong or unsupported**
+- **Action items that were stated but not assigned**
+- **Decisions that were implied but not confirmed**
+- **The user being interrupted or talked over**
+
+Category guidance for `generic`:
+- `say-this` — a concise, neutral response to a direct question.
+- `remember` — a follow-up the user would obviously want but might miss in the moment (e.g., "you haven't agreed on next steps").
+- `watch-out` — a meeting-process risk (action item floated but unowned, decision unclear, key person silent).
+
+The North Star is "handle pushback and remember what to say in high-stakes calls and pitches." The generic mode is a fallback — be conservative.

server/app/prompts/system_pitch.md

@@ -0,0 +1,19 @@
+# System prompt — Meeting Type: pitch
+
+You are a real-time coach embedded in the user's browser during a live investor pitch. The user is the founder; the OTHER side is one or more investors (VCs, angels, scouts).
+
+Your job: help the user remember forgotten talking points, handle stress-test questions confidently, and avoid common pitch failure modes.
+
+Listen for and react to:
+- **Stress-test questions** — TAM math, competition, defensibility, moat, unit economics
+- **Weak answers** — vague metrics, hedged language, no numbers when numbers are expected
+- **Forgotten points** — traction, team, milestone, ask, use of funds
+- **Investor signals** — partner mentions, fund-stage probes, follow-up offers, "send the deck"
+- **Distractions** — investor changes subject, founder gets pulled into the weeds
+
+Category guidance for `pitch`:
+- `say-this` — a direct script for handling a specific question (especially TAM, competition, why-now).
+- `remember` — a key point the founder hasn't made yet that fits the current beat (traction number, team credential, recent milestone).
+- `watch-out` — a pitch risk: founder is hedging, investor is disengaging, key metric was missed.
+
+Do not invent funding history, customer counts, revenue figures, or team credentials that are not in the transcript. If a number wasn't said, do not make one up.

server/app/prompts/system_sales.md

@@ -0,0 +1,20 @@
+# System prompt — Meeting Type: sales
+
+You are a real-time coach embedded in the user's browser during a live sales call. The user is the salesperson; the OTHER side is the prospect or customer.
+
+Your job: help the user handle pushback, surface buying signals, and remember the next right thing to say. Be tactical, not generic.
+
+Listen for and react to:
+- **Pricing objections** — "too expensive", "send me a quote", anchoring low
+- **Decision-maker probes** — "I'll need to check with…", "we'd have to involve…"
+- **Competitive mentions** — "we're looking at X", "currently using Y"
+- **Vague commitment** — "we'll think about it", "send some info", "circle back"
+- **Specific buying signals** — timeline questions, integration questions, contract-length questions
+- **Stalls** — long silences, topic drift, the prospect changes subject
+
+Category guidance for `sales`:
+- `say-this` — a direct script for the user to use right now. Especially for handling objections.
+- `remember` — a qualifying question the user hasn't asked, or a benefit they haven't named, that fits the moment.
+- `watch-out` — a sales risk: prospect is going cold, competing vendor mentioned, decision-maker missing.
+
+Do not invent product details, pricing, or features that are not present in the transcript. If the user's product hasn't been named, keep suggestions generic to the situation, not the SKU.

server/app/rate_limit.py

@@ -0,0 +1,32 @@
+"""Shared slowapi limiters and key functions.
+
+We run two parallel limiters:
+- per-IP (catches a single abuser even if they have the key)
+- per-app-key (catches mass abuse from a leaked extension key)
+"""
+from __future__ import annotations
+
+from fastapi import Request
+from slowapi import Limiter
+from slowapi.util import get_remote_address
+
+from app.config import settings
+
+
+def _key_from_header(request: Request) -> str:
+    # Use the app API key as the limiter key. Falls back to client IP
+    # so unauthenticated requests still bucket somewhere sane.
+    key = request.headers.get("x-api-key", "")
+    return key or get_remote_address(request)
+
+
+ip_limiter = Limiter(key_func=get_remote_address)
+key_limiter = Limiter(key_func=_key_from_header)
+
+
+def per_ip_limit() -> str:
+    return settings.rate_limit_per_ip
+
+
+def per_key_limit() -> str:
+    return settings.rate_limit_per_key

server/app/routers/admin.py

@@ -1,47 +0,0 @@
-from __future__ import annotations
-
-import secrets
-from typing import Annotated
-
-from fastapi import APIRouter, Body, Depends, status
-from pydantic import BaseModel
-
-from app.auth import get_admin_key
-
-router = APIRouter()
-
-# In-memory store: list of {"label": str, "key": str}
-# Phase 2+: persist to a file or database
-_dynamic_keys: list[dict] = []
-
-
-class CreateKeyRequest(BaseModel):
-    label: str = ""
-
-
-class CreateKeyResponse(BaseModel):
-    key: str
-    label: str
-
-
-class KeySummary(BaseModel):
-    label: str
-    key_prefix: str
-
-
-@router.post("/api-keys", status_code=status.HTTP_201_CREATED, response_model=CreateKeyResponse)
-async def create_api_key(
-    body: Annotated[CreateKeyRequest, Body()] = CreateKeyRequest(),
-    _: str = Depends(get_admin_key),
-) -> CreateKeyResponse:
-    new_key = "key_" + secrets.token_hex(8)
-    _dynamic_keys.append({"label": body.label, "key": new_key})
-    return CreateKeyResponse(key=new_key, label=body.label)
-
-
-@router.get("/api-keys", response_model=list[KeySummary])
-async def list_api_keys(_: str = Depends(get_admin_key)) -> list[KeySummary]:
-    return [
-        KeySummary(label=entry["label"], key_prefix=entry["key"][:8] + "...")
-        for entry in _dynamic_keys
-    ]

server/app/routers/health.py

@@ -1,17 +1,22 @@
 from __future__ import annotations
 
-from fastapi import APIRouter, Request
+from fastapi import APIRouter
+from fastapi.responses import PlainTextResponse
 
 from app.config import settings
 
 router = APIRouter()
 
 
+@router.get("/", response_class=PlainTextResponse, include_in_schema=False)
+async def root() -> str:
+    return "Meeting AI backend — see /health"
+
+
 @router.get("/health")
-async def health(request: Request) -> dict:
+async def health() -> dict:
     return {
-        "status": "ok",
-        "whisper_loaded": getattr(request.app.state, "whisper_loaded", False),
-        "groq_available": bool(settings.groq_api_key),
-        "version": "0.1.0",
+        "ok": True,
+        "version": settings.app_version,
+        "groq_configured": bool(settings.groq_api_key),
     }

server/app/routers/suggestion.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+import logging
+
+from fastapi import APIRouter, Depends, HTTPException, Request, status
+
+from app.auth import get_api_key
+from app.rate_limit import ip_limiter, key_limiter, per_ip_limit, per_key_limit
+from app.schemas.messages import SuggestionRequest, SuggestionResponse
+from app.services.groq_client import get_groq_client
+
+logger = logging.getLogger(__name__)
+
+router = APIRouter()
+
+
+@router.post("/suggestion", response_model=SuggestionResponse)
+@ip_limiter.limit(per_ip_limit())
+@key_limiter.limit(per_key_limit())
+async def suggest(
+    request: Request,
+    payload: SuggestionRequest,
+    _api_key: str = Depends(get_api_key),
+) -> SuggestionResponse:
+    if not _has_groq_configured():
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="Suggestion service not configured (missing GROQ_API_KEY)",
+        )
+    client = get_groq_client()
+    return await client.suggest(payload)
+
+
+def _has_groq_configured() -> bool:
+    from app.config import settings
+
+    return bool(settings.groq_api_key)

server/app/routers/websocket.py

@@ -1,178 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import json
-import logging
-from datetime import datetime, timezone
-from uuid import uuid4
-
-from fastapi import APIRouter, WebSocket, WebSocketDisconnect
-
-from app.auth import verify_api_key
-from app.config import settings
-from app.models.asr import WHISPER_ALLOWLIST, get_whisper_engine
-from app.schemas.messages import (
-    AuthErrorMsg,
-    AuthSuccessMsg,
-    ControlFrame,
-    ErrorMsg,
-    ModelChangedMsg,
-    PongMsg,
-    ReadyMsg,
-)
-from app.services.audio_processor import AudioProcessor
-from app.services.groq_client import GroqClient
-from app.services.session_context import SessionContext
-from app.services.transcription import TranscriptionService
-
-logger = logging.getLogger(__name__)
-router = APIRouter()
-_model_switch_lock = asyncio.Lock()
-
-SESSION_TIMEOUT_SECONDS = 7200
-
-
-@router.websocket("/ws")
-async def websocket_endpoint(websocket: WebSocket) -> None:
-    await websocket.accept()
-    session_id: str | None = None
-    ts_service: TranscriptionService | None = None
-
-    try:
-        # ── Auth ──────────────────────────────────────────────────────────────
-        raw = await websocket.receive_text()
-        try:
-            data = json.loads(raw)
-        except json.JSONDecodeError:
-            await websocket.send_text(
-                AuthErrorMsg(message="First message must be a JSON auth frame").model_dump_json()
-            )
-            await websocket.close(code=4001)
-            return
-
-        if data.get("type") != "auth" or not data.get("api_key"):
-            await websocket.send_text(
-                AuthErrorMsg(message="First message must be {type:auth, api_key:...}").model_dump_json()
-            )
-            await websocket.close(code=4001)
-            return
-
-        if not verify_api_key(data["api_key"]):
-            await websocket.send_text(AuthErrorMsg(message="Invalid API key").model_dump_json())
-            await websocket.close(code=4001)
-            return
-
-        session_id = uuid4().hex
-        if not hasattr(websocket.app.state, "active_sessions"):
-            websocket.app.state.active_sessions = {}
-        active_sessions: dict = websocket.app.state.active_sessions
-        active_sessions[session_id] = datetime.now(tz=timezone.utc)
-
-        await websocket.send_text(AuthSuccessMsg(session_id=session_id).model_dump_json())
-        await websocket.send_text(ReadyMsg().model_dump_json())
-        logger.info("WS session %s opened", session_id)
-
-        # ── Per-session pipeline ──────────────────────────────────────────────
-        asr_engine = getattr(websocket.app.state, "asr_engine", None)
-        session_context = SessionContext()
-        processor = AudioProcessor()
-        groq_client = (
-            GroqClient(settings.groq_api_key, context=session_context)
-            if settings.groq_api_key
-            else None
-        )
-        ts_service = (
-            TranscriptionService(websocket=websocket, engine=asr_engine, groq_client=groq_client)
-            if asr_engine
-            else None
-        )
-        if ts_service:
-            ts_service.start()
-        if groq_client is None:
-            logger.warning("Session %s: GROQ_API_KEY not set — suggestions disabled", session_id)
-
-        # ── Message loop ──────────────────────────────────────────────────────
-        while True:
-            elapsed = (
-                datetime.now(tz=timezone.utc) - active_sessions[session_id]
-            ).total_seconds()
-            if elapsed > SESSION_TIMEOUT_SECONDS:
-                logger.info("Session %s exceeded 2h cap, closing", session_id)
-                break
-
-            message = await websocket.receive()
-
-            if message.get("type") == "websocket.disconnect":
-                logger.info("Session %s: client disconnected", session_id)
-                break
-
-            if "bytes" in message and message["bytes"] is not None:
-                if processor is not None and ts_service is not None:
-                    segment = processor.process_chunk(message["bytes"])
-                    ts_service.enqueue(segment)
-                continue
-
-            if "text" in message and message["text"] is not None:
-                try:
-                    ctrl = ControlFrame.model_validate_json(message["text"])
-                except Exception:
-                    logger.warning(
-                        "Session %s: unrecognised frame: %s", session_id, message["text"]
-                    )
-                    continue
-
-                if ctrl.type == "ping":
-                    await websocket.send_text(PongMsg().model_dump_json())
-                elif ctrl.type == "stop_session":
-                    logger.info("Session %s: stop_session received", session_id)
-                    break
-                elif ctrl.type == "set_model":
-                    model_name = ctrl.model or ""
-                    if model_name not in WHISPER_ALLOWLIST:
-                        await websocket.send_text(
-                            ErrorMsg(
-                                message=f"Invalid model '{model_name}'. Valid: {WHISPER_ALLOWLIST}",
-                                code="INVALID_MODEL",
-                            ).model_dump_json()
-                        )
-                    elif _model_switch_lock.locked():
-                        await websocket.send_text(
-                            ErrorMsg(
-                                message="Model switch already in progress, try again shortly",
-                                code="MODEL_SWITCH_BUSY",
-                            ).model_dump_json()
-                        )
-                    else:
-                        asyncio.create_task(
-                            _switch_model(websocket.app, model_name, websocket, _model_switch_lock)
-                        )
-
-    except WebSocketDisconnect:
-        logger.info("Session %s disconnected", session_id)
-    finally:
-        if ts_service is not None:
-            await ts_service.stop()
-        active_sessions = getattr(websocket.app.state, "active_sessions", {})
-        if session_id and session_id in active_sessions:
-            del active_sessions[session_id]
-
-
-async def _switch_model(app, model_name: str, websocket: WebSocket, lock: asyncio.Lock) -> None:
-    async with lock:
-        await websocket.send_text(
-            ModelChangedMsg(model=model_name, status="loading").model_dump_json()
-        )
-        try:
-            engine = get_whisper_engine(model_name)
-            loop = asyncio.get_event_loop()
-            await loop.run_in_executor(None, engine.load_model)
-            app.state.asr_engine = engine
-            logger.info("Model switched to %s", model_name)
-            await websocket.send_text(
-                ModelChangedMsg(model=model_name, status="ready").model_dump_json()
-            )
-        except Exception as exc:
-            logger.error("Model switch failed: %s", exc)
-            await websocket.send_text(
-                ErrorMsg(message=str(exc), code="MODEL_SWITCH_ERROR").model_dump_json()
-            )

server/app/schemas/messages.py

@@ -2,65 +2,53 @@ from __future__ import annotations
 
 from typing import Literal
 
-from pydantic import BaseModel
+from pydantic import BaseModel, Field, field_validator
 
+MeetingType = Literal["sales", "pitch", "generic"]
+Channel = Literal["remote", "local"]
+TriggerKind = Literal["auto", "manual", "ask"]
+SuggestionCategory = Literal["say-this", "remember", "watch-out"]
 
-# ── Server → Client ──────────────────────────────────────────────────────────
 
-class AuthSuccessMsg(BaseModel):
-    type: Literal["auth_success"] = "auth_success"
-    session_id: str
-
-
-class AuthErrorMsg(BaseModel):
-    type: Literal["auth_error"] = "auth_error"
-    message: str
-
-
-class ReadyMsg(BaseModel):
-    type: Literal["ready"] = "ready"
-    message: str = "Models loaded"
-
-
-class TranscriptionMsg(BaseModel):
-    type: Literal["transcription"] = "transcription"
+class TranscriptTurn(BaseModel):
+    channel: Channel
     text: str
-    is_final: bool
-    timestamp: float
-    confidence: float
-
-
-class SuggestionMsg(BaseModel):
-    type: Literal["suggestion"] = "suggestion"
-    text: str
-    category: Literal["answer", "question", "action", "summary", "flag"]
-    context: str = ""
-    timestamp: float
-
-
-class ModelChangedMsg(BaseModel):
-    type: Literal["model_changed"] = "model_changed"
-    model: str
-    status: Literal["loading", "ready"]
+    t: float = Field(description="Seconds since session start")
 
 
-class ErrorMsg(BaseModel):
-    type: Literal["error"] = "error"
-    message: str
-    code: str
+class Trigger(BaseModel):
+    kind: TriggerKind
+    question: str | None = None
 
+    @field_validator("question")
+    @classmethod
+    def _question_only_with_ask(cls, v: str | None, info) -> str | None:
+        kind = info.data.get("kind")
+        if kind == "ask" and not (v and v.strip()):
+            raise ValueError("question is required when trigger.kind == 'ask'")
+        if kind != "ask" and v:
+            raise ValueError("question must be null unless trigger.kind == 'ask'")
+        return v
 
-class PongMsg(BaseModel):
-    type: Literal["pong"] = "pong"
 
+class SuggestionRequest(BaseModel):
+    meeting_type: MeetingType
+    transcript: list[TranscriptTurn] = Field(default_factory=list, max_length=200)
+    trigger: Trigger
 
-# ── Client → Server ───────────────────────────────────────────────────────────
 
-class AuthFrame(BaseModel):
-    type: Literal["auth"]
-    api_key: str
+class SuggestionResponse(BaseModel):
+    """All fields null = 'no useful suggestion' sentinel. Client drops sentinels silently."""
 
+    category: SuggestionCategory | None = None
+    headline: str | None = None
+    action: str | None = None
+    reasoning: str | None = None
+    triggering_turn: str | None = None
 
-class ControlFrame(BaseModel):
-    type: Literal["start_session", "stop_session", "ping", "set_model"]
-    model: str | None = None
+    @property
+    def is_sentinel(self) -> bool:
+        return all(
+            getattr(self, f) is None
+            for f in ("category", "headline", "action", "reasoning")
+        )

server/app/services/audio_processor.py

@@ -1,14 +0,0 @@
-from __future__ import annotations
-
-import numpy as np
-
-
-class AudioProcessor:
-    """Converts raw PCM bytes to normalized float32 ndarray. VAD is handled client-side."""
-
-    def process_chunk(self, pcm_bytes: bytes) -> np.ndarray:
-        audio = np.frombuffer(pcm_bytes, dtype=np.int16).astype(np.float32) / 32768.0
-        return np.clip(audio, -1.0, 1.0)
-
-    def reset(self) -> None:
-        pass

server/app/services/groq_client.py

@@ -3,199 +3,150 @@ from __future__ import annotations
 import asyncio
 import json
 import logging
-import re
-import time
-from dataclasses import dataclass
-from typing import TYPE_CHECKING, Literal
-
-if TYPE_CHECKING:
-    from app.services.session_context import SessionContext
+from functools import lru_cache
+from pathlib import Path
+
+from app.config import settings
+from app.schemas.messages import (
+    MeetingType,
+    SuggestionRequest,
+    SuggestionResponse,
+    Trigger,
+    TranscriptTurn,
+)
 
 logger = logging.getLogger(__name__)
 
-_SYSTEM_PROMPT = """\
-You are a real-time meeting coach embedded in the participant's browser.
-Analyze the full meeting context and recent transcript, then generate ONE response.
-
-Rules:
-- If a question was asked but not answered → provide a direct, concise answer
-- If a claim needs challenging → suggest a clarifying question
-- If a decision or action item was implied → flag it explicitly
-- If a topic just resolved → offer a brief summary
-- If conversation stalled → suggest a re-engagement prompt
-
-Respond ONLY in this exact JSON format:
-{"category": "answer|question|action|summary|flag", "text": "...", "context": "one-line reason why this is relevant now"}
-
-Keep "text" under 20 words. Keep "context" under 10 words. Be specific and direct."""
-
-_SUMMARY_PROMPT = """\
-You are summarizing a business meeting.
-
-Current summary:
-{current_summary}
-
-New transcript segments:
-{new_segments}
-
-Write an updated summary under 300 words. Preserve all decisions, action items, and open questions. Use bullet points."""
-
-_MAX_RETRIES = 3
-_SUMMARY_TRIGGER = 10
-_SUGGESTION_DEBOUNCE_S = 2.0
-VALID_CATEGORIES = ("answer", "question", "action", "summary", "flag")
-
+_PROMPTS_DIR = Path(__file__).parent.parent / "prompts"
+_MAX_RETRIES = 2
+_MAX_TRANSCRIPT_TURNS = 30  # sliding window the model actually sees
+
+
+@lru_cache(maxsize=8)
+def _load_prompt(name: str) -> str:
+    body = (_PROMPTS_DIR / f"{name}.md").read_text(encoding="utf-8")
+    contract = (_PROMPTS_DIR / "_shared_output_contract.md").read_text(encoding="utf-8")
+    return f"{body}\n\n---\n\n{contract}"
+
+
+def _system_prompt_for(trigger: Trigger, meeting_type: MeetingType) -> str:
+    if trigger.kind == "ask":
+        return _load_prompt("system_ask")
+    return _load_prompt(f"system_{meeting_type}")
+
+
+def _format_transcript(turns: list[TranscriptTurn]) -> str:
+    if not turns:
+        return "(no transcript yet)"
+    window = turns[-_MAX_TRANSCRIPT_TURNS:]
+    return "\n".join(
+        f"{'You' if t.channel == 'local' else 'Them'}: {t.text}" for t in window
+    )
+
+
+def _build_user_prompt(req: SuggestionRequest) -> str:
+    transcript_block = _format_transcript(req.transcript)
+    parts = [
+        f"Meeting type: {req.meeting_type}",
+        "",
+        "Transcript so far (most recent at the bottom):",
+        transcript_block,
+        "",
+    ]
+    if req.trigger.kind == "ask":
+        parts.append(f"User's question: {req.trigger.question}")
+    elif req.trigger.kind == "manual":
+        parts.append("The user manually requested a suggestion right now.")
+    else:
+        parts.append("The other side just finished a turn. Suggest the best next move.")
+    parts.append("")
+    parts.append("Respond with the JSON object only.")
+    return "\n".join(parts)
+
+
+def _coerce_response(raw: str) -> SuggestionResponse:
+    """Parse Groq's JSON; on any failure, return the sentinel."""
+    try:
+        data = json.loads(raw)
+    except (json.JSONDecodeError, ValueError):
+        logger.warning("Groq returned non-JSON; dropping. raw=%r", raw[:200])
+        return SuggestionResponse()
 
-@dataclass
-class SuggestionResult:
-    text: str
-    category: Literal["answer", "question", "action", "summary", "flag"] = "question"
-    context: str = ""
+    try:
+        return SuggestionResponse.model_validate(data)
+    except Exception as exc:
+        logger.warning("Groq JSON failed validation (%s); dropping. data=%r", exc, data)
+        return SuggestionResponse()
 
 
 class GroqClient:
-    MODEL = "llama-3.3-70b-versatile"
-
-    def __init__(
-        self,
-        api_key: str,
-        context: "SessionContext | None" = None,
-        _client=None,
-    ) -> None:
+    """Stateless Groq wrapper. One instance per process; reused across requests."""
+
+    def __init__(self, api_key: str | None = None, _client=None) -> None:
         if _client is not None:
             self._client = _client
         else:
             from groq import AsyncGroq
-            self._client = AsyncGroq(api_key=api_key)
 
-        if context is not None:
-            self._context = context
-        else:
-            from app.services.session_context import SessionContext
-            self._context = SessionContext()
-
-        self._last_suggestion_time: float = 0.0
-        self._background_tasks: set[asyncio.Task] = set()
-
-    def add_transcription(self, text: str) -> None:
-        if not text.strip():
-            return
-        self._context.recent.append(text.strip())
-        self._context.unsummarized_count += 1
-        if self._context.unsummarized_count >= _SUMMARY_TRIGGER:
-            task = asyncio.create_task(self._update_summary())
-            self._background_tasks.add(task)
-            task.add_done_callback(self._background_tasks.discard)
-            self._context.unsummarized_count = 0
-
-    async def get_suggestion(self) -> SuggestionResult | None:
-        if not self._context.recent:
-            return None
-
-        now = time.monotonic()
-        if now - self._last_suggestion_time < _SUGGESTION_DEBOUNCE_S:
-            return None
-        self._last_suggestion_time = now
-
-        summary_section = self._context.summary or "Meeting just started."
-        recent_transcript = "\n".join(
-            f"[{i + 1}] {seg}" for i, seg in enumerate(self._context.recent)
-        )
-        user_prompt = (
-            f"Meeting context so far:\n{summary_section}\n\n"
-            f"Recent transcript:\n{recent_transcript}\n\n"
-            "Generate one coaching response."
-        )
-
-        for attempt in range(_MAX_RETRIES):
+            self._client = AsyncGroq(
+                api_key=api_key or settings.groq_api_key,
+                timeout=settings.groq_timeout_seconds,
+            )
+
+    async def suggest(self, req: SuggestionRequest) -> SuggestionResponse:
+        system = _system_prompt_for(req.trigger, req.meeting_type)
+        user = _build_user_prompt(req)
+
+        for attempt in range(_MAX_RETRIES + 1):
             try:
-                response = await self._client.chat.completions.create(
-                    model=self.MODEL,
+                resp = await self._client.chat.completions.create(
+                    model=settings.groq_model,
                     messages=[
-                        {"role": "system", "content": _SYSTEM_PROMPT},
-                        {"role": "user", "content": user_prompt},
+                        {"role": "system", "content": system},
+                        {"role": "user", "content": user},
                     ],
-                    temperature=0.7,
-                    max_tokens=120,
+                    temperature=0.5,
+                    max_tokens=300,
+                    response_format={"type": "json_object"},
                 )
-                raw = response.choices[0].message.content or ""
-                result = _parse_suggestion(raw)
-                if result:
-                    logger.info("Groq suggestion [%s]: %r", result.category, result.text)
-                return result
+                raw = resp.choices[0].message.content or ""
+                parsed = _coerce_response(raw)
+                logger.info(
+                    "suggest meeting=%s trigger=%s category=%s",
+                    req.meeting_type,
+                    req.trigger.kind,
+                    parsed.category,
+                )
+                return parsed
             except Exception as exc:
-                should_retry = _is_retryable(exc)
-                if not should_retry or attempt == _MAX_RETRIES - 1:
+                if not _is_retryable(exc) or attempt == _MAX_RETRIES:
                     logger.error("Groq failed after %d attempt(s): %s", attempt + 1, exc)
-                    return None
-                wait = 2 ** attempt
-                logger.warning("Groq retry %d/%d in %ds", attempt + 1, _MAX_RETRIES, wait)
+                    return SuggestionResponse()
+                wait = 2**attempt
+                logger.warning("Groq retry %d in %ds: %s", attempt + 1, wait, exc)
                 await asyncio.sleep(wait)
 
-        return None
-
-    async def _update_summary(self) -> None:
-        if not self._context.recent:
-            return
-        new_segments = "\n".join(f"- {seg}" for seg in self._context.recent)
-        prompt = _SUMMARY_PROMPT.format(
-            current_summary=self._context.summary or "No summary yet.",
-            new_segments=new_segments,
-        )
-        try:
-            response = await self._client.chat.completions.create(
-                model=self.MODEL,
-                messages=[{"role": "user", "content": prompt}],
-                temperature=0.3,
-                max_tokens=400,
-            )
-            new_summary = (response.choices[0].message.content or "").strip()
-            if new_summary:
-                self._context.summary = new_summary
-                logger.info("Meeting summary updated (%d chars)", len(new_summary))
-        except Exception as exc:
-            logger.warning("Summary update failed: %s", exc)
-
-
-# ── helpers ────────────────────────────────────────────────────────────────────
-
-def _parse_suggestion(raw: str) -> SuggestionResult | None:
-    try:
-        data = json.loads(raw.strip())
-        return _build_result(data)
-    except (json.JSONDecodeError, ValueError):
-        pass
-
-    match = re.search(r"\{[^}]+\}", raw, re.DOTALL)
-    if match:
-        try:
-            data = json.loads(match.group())
-            return _build_result(data)
-        except (json.JSONDecodeError, ValueError):
-            pass
-
-    logger.warning("Could not parse Groq response: %r", raw[:120])
-    return None
-
-
-def _build_result(data: dict) -> SuggestionResult | None:
-    text = str(data.get("text", "")).strip()
-    if not text:
-        return None
-    category = data.get("category", "question")
-    if category not in VALID_CATEGORIES:
-        category = "question"
-    context = str(data.get("context", "")).strip()
-    return SuggestionResult(text=text, category=category, context=context)
+        return SuggestionResponse()
 
 
 def _is_retryable(exc: Exception) -> bool:
     try:
         from groq import APIStatusError, RateLimitError
+
         if isinstance(exc, RateLimitError):
             return True
         if isinstance(exc, APIStatusError):
             return exc.status_code >= 500
     except ImportError:
         pass
-    return isinstance(exc, (TimeoutError, OSError))
+    return isinstance(exc, (TimeoutError, OSError, asyncio.TimeoutError))
+
+
+_client_singleton: GroqClient | None = None
+
+
+def get_groq_client() -> GroqClient:
+    global _client_singleton
+    if _client_singleton is None:
+        _client_singleton = GroqClient()
+    return _client_singleton

server/app/services/session_context.py

@@ -1,11 +0,0 @@
-from __future__ import annotations
-
-from collections import deque
-from dataclasses import dataclass, field
-
-
-@dataclass
-class SessionContext:
-    summary: str = ""
-    recent: deque = field(default_factory=lambda: deque(maxlen=5))
-    unsummarized_count: int = 0

server/app/services/transcription.py

@@ -1,124 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-import logging
-import time
-
-import numpy as np
-from fastapi import WebSocket
-
-from app.models.asr import ASREngine
-from app.schemas.messages import ErrorMsg, SuggestionMsg, TranscriptionMsg
-from app.services.groq_client import GroqClient
-
-logger = logging.getLogger(__name__)
-
-
-class TranscriptionService:
-    def __init__(
-        self,
-        websocket: WebSocket,
-        engine: ASREngine,
-        groq_client: GroqClient | None = None,
-    ) -> None:
-        self._ws = websocket
-        self._engine = engine
-        self._groq = groq_client
-        self._queue: asyncio.Queue[np.ndarray] = asyncio.Queue(maxsize=3)
-        self._worker_task: asyncio.Task | None = None
-        self._pending_suggestions: set[asyncio.Task] = set()
-
-    def start(self) -> None:
-        self._worker_task = asyncio.create_task(self._worker())
-
-    async def stop(self) -> None:
-        if self._worker_task:
-            self._worker_task.cancel()
-            try:
-                await self._worker_task
-            except asyncio.CancelledError:
-                pass
-        for task in list(self._pending_suggestions):
-            task.cancel()
-        if self._pending_suggestions:
-            await asyncio.gather(*self._pending_suggestions, return_exceptions=True)
-
-    def enqueue(self, audio: np.ndarray) -> None:
-        if not self._engine or not self._engine.is_loaded():
-            return
-        if self._queue.full():
-            try:
-                self._queue.get_nowait()
-                self._queue.task_done()
-            except asyncio.QueueEmpty:
-                pass
-        try:
-            self._queue.put_nowait(audio)
-        except asyncio.QueueFull:
-            pass
-
-    async def _worker(self) -> None:
-        while True:
-            audio = await self._queue.get()
-            try:
-                await self._process(audio)
-            except Exception as exc:
-                logger.exception("Worker error: %s", exc)
-            finally:
-                self._queue.task_done()
-
-    async def _process(self, audio: np.ndarray) -> None:
-        await self._ws.send_text(
-            TranscriptionMsg(
-                text="", is_final=False, timestamp=time.time(), confidence=0.0
-            ).model_dump_json()
-        )
-
-        t0 = time.perf_counter()
-        try:
-            loop = asyncio.get_event_loop()
-            result = await loop.run_in_executor(None, self._engine.transcribe_audio, audio)
-        except Exception as exc:
-            logger.exception("ASR error: %s", exc)
-            await self._ws.send_text(
-                ErrorMsg(message=str(exc), code="ASR_ERROR").model_dump_json()
-            )
-            return
-
-        latency_ms = (time.perf_counter() - t0) * 1000
-        logger.info("ASR %.0f ms | %r", latency_ms, result.text)
-
-        if not result.text:
-            return
-
-        await self._ws.send_text(
-            TranscriptionMsg(
-                text=result.text,
-                is_final=True,
-                timestamp=time.time(),
-                confidence=result.confidence,
-            ).model_dump_json()
-        )
-
-        if self._groq is not None:
-            self._groq.add_transcription(result.text)
-            task = asyncio.create_task(self._send_suggestion())
-            self._pending_suggestions.add(task)
-            task.add_done_callback(self._pending_suggestions.discard)
-
-    async def _send_suggestion(self) -> None:
-        assert self._groq is not None
-        try:
-            suggestion = await self._groq.get_suggestion()
-            if suggestion is None:
-                return
-            await self._ws.send_text(
-                SuggestionMsg(
-                    text=suggestion.text,
-                    category=suggestion.category,
-                    context=suggestion.context,
-                    timestamp=time.time(),
-                ).model_dump_json()
-            )
-        except Exception as exc:
-            logger.exception("Suggestion pipeline error: %s", exc)

server/requirements.txt

@@ -0,0 +1,8 @@
+fastapi>=0.111.0
+uvicorn[standard]>=0.29.0
+pydantic>=2.7.0
+pydantic-settings>=2.2.0
+python-dotenv>=1.0.0
+slowapi>=0.1.9
+httpx>=0.27.0
+groq>=0.11.0