feat(stats): add server-side scan counter to /v1/stats + scan_events.jsonl

- /v1/stats now returns assessments_completed alongside proofs_generated
- POST /v1/assess logs each scan to data/scan_events.jsonl for persistent analytics
- Enables measuring real API scan volume independently from local CLI scans

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: desiorac

trust_layer/app.py

@@ -1918,15 +1918,19 @@ async def get_agent_json():
 
 @app.get("/v1/stats")
 async def get_stats():
-    """Public proof count — no auth required. Cached 60s."""
-    from .config import PROOFS_DIR
+    """Public proof + assess counts — no auth required. Cached 60s."""
+    from .config import PROOFS_DIR, ASSESSMENTS_DIR
     import time
     cache = getattr(get_stats, "_cache", None)
     now = time.monotonic()
     if cache and now - cache["ts"] < 60:
         return cache["data"]
-    count = sum(1 for f in PROOFS_DIR.iterdir() if f.suffix == ".json") if PROOFS_DIR.exists() else 0
-    data = {"proofs_generated": count}
+    proof_count = sum(1 for f in PROOFS_DIR.iterdir() if f.suffix == ".json") if PROOFS_DIR.exists() else 0
+    assess_count = sum(1 for f in ASSESSMENTS_DIR.iterdir() if f.suffix == ".json") if ASSESSMENTS_DIR.exists() else 0
+    data = {
+        "proofs_generated": proof_count,
+        "assessments_completed": assess_count,
+    }
     get_stats._cache = {"ts": now, "data": data}
     return data
 

trust_layer/routers/assess.py

@@ -5,17 +5,21 @@
 """
 
 import hashlib
+import json
 import logging
 from datetime import datetime, timezone
 from typing import Optional
 
 from fastapi import APIRouter, Header, Request
 from fastapi.responses import JSONResponse
 
+from ..config import DATA_DIR
 from ..keys import validate_api_key
 from ..mcp_assess import build_assessment, ASSESS_DAILY_LIMIT
 from ..redis_client import get_redis
 
+SCAN_EVENTS_LOG = DATA_DIR / "scan_events.jsonl"
+
 logger = logging.getLogger("trust_layer.routers.assess")
 
 router = APIRouter()
@@ -138,4 +142,18 @@ async def assess_endpoint(
         logger.error("Assessment failed for server_id=%s: %s", server_id, e)
         return _error("assessment_error", "Assessment failed. Please retry.", 500)
 
+    # --- Log scan event for server-side analytics ---
+    try:
+        with open(SCAN_EVENTS_LOG, "a") as f:
+            f.write(json.dumps({
+                "ts": datetime.now(timezone.utc).isoformat(),
+                "event": "assess",
+                "server_id": server_id,
+                "tools_count": len(tools),
+                "risk_score": assessment.get("risk_score"),
+                "key_hash": fp[:8],
+            }) + "\n")
+    except Exception:
+        pass  # non-critical
+
     return JSONResponse(status_code=200, content=assessment)