feat(obs): add /metrics endpoint and structured ai_call logging (#17)

* ci: run workflow only on PRs and skip eval on drafts

* feat(ai): track input/output tokens in ai_call logging

* feat(api): wrap hub routes with observability logger

* feat(obs): expose /metrics endpoint via transparent-fastapi

* test: add metrics test and update ai mocks for token counts

.github/workflows/ci.yml

@@ -1,6 +1,14 @@
 name: ci
 
-on: [push, pull_request]
+# ONLY run on pull requests targeting the master branch
+on:
+  pull_request:
+    branches: [ master ]
+
+# Cancel in-progress runs if a new commit is pushed to the same PR
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
 
 jobs:
   test:
@@ -42,6 +50,8 @@ jobs:
   eval:
     runs-on: ubuntu-latest
     needs: test
+    # ONLY run the eval job if it is a PR and it's NOT a draft
+    if: github.event.pull_request.draft == false
     services:
       postgres:
         image: pgvector/pgvector:pg16

app/main.py

@@ -1,13 +1,18 @@
 import uuid
 import logging
 from fastapi import FastAPI, Request, Depends
+from fastapi.middleware.cors import CORSMiddleware
 from fastapi.exceptions import RequestValidationError
 from starlette.exceptions import HTTPException as StarletteHTTPException
-from app.errors import AppError, error_response
+from transparent_fastapi import install
+
+from app.config import settings
 from app.routers import auth, hubs, documents
+from app.errors import AppError, error_response
 from app.deps import get_current_user
 from app.models import User
 from app.services.groq_service import AiUnavailable
+
 logger = logging.getLogger("uvicorn.error")
 
 def create_app() -> FastAPI:
@@ -17,6 +22,17 @@ def create_app() -> FastAPI:
         docs_url="/docs",
         redoc_url=None,
     )
+    
+    # Initialize transparent-fastapi for /metrics
+    install(app)
+
+    app.add_middleware(
+        CORSMiddleware,
+        allow_origins=settings.cors_origins_list,
+        allow_credentials=True,
+        allow_methods=["*"],
+        allow_headers=["*"],
+    )
 
     app.include_router(auth.router, prefix="/api/v1")
     app.include_router(hubs.router, prefix="/api/v1")

app/routers/hubs.py

@@ -3,7 +3,10 @@ from fastapi import APIRouter, Depends, status, Query, Response
 from sqlalchemy.orm import Session
 from sqlalchemy.exc import IntegrityError
 from sqlalchemy import text
+
 from app.services.groq_service import hub_insights, answer_question, expand_query
+from app.services.embeddings import embed_texts
+from app.services.observability import log_ai_call
 
 from app.db import get_db
 from app.deps import get_current_user
@@ -16,8 +19,6 @@ from app.idempotency import (
     claim_key, complete_key, release_claim, replay_or_conflict
 )
 from app.chunking import chunk_text
-from app.services.embeddings import embed_texts
-from app.services.observability import log_ai_call
 from app.config import settings
 from app.rate_limit import check_query_rate
 
@@ -113,6 +114,7 @@ def create_hub(
             try:
                 chunks_data = chunk_text(body.content)
                 texts = [c[0] for c in chunks_data]
+                # Wrap embedding call
                 embeddings = log_ai_call("embed_texts", user_id, new_hub.id, embed_texts, texts)
                 
                 for i, ((text, tokens), emb) in enumerate(zip(chunks_data, embeddings)):
@@ -156,12 +158,14 @@ def query_hub(
     if not hub:
         raise AppError(404, "HUB_NOT_FOUND", "Hub not found.")
 
-    alt_queries = expand_query(req.question)
+    # Wrap expand_query
+    alt_queries = log_ai_call("expand_query", current_user.id, hub_id, expand_query, req.question)
+    
     all_queries = [req.question] + alt_queries
-    embeddings = embed_texts(all_queries)
-
-    chunks = repo.retrieve_chunks_for_queries(db, current_user.id, hub_id, embeddings)
     
+    # Wrap embed_texts
+    embeddings = log_ai_call("embed_texts", current_user.id, hub_id, embed_texts, all_queries)
+
     raw_sql = text("""
         SELECT id, text, chunk_index, document_id, 1 - (embedding <=> :q) as similarity
         FROM chunks
@@ -187,7 +191,8 @@ def query_hub(
     context_texts = [row.text for row in results]
     context = "\n\n---\n\n".join(context_texts)
 
-    ai_result = answer_question(context, req.question)
+    # Wrap answer_question
+    ai_result = log_ai_call("answer_question", current_user.id, hub_id, answer_question, context, req.question)
 
     sources = []
     if ai_result["evidence"]:
@@ -208,4 +213,4 @@ def query_hub(
         grounded=ai_result["grounded"],
         sources=sources,
         retrieved_chunks=len(results)
-    )
+    )

app/services/groq_service.py

@@ -25,7 +25,7 @@ EXPAND_SYSTEM = (
     '{"queries": ["alternative 1", "alternative 2"]}'
 )
 
-def _call(system: str, user: str) -> dict:
+def _call(system: str, user: str) -> tuple[dict, int, int]:
     try:
         chat = _client.chat.completions.create(
             messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
@@ -34,29 +34,31 @@ def _call(system: str, user: str) -> dict:
             response_format={"type": "json_object"},
             max_tokens=1024
         )
-        return json.loads(chat.choices[0].message.content)
+        return json.loads(chat.choices[0].message.content), chat.usage.prompt_tokens, chat.usage.completion_tokens
     except Exception as exc:
         raise AiUnavailable("provider", str(exc)) from exc
 
-def hub_insights(content: str) -> tuple[str | None, list[str]]:
+def hub_insights(content: str) -> tuple[tuple[str | None, list[str]], int, int]:
     try:
-        d = _call(INSIGHTS_SYSTEM, f"<context>{content}</context>")
-        return str(d.get("summary", ""))[:500], [str(q)[:200] for q in d.get("questions", [])][:3]
+        d, in_tok, out_tok = _call(INSIGHTS_SYSTEM, f"<context>{content}</context>")
+        return (str(d.get("summary", ""))[:500], [str(q)[:200] for q in d.get("questions", [])][:3]), in_tok, out_tok
     except (AiUnavailable, KeyError, TypeError):
-        return None, []
+        return (None, []), 0, 0
 
-def answer_question(content: str, question: str) -> dict:
-    d = _call(ANSWER_SYSTEM, f"<context>{content}</context>\n<question>{question}</question>")
+def answer_question(content: str, question: str) -> tuple[dict, int, int]:
+    d, in_tok, out_tok = _call(ANSWER_SYSTEM, f"<context>{content}</context>\n<question>{question}</question>")
     evidence = d.get("evidence")
     verified = (isinstance(evidence, str) and bool(evidence.strip())
                 and evidence.lower() in content.lower())
-    return {"answer": str(d.get("answer", "")).strip(),
-            "evidence": evidence if verified else None,
-            "grounded": bool(d.get("grounded")) and verified}
+    return {
+        "answer": str(d.get("answer", "")).strip(),
+        "evidence": evidence if verified else None,
+        "grounded": bool(d.get("grounded")) and verified
+    }, in_tok, out_tok
 
-def expand_query(question: str) -> list[str]:
+def expand_query(question: str) -> tuple[list[str], int, int]:
     try:
-        d = _call(EXPAND_SYSTEM, question)
-        return [str(q) for q in d.get("queries", [])][:2]
+        d, in_tok, out_tok = _call(EXPAND_SYSTEM, question)
+        return [str(q) for q in d.get("queries", [])][:2], in_tok, out_tok
     except Exception:
-        return []
+        return [], 0, 0

app/services/observability.py

@@ -8,15 +8,21 @@ from app.services.groq_service import AiUnavailable
 logger = logging.getLogger("uvicorn.error")
 
 def log_ai_call(operation: str, user_id: UUID | str, hub_id: UUID | str | None, fn: Callable, *args, **kwargs) -> Any:
-    """Wraps an AI call, logs the duration and status, and re-raises exceptions."""
+    """Wraps an AI call, logs the duration/tokens/status, and re-raises exceptions."""
     start = time.monotonic()
     status = "success"
+    input_tokens = 0
+    output_tokens = 0
     
     try:
         result = fn(*args, **kwargs)
+        # If the wrapped function returns (data, in_tok, out_tok), unpack it
+        if isinstance(result, tuple) and len(result) == 3 and isinstance(result[1], int) and isinstance(result[2], int):
+            data, input_tokens, output_tokens = result
+            return data
         return result
     except Exception as e:
-        status = "error" if not isinstance(e, AiUnavailable) else e.kind
+        status = getattr(e, "kind", "error")
         raise e
     finally:
         duration_ms = int((time.monotonic() - start) * 1000)
@@ -27,7 +33,9 @@ def log_ai_call(operation: str, user_id: UUID | str, hub_id: UUID | str | None,
             "user_id": str(user_id),
             "hub_id": str(hub_id) if hub_id else None,
             "duration_ms": duration_ms,
+            "input_tokens": input_tokens,
+            "output_tokens": output_tokens,
             "status": status,
             "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
         }
-        logger.info(json.dumps(log_data))
+        logger.info(json.dumps(log_data))

tests/test_ai.py

@@ -7,14 +7,19 @@ def test_hub_insights_success():
         mock_res = MagicMock()
         mock_res.choices = [MagicMock()]
         mock_res.choices[0].message.content = '{"summary": "Test summary", "questions": ["Q1", "Q2"]}'
+        mock_res.usage.prompt_tokens = 10
+        mock_res.usage.completion_tokens = 20
         mock_create.return_value = mock_res
-        summary, questions = hub_insights("Test content")
+
+        (summary, questions), in_tok, out_tok = hub_insights("Test content")
         assert summary == "Test summary"
         assert questions == ["Q1", "Q2"]
+        assert in_tok == 10
+        assert out_tok == 20
 
 def test_hub_insights_failure_returns_none():
     with patch.object(_client.chat.completions, "create", side_effect=Exception("API Down")):
-        summary, questions = hub_insights("Test content")
+        (summary, questions), in_tok, out_tok = hub_insights("Test content")
         assert summary is None
         assert questions == []
 
@@ -23,15 +28,33 @@ def test_answer_question_verifies_evidence():
         mock_res1 = MagicMock()
         mock_res1.choices = [MagicMock()]
         mock_res1.choices[0].message.content = '{"answer": "20 days", "evidence": "20 days", "grounded": true}'
+        mock_res1.usage.prompt_tokens = 10
+        mock_res1.usage.completion_tokens = 20
         mock_create.return_value = mock_res1
-        result = answer_question("Employees get 20 days of leave.", "How many days?")
+
+        result, _, _ = answer_question("Employees get 20 days of leave.", "How many days?")
         assert result["grounded"] is True
         assert result["evidence"] == "20 days"
 
         mock_res2 = MagicMock()
         mock_res2.choices = [MagicMock()]
         mock_res2.choices[0].message.content = '{"answer": "20 days", "evidence": "40 days", "grounded": true}'
+        mock_res2.usage.prompt_tokens = 10
+        mock_res2.usage.completion_tokens = 20
         mock_create.return_value = mock_res2
-        result = answer_question("Employees get 20 days of leave.", "How many days?")
+
+        result, _, _ = answer_question("Employees get 20 days of leave.", "How many days?")
         assert result["grounded"] is False
-        assert result["evidence"] is None
+        assert result["evidence"] is None
+
+def test_expand_query_success():
+    with patch.object(_client.chat.completions, "create") as mock_create:
+        mock_res = MagicMock()
+        mock_res.choices = [MagicMock()]
+        mock_res.choices[0].message.content = '{"queries": ["alt1", "alt2"]}'
+        mock_res.usage.prompt_tokens = 10
+        mock_res.usage.completion_tokens = 20
+        mock_create.return_value = mock_res
+
+        alternatives, _, _ = expand_query("What is the policy?")
+        assert alternatives == ["alt1", "alt2"]

tests/test_metrics.py

@@ -0,0 +1,18 @@
+import pytest
+
+def test_metrics_endpoint_returns_prometheus_text(client):
+    """Verify that /metrics is public and returns valid Prometheus format."""
+    res = client.get("/metrics")
+    assert res.status_code == 200
+    assert "text/plain" in res.headers["content-type"]
+    
+    # Prometheus format usually includes HELP and TYPE lines
+    content = res.text
+    assert "# HELP" in content or "# TYPE" in content
+    
+    # It should track HTTP requests (this might require a request to happen first)
+    # Trigger a request to increment the counter
+    client.get("/healthz")
+    
+    res2 = client.get("/metrics")
+    assert "http_requests_total" in res2.text