Deploy full Fitness AI Agents backend: auth, DB, watch sync, GPS routes, AI analysis

backend/auth.py

@@ -0,0 +1,44 @@
+import os
+import jwt
+from jwt import PyJWKClient
+from fastapi import Depends, HTTPException, Security
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
+
+bearer = HTTPBearer()
+
+CLERK_JWKS_URL    = os.getenv("CLERK_JWKS_URL")
+CLERK_ISSUER_URL  = os.getenv("CLERK_ISSUER_URL", "")  # e.g. https://pretty-bird-74.clerk.accounts.dev
+
+_jwks_client: PyJWKClient | None = None
+
+
+def get_jwks_client() -> PyJWKClient:
+    global _jwks_client
+    if _jwks_client is None:
+        _jwks_client = PyJWKClient(CLERK_JWKS_URL, cache_keys=True)
+    return _jwks_client
+
+
+async def verify_token(credentials: HTTPAuthorizationCredentials = Security(bearer)) -> dict:
+    token = credentials.credentials
+    try:
+        signing_key = get_jwks_client().get_signing_key_from_jwt(token)
+        decode_kwargs: dict = {
+            "algorithms": ["RS256"],
+            "options": {"verify_aud": False},
+        }
+        if CLERK_ISSUER_URL:
+            decode_kwargs["issuer"] = CLERK_ISSUER_URL
+        payload = jwt.decode(token, signing_key.key, **decode_kwargs)
+        return payload
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=401, detail="Token expired")
+    except jwt.InvalidTokenError as e:
+        raise HTTPException(status_code=401, detail=f"Invalid token: {str(e)}")
+
+
+def get_user_id(payload: dict = Depends(verify_token)) -> str:
+    user_id = payload.get("sub")
+    if not user_id:
+        raise HTTPException(status_code=401, detail="No user ID in token")
+    return user_id

backend/bots.py

@@ -1,26 +1,42 @@
-from crewai import Agent, Crew, Task, Process, LLM
-from crewai_tools import FileReadTool
+"""
+DataFlow AI — single-file backend.
+FastAPI SSE server + CrewAI 6-agent pipeline.
+"""
+
+# ── Stdlib ────────────────────────────────────────────────────────────────────
+import sys
+import io
+import queue
+import threading
+import tempfile
 import os
+import re as _re
 import time as _time
+import asyncio
+import json
+import json as _json
 from threading import Lock
-from typing import Optional, Literal
+from typing import Optional, Literal, List as _List
+
+# ── Third-party ───────────────────────────────────────────────────────────────
+from dotenv import load_dotenv
+load_dotenv()
+
 from pydantic import BaseModel
 
-import re as _re
-import json as _json
+from crewai import Agent, Crew, Task, Process, LLM
+from crewai_tools import FileReadTool
+
 import litellm
-litellm.cache = None        # Disable response caching
-litellm.drop_params = True  # Silently drop unsupported params per provider
+litellm.cache = None
+litellm.drop_params = True
 
+# ── Groq cache_breakpoint patch ───────────────────────────────────────────────
 # Groq rejects messages that contain a 'cache_breakpoint' property.
-# Two-layer fix:
-#  1. Force caching=False so litellm's client wrapper never injects cache_breakpoint
-#     into messages internally (happens when CrewAI passes caching=True in kwargs).
-#  2. Strip cache_breakpoint from any message that already has it (belt + suspenders).
 _real_completion = litellm.completion
 
 def _completion_no_cache_breakpoint(*args, **kwargs):
-    kwargs["caching"] = False   # prevents litellm wrapper from re-injecting cache_breakpoint
+    kwargs["caching"] = False
     for msg in kwargs.get("messages", []):
         if isinstance(msg, dict):
             msg.pop("cache_breakpoint", None)
@@ -33,53 +49,41 @@ def _completion_no_cache_breakpoint(*args, **kwargs):
 litellm.completion = _completion_no_cache_breakpoint
 
 # ── Model rotation pools ──────────────────────────────────────────────────────
-# Tier 1 (Groq): 14,400 req/day free, fast, tried first.
-# Tier 2 (OpenRouter :free): shared global rate limits — deep fallback only.
 _FAST_MODELS = [
-    # ── Tier 1: Groq ──────────────────────────────────────────────────────────
-    "groq/llama-3.1-8b-instant",                                   # Groq — primary
-    "groq/gemma2-9b-it",                                           # Groq
-    "groq/llama3-8b-8192",                                         # Groq — stable fallback
-    # ── Tier 2: OpenRouter free fallback ──────────────────────────────────────
-    "openrouter/nvidia/nemotron-nano-9b-v2:free",                  # NVIDIA
-    "openrouter/minimax/minimax-m2.5:free",                        # OpenInference
-    "openrouter/meta-llama/llama-3.1-8b-instruct:free",            # Meta/Lepton
-    "openrouter/mistralai/mistral-7b-instruct:free",               # Mistral
-    "openrouter/google/gemma-3-12b-it:free",                       # Google
-    "openrouter/qwen/qwen3-8b:free",                               # Qwen small
-    "openrouter/meta-llama/llama-4-scout:free",                    # Meta Llama 4
-    "openrouter/microsoft/phi-3-mini-128k-instruct:free",          # Microsoft
+    "groq/llama-3.1-8b-instant",
+    "groq/gemma2-9b-it",
+    "groq/llama3-8b-8192",
+    "openrouter/nvidia/nemotron-nano-9b-v2:free",
+    "openrouter/minimax/minimax-m2.5:free",
+    "openrouter/meta-llama/llama-3.1-8b-instruct:free",
+    "openrouter/mistralai/mistral-7b-instruct:free",
+    "openrouter/google/gemma-3-12b-it:free",
+    "openrouter/qwen/qwen3-8b:free",
+    "openrouter/meta-llama/llama-4-scout:free",
+    "openrouter/microsoft/phi-3-mini-128k-instruct:free",
 ]
 _SMART_MODELS = [
-    # ── Tier 1: Groq ──────────────────────────────────────────────────────────
-    "groq/llama-3.3-70b-versatile",                                # Groq — best tool use
-    "groq/llama3-70b-8192",                                        # Groq — stable 70b fallback
-    # ── Tier 2: OpenRouter free fallback ──────────────────────────────────────
-    "openrouter/google/gemma-3-27b-it:free",                       # Google
-    "openrouter/qwen/qwen3-coder:free",                            # Qwen
-    "openrouter/meta-llama/llama-3.3-70b-instruct:free",           # Meta
-    "openrouter/deepseek/deepseek-chat-v3-0324:free",              # DeepSeek
-    "openrouter/meta-llama/llama-4-maverick:free",                 # Meta Llama 4
-    "openrouter/mistralai/mistral-small-3.1-24b-instruct:free",    # Mistral
-    "openrouter/google/gemma-3-12b-it:free",                       # Google smaller
+    "groq/llama-3.3-70b-versatile",
+    "groq/llama3-70b-8192",
+    "openrouter/google/gemma-3-27b-it:free",
+    "openrouter/qwen/qwen3-coder:free",
+    "openrouter/meta-llama/llama-3.3-70b-instruct:free",
+    "openrouter/deepseek/deepseek-chat-v3-0324:free",
+    "openrouter/meta-llama/llama-4-maverick:free",
+    "openrouter/mistralai/mistral-small-3.1-24b-instruct:free",
+    "openrouter/google/gemma-3-12b-it:free",
 ]
 
-
-# ── Module-level cooldown state (persists across requests) ───────────────────
-# Groq models share one org-wide TPM quota — rate-limiting one limits all.
+# ── Cooldown state ────────────────────────────────────────────────────────────
 _GROQ_MODELS_ALL: frozenset = frozenset(
     m for m in _FAST_MODELS + _SMART_MODELS if m.startswith("groq/")
 )
-_cooldown: dict[str, float] = {}   # model → monotonic timestamp when available again
+_cooldown: dict[str, float] = {}
 _cooldown_lock = Lock()
-
-# Serialize concurrent HTTP requests so they don't collide on shared Groq TPM.
-# Without this, two simultaneous requests each consume ~12000 TPM and both fail.
 _crew_lock = Lock()
 
 
 def _set_cooldown(model: str, seconds: float) -> None:
-    """Mark model unavailable. Groq models cool together (shared org TPM quota)."""
     until = _time.monotonic() + seconds
     with _cooldown_lock:
         if model.startswith("groq/"):
@@ -90,22 +94,16 @@ def _set_cooldown(model: str, seconds: float) -> None:
 
 
 def _pick_model(pool: list[str]) -> tuple[str, int]:
-    """Return (model, index) of the highest-priority model not in cooldown.
-    Always scans from index 0 so higher-priority models (Groq) are preferred
-    the moment their cooldown expires. Falls back to soonest-available if all
-    are still cooling."""
     now = _time.monotonic()
     with _cooldown_lock:
         for idx, model in enumerate(pool):
             if _cooldown.get(model, 0.0) <= now:
                 return model, idx
-        # All cooling — return the one whose cooldown expires soonest
         best = min(range(len(pool)), key=lambda i: _cooldown.get(pool[i], 0.0))
         return pool[best], best
 
 
 def _wait_until_available() -> None:
-    """Sleep until at least one model in each pool is ready. No-op if already ready."""
     now = _time.monotonic()
     with _cooldown_lock:
         fast_waits = [max(0.0, _cooldown.get(m, 0.0) - now) for m in _FAST_MODELS]
@@ -117,12 +115,6 @@ def _wait_until_available() -> None:
 
 
 def _parse_retry_after(err_str: str) -> float:
-    """Extract retry delay from provider error string, default 35s.
-
-    Handles two formats:
-    - OpenRouter: retry_after_seconds: 30
-    - Groq:       Please try again in 2.3s.
-    """
     m = _re.search(r"retry_after_seconds['\"\s:]+(\d+(?:\.\d+)?)", err_str)
     if m:
         return float(m.group(1)) + 5
@@ -133,26 +125,14 @@ def _parse_retry_after(err_str: str) -> float:
 
 
 def _extract_json(text: str) -> str:
-    """
-    Robustly extract a JSON object from LLM output.
-
-    LLMs often wrap output in markdown fences (```json...```) despite instructions.
-    This strips fences first, then uses brace-counting to find the first complete
-    JSON object — avoiding greedy regex that matches across multiple objects.
-    """
-    # Strip markdown fences
     text = _re.sub(r"^```(?:json)?\s*", "", text.strip(), flags=_re.MULTILINE)
     text = _re.sub(r"\s*```$", "", text.strip(), flags=_re.MULTILINE)
     text = text.strip()
-
-    # Try parsing the cleaned text directly
     try:
         _json.loads(text)
         return text
     except _json.JSONDecodeError:
         pass
-
-    # Walk characters counting braces to find the first balanced JSON object
     start = text.find("{")
     if start != -1:
         depth = 0
@@ -175,26 +155,21 @@ def _extract_json(text: str) -> str:
             elif ch == "}":
                 depth -= 1
                 if depth == 0:
-                    candidate = text[start : i + 1]
+                    candidate = text[start: i + 1]
                     try:
                         _json.loads(candidate)
                         return candidate
                     except _json.JSONDecodeError:
                         break
-
     return text
 
 
-# LiteLLM reads provider keys from environment automatically for known prefixes.
-# Mirror the OpenRouter key to OPENAI_API_KEY so LiteLLM internal paths that
-# fall back to OpenAI don't error when no OpenAI key is set.
 _OR_KEY = os.getenv("OPENROUTER_API_KEY", "")
 if _OR_KEY:
     os.environ.setdefault("OPENAI_API_KEY", _OR_KEY)
 
 
 def _api_key_for(model: str) -> str | None:
-    """Return the correct API key for a given model string."""
     if model.startswith("groq/"):
         return os.getenv("GROQ_API_KEY")
     return os.getenv("OPENROUTER_API_KEY")
@@ -206,59 +181,103 @@ class DataPoint(BaseModel):
     label: str
     value: float
     category: Optional[str] = None
+    x_value: Optional[float] = None   # scatter: second axis
+    value2: Optional[float] = None    # radar: second series value
+
+
+class CodeBlock(BaseModel):
+    language: str    # python | sql | bash | r | javascript
+    title: str
+    code: str        # keep under 400 chars
+
+
+class MetricItem(BaseModel):
+    label: str
+    value: str       # formatted string, e.g. "1,234" or "98.5%"
+    unit: Optional[str] = None
+    trend: Optional[str] = None    # up | down | neutral
+    change: Optional[str] = None   # e.g. "+12%"
+    context: Optional[str] = None  # e.g. "vs last week"
+
+
+class ComparisonRow(BaseModel):
+    metric: str          # e.g. "Avg Revenue"
+    value_a: str         # formatted value for entity A
+    value_b: str         # formatted value for entity B
+    winner: Optional[Literal["a", "b", "tie"]] = None
 
 
 class FormattedOutput(BaseModel):
     """
-    Strict output contract for the output_formatter agent.
-
-    RULES:
-    - output_type = "chart" when the findings contain numerical data that can
-      be meaningfully compared across discrete categories or over time.
-      Examples: revenue by product, sessions by user, requests by endpoint,
-      errors by day.
-    - output_type = "report" when the findings are qualitative, narrative,
-      or mixed — no clear numerical comparison is possible.
-
-    CHART TYPE RULES (only when output_type = "chart"):
-    - chart_type = "bar"  → comparing discrete categories (products, regions,
-      users, endpoints). Most common choice.
-    - chart_type = "line" → trend over time (daily counts, weekly averages).
-      Use when labels are dates or sequential time periods.
-    - chart_type = "pie"  → parts of a whole where values sum to a meaningful
-      total and there are 2-6 categories. Do NOT use for more than 6 slices.
-
-    DATA RULES:
-    - data_points must be populated whenever output_type = "chart".
-    - label: the category name (short, no line breaks).
-    - value: the primary numeric metric (revenue, count, duration, etc.).
-    - category: optional secondary grouping (e.g. "Hardware", "Software").
-    - x_axis_label: what the labels represent (e.g. "Product", "Date", "Region").
-    - y_axis_label: what the values represent (e.g. "Revenue ($)", "Sessions").
-
-    REPORT FIELDS (always required):
-    - summary: 2-3 sentences answering the user's original question directly.
-    - findings: exactly 3-5 bullet-ready strings, each a concrete fact from data.
-    - recommendations: exactly 2-3 actionable strings based solely on the data.
+    OUTPUT TYPE DECISION RULES — pick the FIRST that matches:
+    1. "code"       → answer is or includes runnable code/queries/scripts.
+                      Populate code_blocks (1-3 blocks).
+    2. "metrics"    → answer is a set of KPIs or key numbers (3-8 items).
+                      Populate metrics list.
+    3. "comparison" → comparing two named entities side-by-side across metrics.
+                      Populate comparison_a_label, comparison_b_label, comparison_rows.
+    4. "heatmap"    → data is a matrix (rows × columns) of numeric values — e.g.
+                      a correlation matrix, time-of-day × day-of-week activity grid,
+                      or category × category frequency table.
+                      Populate heatmap_row_labels, heatmap_col_labels, heatmap_values.
+                      Max 10 rows × 10 cols.
+    5. "table"      → ranked/multi-attribute list best shown as labelled rows+columns.
+                      Populate table_headers and table_rows (max 20 rows).
+    6. "chart"      → 2+ numeric values that can be compared visually.
+                      Choose chart_type:
+                        bar     → named categories
+                        line    → sequential time periods
+                        pie     → parts of a whole, 2-6 slices
+                        scatter → correlation (set x_value + value per DataPoint)
+                        funnel  → sequential stages with drop-off (conversion pipelines)
+                        radar   → multi-attribute profile comparison across dimensions;
+                                  set value for series A; set value2 + radar_b_label
+                                  if comparing two entities on the same axes.
+    7. "report"     → qualitative or narrative findings.
+
+    ALWAYS REQUIRED:
+    - summary: 2-3 sentences directly answering the original question.
+    - findings: 3-5 specific factual strings from the data.
+    - recommendations: 2-3 actionable strings.
+    - Set unused fields to null.
     """
-    output_type: Literal["chart", "report"]
-    chart_type: Optional[Literal["bar", "line", "pie"]] = None
+    output_type: Literal["chart", "report", "code", "table", "metrics", "comparison", "heatmap"]
+    # chart
+    chart_type: Optional[Literal["bar", "line", "pie", "scatter", "funnel", "radar"]] = None
     chart_title: Optional[str] = None
     x_axis_label: Optional[str] = None
     y_axis_label: Optional[str] = None
     data_points: Optional[list[DataPoint]] = None
+    radar_b_label: Optional[str] = None   # label for value2 series in radar
+    # code
+    code_blocks: Optional[list[CodeBlock]] = None
+    # table
+    table_headers: Optional[list[str]] = None
+    table_rows: Optional[list[list[str]]] = None
+    # metrics
+    metrics: Optional[list[MetricItem]] = None
+    # comparison
+    comparison_a_label: Optional[str] = None
+    comparison_b_label: Optional[str] = None
+    comparison_rows: Optional[list[ComparisonRow]] = None
+    # heatmap
+    heatmap_title: Optional[str] = None
+    heatmap_row_labels: Optional[list[str]] = None
+    heatmap_col_labels: Optional[list[str]] = None
+    heatmap_values: Optional[list[list[float]]] = None  # [row_idx][col_idx]
+    # always
     summary: str
     findings: list[str]
     recommendations: list[str]
+    quality_score: Optional[int] = None
+    quality_verdict: Optional[str] = None
 
 
-# ── Bots ──────────────────────────────────────────────────────────────────────
+# ── Agent pipeline ────────────────────────────────────────────────────────────
 
 class Bots:
     def __init__(self, context: str):
         self.context = context
-        # Escape braces so CrewAI's .format() interpolation doesn't treat user
-        # text like "{something}" as a template variable and raise KeyError.
         self._ctx = context.replace("{", "{{").replace("}", "}}")
         self._fast_idx = 0
         self._smart_idx = 0
@@ -266,8 +285,6 @@ class Bots:
 
     def _smart_llm(self, temperature: float) -> LLM:
         model = _SMART_MODELS[self._smart_idx % len(_SMART_MODELS)]
-        # 1024 keeps total TPM per call under Groq's 12000/min limit when
-        # input is large (file contents + long prompt chain).
         return LLM(
             model=model,
             api_key=_api_key_for(model),
@@ -277,12 +294,12 @@ class Bots:
             temperature=temperature,
         )
 
-    def _fast_llm(self, temperature: float) -> LLM:
+    def _fast_llm(self, temperature: float, max_tokens: int = 1024) -> LLM:
         model = _FAST_MODELS[self._fast_idx % len(_FAST_MODELS)]
         return LLM(
             model=model,
             api_key=_api_key_for(model),
-            max_tokens=1024,
+            max_tokens=max_tokens,
             max_retries=0,
             timeout=120,
             temperature=temperature,
@@ -294,15 +311,11 @@ class Bots:
             goal=(
                 "Read the user's raw context and rewrite it as a precise, unambiguous "
                 "analysis directive. Identify the core question, the most relevant columns "
-                "or metrics, and the exact type of analysis needed (trend, comparison, "
-                "anomaly, summary, correlation). Output only the directive — nothing else."
+                "or metrics, and the exact type of analysis needed."
             ),
             backstory=(
-                "You are an expert at translating vague or freeform requests into sharp, "
-                "actionable instructions for data analysts. You have a talent for identifying "
-                "what someone actually wants to know versus what they literally said. "
-                "You never perform analysis yourself — your only job is to make the analyst's "
-                "directive so clear that there is no room for misinterpretation."
+                "You translate vague requests into sharp, actionable instructions. "
+                "You never perform analysis — you only clarify the directive."
             ),
             tools=[],
             verbose=True,
@@ -312,20 +325,35 @@ class Bots:
             cache=False,
         )
 
+        self.data_cleaner = Agent(
+            role="Data Quality Inspector",
+            goal=(
+                "Read every file provided and produce a concise data quality report: "
+                "column names, row count, missing values, duplicate rows, data type issues. "
+                "Keep under 200 words."
+            ),
+            backstory=(
+                "You are a meticulous data auditor. You use FileReadTool once per file, "
+                "then summarise its structure and flag obvious problems."
+            ),
+            tools=[self.file_read],
+            verbose=True,
+            memory=False,
+            max_iter=5,
+            llm=self._fast_llm(0.1, max_tokens=512),
+            allow_delegation=False,
+            cache=False,
+        )
+
         self.prompt_engineer = Agent(
             role="Data Analysis Prompt Engineer",
             goal=(
-                "Take the raw FastAPI input and the analysis directive, then construct a "
-                "precise, step-by-step analysis prompt for the data analyst. The prompt must "
-                "specify exactly which columns to examine, what calculations to run, what "
-                "patterns to look for, and in what order to approach the analysis."
+                "Construct a precise, step-by-step analysis prompt for the data analyst. "
+                "Specify exact columns, calculations, patterns to look for, and order of steps."
             ),
             backstory=(
-                "You are a specialist in writing technical prompts for data analysis pipelines. "
-                "You understand how LLM-based analysts think and know that vague instructions "
-                "produce vague results. You break every analysis job into clear, ordered steps "
-                "with explicit column names, metric names, and success criteria. You never "
-                "perform the analysis yourself — you only write the instruction that makes it happen."
+                "You write technical prompts for data analysis pipelines. "
+                "Vague instructions produce vague results — you are never vague."
             ),
             tools=[],
             verbose=True,
@@ -338,17 +366,12 @@ class Bots:
         self.data_analyst = Agent(
             role="Senior Data Analyst",
             goal=(
-                "Follow the analysis prompt exactly. If a file path is provided, call FileReadTool "
-                "ONCE to read the file, then reason over its contents to answer the prompt. "
-                "If no file is provided, reason from the context and prompt alone. "
-                "Never speculate beyond what the data or context shows."
+                "Follow the analysis prompt exactly. Call FileReadTool ONCE per file path. "
+                "Reason over the content to answer the prompt. Never speculate beyond the data."
             ),
             backstory=(
-                "You are a rigorous data analyst with experience across many domains and file formats. "
-                "You call FileReadTool exactly once per task — re-reading the same file wastes tokens "
-                "and produces no new information. After reading, you reason directly over the content. "
-                "When no file is provided, you produce a thorough analytical response from context alone. "
-                "You back every finding with evidence from the data."
+                "You are a rigorous analyst. You call FileReadTool exactly once per file — "
+                "re-reading wastes tokens. You back every finding with evidence."
             ),
             tools=[self.file_read],
             verbose=True,
@@ -362,22 +385,30 @@ class Bots:
         self.output_formatter = Agent(
             role="Structured Output Specialist",
             goal=(
-                "Convert the analyst's findings into a strict FormattedOutput JSON object. "
-                "Decide output_type based on one rule: if there are numerical values that can "
-                "be meaningfully compared across 2 or more categories, use 'chart'. Otherwise "
-                "use 'report'. Never invent data — only use what the analyst found."
+                "Convert analyst findings into a strict FormattedOutput JSON object. "
+                "Choose output_type by priority: code → metrics → comparison → heatmap → table → chart → report."
             ),
             backstory=(
-                "You are an expert in structured data serialization. You always output valid "
-                "JSON matching the FormattedOutput schema exactly — no extra keys, no missing "
-                "required fields, no markdown fences around the JSON. "
-                "Your chart selection rules are strict: "
-                "bar for category comparisons, line for time-series trends, pie only for "
-                "2-6 slices that sum to a whole. You populate data_points with the actual "
-                "numbers from the analyst's report, converted to floats. "
-                "You always write summary in plain English (2-3 sentences), "
-                "findings as 3-5 specific factual strings, and recommendations as 2-3 "
-                "actionable strings. You never add commentary outside the JSON object."
+                "You serialise analysis results into one of 7 output modes:\n"
+                "• code       — runnable scripts, queries, or algorithms\n"
+                "• metrics    — key numbers / KPIs (3-8 items)\n"
+                "• comparison — two named entities compared across metrics\n"
+                "• heatmap    — matrix of values (correlation, frequency, activity)\n"
+                "• table      — ranked/multi-attribute list (max 20 rows)\n"
+                "• chart      — bar, line, pie, scatter, funnel, or radar\n"
+                "• report     — qualitative or narrative findings\n\n"
+                "CHART TYPE SELECTION:\n"
+                "  funnel → sequential conversion stages with drop-off\n"
+                "  radar  → multi-attribute profile (use value2+radar_b_label for dual series)\n"
+                "  scatter → correlation (x_value + value per point)\n"
+                "  pie → 2-6 parts of a whole\n"
+                "  line → time series\n"
+                "  bar → named categories\n\n"
+                "COMPARISON: comparison_a_label and comparison_b_label name the two entities. "
+                "Each comparison_row has metric, value_a, value_b, and winner (a/b/tie).\n\n"
+                "HEATMAP: heatmap_values is a 2D list [row][col] of floats. Max 10×10.\n\n"
+                "Output ONLY the raw JSON object. No markdown fences, no preamble. "
+                "summary=2-3 sentences, findings=3-5 strings, recommendations=2-3 strings."
             ),
             tools=[],
             verbose=True,
@@ -387,64 +418,89 @@ class Bots:
             cache=False,
         )
 
+        self.qa_critic = Agent(
+            role="Analysis Quality Critic",
+            goal=(
+                "Rate how well the analysis answered the original question. "
+                'Output ONLY: {"score": <int 1-10>, "verdict": "<1-2 sentences>"}'
+            ),
+            backstory=(
+                "You review analyses for completeness, specificity, and evidence quality. "
+                "Score 10 = every aspect answered with data. Score <5 = question not answered. "
+                "Output ONLY the raw JSON — no markdown, no preamble."
+            ),
+            tools=[],
+            verbose=True,
+            memory=False,
+            llm=self._fast_llm(0.2, max_tokens=512),
+            allow_delegation=False,
+            cache=False,
+        )
+
     def create_tasks(self):
         self.interpret_task = Task(
             description=(
-                f"The user has provided this context about what they want analyzed:\n\n"
-                f"CONTEXT: {self._ctx}\n\n"
-                "Rewrite this into a structured analysis directive by answering these four questions:\n"
-                "1. What is the single core question to answer?\n"
-                "2. Which columns or metrics are most relevant to that question?\n"
-                "3. What analysis type is needed — trend over time, comparison between groups, "
-                "anomaly detection, statistical summary, or correlation?\n"
-                "4. Are there any constraints or focus areas implied by the context "
-                "(e.g. a date range, a specific user, a threshold)?\n\n"
-                "Write the final directive as 3-5 plain sentences addressed directly to a data analyst."
+                f"The user wants: {self._ctx}\n\n"
+                "Rewrite this into a structured analysis directive:\n"
+                "1. The single core question to answer\n"
+                "2. Relevant columns/metrics\n"
+                "3. Analysis type (trend, comparison, anomaly, summary, correlation)\n"
+                "4. Any constraints (date range, thresholds, focus areas)\n\n"
+                "Write 3-5 plain sentences addressed directly to a data analyst."
             ),
             expected_output=(
-                "A single block of 3-5 plain sentences. No headers, no bullet points, no preamble. "
-                "Written as a direct instruction to a data analyst. Must specify: the question to answer, "
-                "the relevant columns, the analysis type, and any constraints."
+                "3-5 plain sentences. No headers, no bullets. "
+                "Direct instruction specifying: the question, relevant columns, analysis type, constraints."
             ),
             agent=self.context_agent,
         )
 
+        self.clean_task = Task(
+            description=(
+                "Dataset path(s):\n{data}\n\n"
+                "If {data} is not '(no file)', use FileReadTool to read each path once.\n"
+                "Report per file: type/size, column names, row count, missing values, "
+                "obvious issues, 2 sample records.\n"
+                "If no file: report 'No file provided — analysis uses context only.'\n"
+                "Keep under 200 words."
+            ),
+            expected_output="Concise data quality report under 200 words. Plain prose or bullets. No JSON.",
+            context=[self.interpret_task],
+            agent=self.data_cleaner,
+        )
+
         self.prompt_task = Task(
             description=(
-                f"You have received the original user request and an analysis directive.\n\n"
-                f"ORIGINAL REQUEST: {self._ctx}\n\n"
-                "Using the directive from the previous step, write a precise step-by-step "
-                "analysis prompt for the data analyst. Your prompt must include:\n"
-                "1. The exact columns to load and examine\n"
-                "2. The specific calculations or aggregations to run\n"
-                "3. What patterns, outliers, or trends to look for\n"
-                "4. The order in which to approach the analysis\n"
-                "5. What a complete, correct answer looks like"
+                f"Original request: {self._ctx}\n\n"
+                "Using the directive and data quality report, write a step-by-step analysis prompt:\n"
+                "1. Exact columns to load\n"
+                "2. Calculations/aggregations to run\n"
+                "3. Patterns, outliers, or trends to look for\n"
+                "4. Order of approach\n"
+                "5. What a complete answer looks like"
             ),
             expected_output=(
-                "A numbered step-by-step prompt addressed to a data analyst. "
-                "Each step must be specific and actionable — no vague instructions. "
-                "Must reference exact column names or metric types where possible."
+                "Numbered step-by-step prompt. Each step specific and actionable. "
+                "References exact column names where possible."
             ),
-            context=[self.interpret_task],
+            context=[self.interpret_task, self.clean_task],
             agent=self.prompt_engineer,
         )
 
         self.analyze_task = Task(
             description=(
-                "You have been given a step-by-step analysis prompt from the previous task.\n\n"
-                "Dataset path: {data}\n\n"
-                "If {data} is not empty, use FileReadTool to read the file contents.\n\n"
-                "If {data} is empty, answer based on the analysis prompt using your knowledge "
-                "and reasoning — state clearly that no file was provided.\n\n"
-                "Follow every step in the prompt exactly. Report only what the data shows."
+                "Dataset path(s):\n{data}\n\n"
+                "If file paths are provided (one per line), read each with FileReadTool exactly once. "
+                "If multiple files, analyze together. "
+                "If no file, answer from the analysis prompt using reasoning.\n\n"
+                "Follow every step in the prompt. Read each file only once. Report only what the data shows."
             ),
             expected_output=(
-                "A thorough data analysis report containing:\n"
-                "1. Data source summary (file type, rows found, or 'no file provided')\n"
-                "2. Key statistics relevant to the prompt (averages, ranges, counts, outliers)\n"
-                "3. 3-5 concrete findings that directly answer the prompt\n"
-                "4. 2-3 actionable recommendations based on the findings"
+                "Thorough analysis containing:\n"
+                "1. Data source summary\n"
+                "2. Key statistics (averages, ranges, counts, outliers)\n"
+                "3. 3-5 concrete findings that answer the prompt\n"
+                "4. 2-3 actionable recommendations"
             ),
             context=[self.prompt_task],
             agent=self.data_analyst,
@@ -452,40 +508,53 @@ class Bots:
 
         self.format_task = Task(
             description=(
-                f"Convert the analyst's findings into a FormattedOutput JSON object.\n\n"
-                f"ORIGINAL USER REQUEST: {self._ctx}\n\n"
-                "DECISION RULE — output_type:\n"
-                "  'chart' if the findings contain at least 2 data points with distinct numerical "
-                "values that can be meaningfully compared (revenue by product, sessions by user, "
-                "requests by endpoint, errors by day, etc.).\n"
-                "  'report' for qualitative, narrative, or text-heavy findings.\n\n"
-                "CHART TYPE SELECTION:\n"
-                "  bar   → comparing named categories (products, regions, users, endpoints)\n"
-                "  line  → trend over sequential time periods (days, weeks, months)\n"
-                "  pie   → parts of a whole, 2-6 categories only\n\n"
-                "OUTPUT REQUIREMENTS:\n"
-                "- Return ONLY the raw JSON object — no markdown fences, no commentary.\n"
-                "- data_points: extract actual numbers from the analyst's report as floats.\n"
-                "- summary: 2-3 sentences answering the original request directly.\n"
-                "- findings: 3-5 specific factual strings (not bullet points, just the text).\n"
-                "- recommendations: 2-3 actionable strings.\n"
-                "- chart_title, x_axis_label, y_axis_label: short, descriptive strings.\n"
-                "- If output_type is 'report', set chart_type, chart_title, x_axis_label, "
-                "y_axis_label, and data_points to null."
+                f"Original request: {self._ctx}\n\n"
+                "Convert the analyst's findings into a FormattedOutput JSON object.\n\n"
+                "OUTPUT TYPE PRIORITY (pick first that fits):\n"
+                "  'code'       → answer is or includes runnable code/queries/scripts\n"
+                "  'metrics'    → answer is a set of KPIs or key numbers (3-8 items)\n"
+                "  'comparison' → comparing two named entities across multiple metrics;\n"
+                "                 set comparison_a_label, comparison_b_label, comparison_rows\n"
+                "                 (each row: metric, value_a, value_b, winner='a'/'b'/'tie')\n"
+                "  'heatmap'    → data is a matrix (rows × cols) of numeric values;\n"
+                "                 set heatmap_row_labels, heatmap_col_labels,\n"
+                "                 heatmap_values (2D float list), heatmap_title. Max 10×10.\n"
+                "  'table'      → ranked/multi-attribute list, max 20 rows\n"
+                "  'chart'      → visual comparison of 2+ values; chart_type options:\n"
+                "                 bar, line, pie, scatter, funnel, radar\n"
+                "                 For funnel: stages in order, value = count/rate at each stage\n"
+                "                 For radar: label=axis, value=series A; optionally value2=series B\n"
+                "                            and set radar_b_label for B's name\n"
+                "  'report'     → qualitative/narrative findings\n\n"
+                "ALWAYS: summary (2-3 sentences), findings (3-5 strings), recommendations (2-3 strings).\n"
+                "Return ONLY the raw JSON object. No markdown, no commentary."
             ),
             expected_output=(
-                "A single raw JSON object matching the FormattedOutput schema. "
-                "No markdown fences, no preamble, no trailing text. "
-                "The JSON must be parseable by json.loads() without modification."
+                "Single raw JSON object matching FormattedOutput. "
+                "No markdown fences. Parseable by json.loads() without modification."
             ),
             context=[self.analyze_task],
             agent=self.output_formatter,
             output_pydantic=FormattedOutput,
         )
 
+        self.qa_task = Task(
+            description=(
+                f"Original request: {self._ctx}\n\n"
+                "Review the completed analysis. Score 1-10 based on:\n"
+                "- Did it directly and specifically answer the original question?\n"
+                "- Are findings backed by concrete numbers from the data?\n"
+                "- Are recommendations actionable and relevant?\n"
+                "- Is anything important missing, vague, or invented?\n\n"
+                "Return ONLY: "
+                '{"score": <int 1-10>, "verdict": "<1-2 sentences: what was done well and what gap remains>"}'
+            ),
+            expected_output='Raw JSON only: {"score": <int>, "verdict": "<string>"}. No markdown.',
+            context=[self.analyze_task, self.format_task],
+            agent=self.qa_critic,
+        )
+
     def create_crew(self, data) -> str:
-        # Hold the lock for the entire pipeline so a second concurrent request
-        # waits rather than hammering the same shared Groq TPM quota.
         with _crew_lock:
             return self._run_pipeline(data)
 
@@ -493,12 +562,7 @@ class Bots:
         max_attempts = (len(_FAST_MODELS) + len(_SMART_MODELS)) * 2
 
         for attempt in range(max_attempts):
-            # Wait BEFORE picking models — ensures cooldowns are honoured
-            # even on the very first attempt of a retry cycle.
             _wait_until_available()
-
-            # Scan from index 0 every time so Groq (index 0) is always preferred
-            # when its cooldown has expired. Cooldowns handle skipping, not the index.
             fast_model, self._fast_idx = _pick_model(_FAST_MODELS)
             smart_model, self._smart_idx = _pick_model(_SMART_MODELS)
 
@@ -506,8 +570,14 @@ class Bots:
             self.create_tasks()
 
             crew = Crew(
-                agents=[self.context_agent, self.prompt_engineer, self.data_analyst, self.output_formatter],
-                tasks=[self.interpret_task, self.prompt_task, self.analyze_task, self.format_task],
+                agents=[
+                    self.context_agent, self.data_cleaner, self.prompt_engineer,
+                    self.data_analyst, self.output_formatter, self.qa_critic,
+                ],
+                tasks=[
+                    self.interpret_task, self.clean_task, self.prompt_task,
+                    self.analyze_task, self.format_task, self.qa_task,
+                ],
                 process=Process.sequential,
                 verbose=True,
                 memory=False,
@@ -536,13 +606,32 @@ class Bots:
                         _set_cooldown(fast_model, 60)
                         _set_cooldown(smart_model, 60)
                         print(f"[SERVER-ERR] fast={fast_model} smart={smart_model} → 60s cooldown")
-
-                    continue  # _wait_until_available() fires at the top of the next iteration
+                    continue
                 raise
 
-            # Best case: pydantic model validated successfully
-            if hasattr(result, "pydantic") and result.pydantic is not None:
-                return result.pydantic.model_dump_json()
+            fmt_task_out = crew.tasks[4].output if len(crew.tasks) > 4 else None
+            formatted = None
+            if fmt_task_out:
+                if getattr(fmt_task_out, "pydantic", None):
+                    formatted = fmt_task_out.pydantic
+                else:
+                    try:
+                        formatted = FormattedOutput(
+                            **_json.loads(_extract_json(fmt_task_out.raw or ""))
+                        )
+                    except Exception:
+                        pass
+
+            if formatted:
+                qa_raw = result.raw if hasattr(result, "raw") else str(result)
+                try:
+                    qa = _json.loads(_extract_json(qa_raw))
+                    formatted.quality_score = int(qa.get("score", 0)) or None
+                    formatted.quality_verdict = qa.get("verdict")
+                except Exception:
+                    pass
+                return formatted.model_dump_json()
 
             raw = result.raw if hasattr(result, "raw") else str(result)
             return _extract_json(raw)
+

backend/calories.py

@@ -0,0 +1,48 @@
+"""Server-side calorie estimation.
+
+Keeps the watch firmware thin: the device no longer hardcodes a body weight or
+runs MET math (the ESP32-C3 has no FPU). The watch sends duration + workout type
+and the backend estimates the burn here, in one place.
+
+Formula: calories ≈ MET × weight_kg × hours. MET ("metabolic equivalent") values
+are moderate-intensity averages from the Compendium of Physical Activities.
+"""
+from typing import Optional
+
+# TODO: replace with the authenticated user's real weight once a profile exists.
+DEFAULT_WEIGHT_KG = 70.0
+
+# MET by activity. Keys are lowercased; the watch currently reports "running".
+_MET = {
+    "running":       9.8,
+    "run":           9.8,
+    "jogging":       7.0,
+    "walking":       3.5,
+    "walk":          3.5,
+    "hiking":        6.0,
+    "cycling":       7.5,
+    "bike":          7.5,
+    "biking":        7.5,
+    "swimming":      8.0,
+    "rowing":        7.0,
+    "elliptical":    5.0,
+    "weightlifting": 6.0,
+    "strength":      6.0,
+    "workout":       7.0,
+}
+_DEFAULT_MET = 7.0
+
+
+def met_for(workout_type: Optional[str]) -> float:
+    if not workout_type:
+        return _DEFAULT_MET
+    return _MET.get(workout_type.strip().lower(), _DEFAULT_MET)
+
+
+def estimate_calories(duration_minutes: float,
+                      workout_type: Optional[str] = None,
+                      weight_kg: float = DEFAULT_WEIGHT_KG) -> float:
+    """Estimate kcal burned for a workout, rounded to 1 decimal place."""
+    if not duration_minutes or duration_minutes <= 0:
+        return 0.0
+    return round(met_for(workout_type) * weight_kg * (duration_minutes / 60.0), 1)

backend/db.py

@@ -0,0 +1,14 @@
+import os
+from supabase import acreate_client, AsyncClient
+
+SUPABASE_URL = os.getenv("SUPABASE_URL")
+SUPABASE_KEY = os.getenv("SUPABASE_KEY")
+
+_client: AsyncClient | None = None
+
+
+async def get_db() -> AsyncClient:
+    global _client
+    if _client is None:
+        _client = await acreate_client(SUPABASE_URL, SUPABASE_KEY)
+    return _client

backend/main.py

@@ -1,201 +1,33 @@
-import sys
-import io
-import queue
-import threading
-import tempfile
-import os
-import re
-import time
-import asyncio
-import json
 from dotenv import load_dotenv
 load_dotenv()
 
-from fastapi import FastAPI, UploadFile, File, Form
+import os
+from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
-from fastapi.responses import StreamingResponse, JSONResponse
-from bots import Bots
+from routes import watch, analysis, user, charts, gps, integrations
+
+app = FastAPI(title="Fitness AI Agents")
 
-app = FastAPI()
+_ALLOWED_ORIGINS = [o.strip() for o in os.getenv(
+    "ALLOWED_ORIGINS",
+    "http://localhost:5173,http://localhost:4173",
+).split(",") if o.strip()]
 
 app.add_middleware(
     CORSMiddleware,
-    allow_origins=["*"],
+    allow_origins=_ALLOWED_ORIGINS,
+    allow_credentials=True,
     allow_methods=["*"],
     allow_headers=["*"],
 )
 
-ANSI_ESCAPE = re.compile(r"\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~])")
-BOX_CHARS  = re.compile(r"[╭╮╰╯│╞╡╢╟╔╗╚╝╬═─┼┤├┬┴┌└┐┘╠╣╦╧╨╩╪╫]")
-
-MAX_FILE_SIZE   = 10 * 1024 * 1024   # 10 MB
-MAX_CONTEXT_LEN = 2000               # chars
-MAX_RUNTIME     = 600                # seconds (10 min total pipeline timeout)
-ALLOWED_EXTS    = {".csv", ".json", ".txt", ".pdf", ".xml"}
-
-
-class LineCapture(io.TextIOBase):
-    """Captures stdout/stderr line-by-line into a queue, stripping ANSI codes."""
-
-    def __init__(self, q: queue.Queue):
-        self._q = q
-        self._buf = ""
-
-    def write(self, text: str) -> int:
-        cleaned = ANSI_ESCAPE.sub("", text)
-        cleaned = BOX_CHARS.sub("", cleaned)
-        self._buf += cleaned
-        while "\n" in self._buf:
-            line, self._buf = self._buf.split("\n", 1)
-            stripped = line.strip()
-            if stripped:
-                self._q.put(stripped)
-        return len(text)
-
-    def flush(self):
-        if self._buf.strip():
-            self._q.put(self._buf.strip())
-            self._buf = ""
-
+app.include_router(watch.router,    prefix="/watch",    tags=["Watch"])
+app.include_router(analysis.router, prefix="/analyze",  tags=["Analysis"])
+app.include_router(user.router,     prefix="/user",     tags=["User"])
+app.include_router(charts.router,   prefix="/charts",   tags=["Charts"])
+app.include_router(gps.router,          prefix="/routes",       tags=["GPS"])
+app.include_router(integrations.router, prefix="/integrations", tags=["Integrations"])
 
 @app.get("/health")
 async def health():
     return {"status": "ok"}
-
-
-@app.post("/analyze")
-async def analyze(context: str = Form(...), file: UploadFile = File(None)):
-    # ── Input validation (before opening the SSE stream) ──────────────────────
-    context = context.strip()
-    if not context:
-        return JSONResponse({"error": "Context is required."}, status_code=400)
-    if len(context) > MAX_CONTEXT_LEN:
-        return JSONResponse(
-            {"error": f"Context too long ({len(context)} chars, max {MAX_CONTEXT_LEN})."},
-            status_code=400,
-        )
-
-    file_content = None
-    file_suffix = ".csv"
-    if file and file.filename:
-        ext = os.path.splitext(file.filename)[1].lower()
-        if ext not in ALLOWED_EXTS:
-            return JSONResponse(
-                {"error": f"File type '{ext}' not supported. Allowed: {', '.join(sorted(ALLOWED_EXTS))}"},
-                status_code=400,
-            )
-        file_content = await file.read()
-        if len(file_content) > MAX_FILE_SIZE:
-            return JSONResponse(
-                {"error": f"File too large ({len(file_content) // 1024}KB, max {MAX_FILE_SIZE // 1024 // 1024}MB)."},
-                status_code=400,
-            )
-        file_suffix = ext
-
-    # ── SSE stream ────────────────────────────────────────────────────────────
-    async def event_stream():
-        q: queue.Queue = queue.Queue()
-
-        def run_crew():
-            old_stdout, old_stderr = sys.stdout, sys.stderr
-            capture = LineCapture(q)
-            sys.stdout = capture
-            sys.stderr = capture
-            data_path = None
-            try:
-                if file_content:
-                    with tempfile.NamedTemporaryFile(
-                        delete=False, suffix=file_suffix
-                    ) as tmp:
-                        tmp.write(file_content)
-                        data_path = tmp.name
-
-                bots = Bots(context)
-                result = bots.create_crew(data_path or "")
-                if result:
-                    q.put({"__result__": result})
-            except Exception as exc:
-                # Flush any buffered partial output before the error
-                capture.flush()
-                import traceback
-                q.put(f"[ERROR] {exc}")
-                for line in traceback.format_exc().splitlines():
-                    if line.strip():
-                        q.put(f"[TRACE] {line}")
-            finally:
-                sys.stdout = old_stdout
-                sys.stderr = old_stderr
-                if data_path and os.path.exists(data_path):
-                    os.unlink(data_path)
-                q.put(None)
-
-        thread = threading.Thread(target=run_crew, daemon=True)
-        thread.start()
-
-        # CrewAI internal noise that clutters the stream during rotation/retry.
-        # We surface our own [ROTATE] / [RETRY] lines instead.
-        _NOISE = (
-            "ERROR:root:",
-            "ERROR:crewai.",
-            "[CrewAIEventsBus]",
-            "An unknown error occurred. Please check",
-            "Error details: Error code:",
-            "Error details: Model ",
-            # CrewAI event ordering warnings (box chars are stripped in LineCapture)
-            "'agent_execution_started'",
-            "Tracing Preference Saved",
-            "Tracing has been disabled",
-            "Your preference has been saved",
-            "To enable tracing later",
-            "Set tracing=True",
-            "Set CREWAI_TRACING_ENABLED",
-            "Run: crewai traces",
-            "[Finalize]",
-            "[TRACE]",
-            "✨ Update Available",
-            "collect traces.",
-            "New version of crewai",
-            "Run `pip install",
-            "pip install --upgrade",
-            "All providers rate-limited",   # litellm internal retry message
-            "Auto-retrying in",             # litellm internal retry message
-            "[RETRY]",                      # litellm router retry prefix
-        )
-
-        loop = asyncio.get_running_loop()
-        deadline = time.monotonic() + MAX_RUNTIME
-
-        while True:
-            # Overall pipeline timeout guard
-            remaining = deadline - time.monotonic()
-            if remaining <= 0:
-                yield f"data: {json.dumps('[ERROR] Analysis timed out after 10 minutes.')}\n\n"
-                yield f"data: {json.dumps('__DONE__')}\n\n"
-                break
-
-            try:
-                item = await loop.run_in_executor(
-                    None, lambda: q.get(timeout=min(30, remaining))
-                )
-            except queue.Empty:
-                # Heartbeat keeps Cloudflare proxy from closing idle connection
-                yield ": ping\n\n"
-                continue
-
-            # Drop internal CrewAI error noise during rotation; keep our own messages
-            if isinstance(item, str) and any(item.startswith(p) or p in item for p in _NOISE):
-                continue
-
-            if item is None:
-                yield f"data: {json.dumps('__DONE__')}\n\n"
-                break
-            if isinstance(item, dict) and "__result__" in item:
-                yield f"data: {json.dumps({'type': 'result', 'content': item['__result__']})}\n\n"
-            else:
-                yield f"data: {json.dumps(item)}\n\n"
-
-    return StreamingResponse(
-        event_stream(),
-        media_type="text/event-stream",
-        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
-    )

backend/models/analysis.py

@@ -0,0 +1,19 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+
+class AnalysisRequest(BaseModel):
+    context: str                       # what the user wants to know
+    date_from: Optional[datetime] = None
+    date_to: Optional[datetime] = None
+
+
+class AnalysisResult(BaseModel):
+    user_id: str
+    context: str
+    summary: str
+    key_findings: list[str]
+    anomalies: list[str]
+    recommendations: list[str]
+    created_at: datetime = datetime.utcnow()

backend/models/watch.py

@@ -0,0 +1,34 @@
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime
+
+
+class WatchReading(BaseModel):
+    timestamp: datetime
+    heart_rate: Optional[float] = None
+    steps: Optional[int] = None
+    calories_burned: Optional[float] = None
+    active_calories: Optional[float] = None
+    distance_meters: Optional[float] = None
+    sleep_hours: Optional[float] = None
+    spo2: Optional[float] = None       # blood oxygen %
+    hrv: Optional[float] = None        # heart rate variability
+
+
+class WorkoutSession(BaseModel):
+    timestamp: datetime
+    workout_type: str                  # running, cycling, weightlifting, etc.
+    duration_minutes: float
+    avg_heart_rate: Optional[float] = None
+    max_heart_rate: Optional[float] = None
+    ending_heart_rate: Optional[float] = None   # BPM at end/cool-down
+    calories_burned: Optional[float] = None
+    distance_meters: Optional[float] = None
+    notes: Optional[str] = None
+
+
+class WatchSyncPayload(BaseModel):
+    readings: list[WatchReading] = []
+    workouts: list[WorkoutSession] = []
+    device: Optional[str] = None      # "apple_watch", "garmin", "fitbit", etc.
+    app_version: Optional[str] = None

backend/requirements.txt

@@ -1,7 +1,11 @@
-fastapi
-uvicorn[standard]
-python-dotenv
-crewai
-crewai-tools
-pydantic
-# build-v2
+fastapi==0.136.1
+uvicorn==0.47.0
+python-dotenv==1.2.2
+python-multipart==0.0.20
+httpx==0.28.1
+PyJWT==2.12.1
+supabase==2.30.1
+crewai==1.14.5
+crewai-tools==1.14.5
+pandas==3.0.3
+fastembed==0.8.0

backend/routes/analysis.py

@@ -0,0 +1,119 @@
+import asyncio
+import csv
+import json
+import os
+import tempfile
+from fastapi import APIRouter, Depends, HTTPException
+from datetime import datetime, timezone
+from auth import get_user_id
+from db import get_db
+from models.analysis import AnalysisRequest
+from bots import Bots
+
+router = APIRouter()
+
+# Columns from the routes table that are meaningful for analysis.
+# Excludes coordinates (raw lat/lng blob), user_id, and id.
+_ROUTE_COLUMNS = [
+    "started_at", "ended_at", "workout_type",
+    "distance_meters", "duration_seconds", "pace", "calories_burned", "notes",
+]
+
+
+def _write_csv(rows: list[dict], columns: list[str]) -> str:
+    """Write rows to a temp CSV (only the given columns) and return the path."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".csv", delete=False, newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=columns, extrasaction="ignore")
+        writer.writeheader()
+        writer.writerows(rows)
+        return f.name
+
+
+@router.post("/")
+async def run_analysis(request: AnalysisRequest, user_id: str = Depends(get_user_id)):
+    """Pull the user's watch data + routes, run the AI crew, store and return the result."""
+    db = await get_db()
+
+    # ── Watch readings + workouts ──────────────────────────────────────────
+    wd_query = db.table("watch_data").select("*").eq("user_id", user_id)
+    if request.date_from:
+        wd_query = wd_query.gte("timestamp", request.date_from.isoformat())
+    if request.date_to:
+        wd_query = wd_query.lte("timestamp", request.date_to.isoformat())
+
+    wd_result = await wd_query.order("timestamp").execute()
+    wd_rows = wd_result.data
+
+    if not wd_rows:
+        raise HTTPException(status_code=404, detail="No watch data found for this user")
+
+    for row in wd_rows:
+        row.pop("user_id", None)
+
+    # ── GPS routes ─────────────────────────────────────────────────────────
+    rt_query = db.table("routes").select(", ".join(_ROUTE_COLUMNS)).eq("user_id", user_id)
+    if request.date_from:
+        rt_query = rt_query.gte("started_at", request.date_from.isoformat())
+    if request.date_to:
+        rt_query = rt_query.lte("started_at", request.date_to.isoformat())
+
+    rt_result = await rt_query.order("started_at").execute()
+    rt_rows = rt_result.data
+
+    # ── Write CSVs ─────────────────────────────────────────────────────────
+    tmp_watch = _write_csv(wd_rows, list(wd_rows[0].keys()))
+    tmp_routes = _write_csv(rt_rows, _ROUTE_COLUMNS) if rt_rows else None
+
+    # Pass both paths (newline-separated) — the crew's task descriptions already
+    # handle multiple files and say "If multiple files, analyze together."
+    data_input = tmp_watch
+    if tmp_routes:
+        data_input = f"{tmp_watch}\n{tmp_routes}"
+
+    try:
+        bots = Bots(context=request.context)
+        loop = asyncio.get_running_loop()
+        raw_json = await loop.run_in_executor(None, lambda: bots.create_crew(data=data_input))
+    except Exception as e:
+        raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
+    finally:
+        if os.path.exists(tmp_watch):
+            os.unlink(tmp_watch)
+        if tmp_routes and os.path.exists(tmp_routes):
+            os.unlink(tmp_routes)
+
+    # Parse the structured FormattedOutput returned by bots.create_crew()
+    parsed: dict = {}
+    try:
+        parsed = json.loads(raw_json)
+    except Exception:
+        parsed = {}
+
+    record = {
+        "user_id": user_id,
+        "context": request.context,
+        "summary": parsed.get("summary") or raw_json,
+        "key_findings": parsed.get("findings") or [],
+        "recommendations": parsed.get("recommendations") or [],
+        "output_type": parsed.get("output_type"),
+        "chart_type": parsed.get("chart_type"),
+        "chart_title": parsed.get("chart_title"),
+        "data_points": parsed.get("data_points"),
+        "metrics": parsed.get("metrics"),
+        "table_headers": parsed.get("table_headers"),
+        "table_rows": parsed.get("table_rows"),
+        "quality_score": parsed.get("quality_score"),
+        "quality_verdict": parsed.get("quality_verdict"),
+        "raw_output": parsed or None,   # full payload — lets frontend render comparison/heatmap/code
+        "created_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+    # output_type must always be present so the frontend can pick the right renderer
+    if not record.get("output_type"):
+        record["output_type"] = "report"
+
+    # Strip remaining None values so Supabase doesn't complain about unknown columns
+    record = {k: v for k, v in record.items() if v is not None}
+
+    await db.table("analyses").insert(record).execute()
+    return record

backend/routes/charts.py

@@ -0,0 +1,94 @@
+from fastapi import APIRouter, Depends
+from auth import get_user_id
+from db import get_db
+from collections import defaultdict
+from datetime import datetime
+
+router = APIRouter()
+
+
+def date_key(iso: str) -> str:
+    return iso[:10] if iso else ""
+
+
+@router.get("/")
+async def get_chart_data(user_id: str = Depends(get_user_id)):
+    db = await get_db()
+
+    workouts_res = await (
+        db.table("watch_data")
+        .select("*")
+        .eq("user_id", user_id)
+        .eq("type", "workout")
+        .order("timestamp")
+        .execute()
+    )
+    readings_res = await (
+        db.table("watch_data")
+        .select("timestamp,heart_rate,steps,calories_burned,sleep_hours,spo2,hrv")
+        .eq("user_id", user_id)
+        .eq("type", "reading")
+        .order("timestamp")
+        .execute()
+    )
+
+    workouts = workouts_res.data or []
+    readings = readings_res.data or []
+
+    # calories burned per day (workouts)
+    cal_by_day: dict = defaultdict(float)
+    for w in workouts:
+        if w.get("calories_burned"):
+            cal_by_day[date_key(w["timestamp"])] += w["calories_burned"]
+
+    # avg heart rate per day (readings)
+    hr_by_day: dict = defaultdict(list)
+    for r in readings:
+        if r.get("heart_rate"):
+            hr_by_day[date_key(r["timestamp"])].append(r["heart_rate"])
+    hr_avg_by_day = {d: round(sum(v) / len(v), 1) for d, v in hr_by_day.items()}
+
+    # steps per day
+    steps_by_day: dict = defaultdict(int)
+    for r in readings:
+        if r.get("steps"):
+            steps_by_day[date_key(r["timestamp"])] += r["steps"]
+
+    # sleep hours per day
+    sleep_by_day: dict = {}
+    for r in readings:
+        if r.get("sleep_hours"):
+            sleep_by_day[date_key(r["timestamp"])] = r["sleep_hours"]
+
+    # workout type breakdown
+    type_counts: dict = defaultdict(int)
+    for w in workouts:
+        if w.get("workout_type"):
+            type_counts[w["workout_type"]] += 1
+
+    # recent workouts list (last 10)
+    recent = sorted(workouts, key=lambda x: x.get("timestamp", ""), reverse=True)[:10]
+
+    return {
+        "calories": {
+            "labels": sorted(cal_by_day.keys()),
+            "values": [cal_by_day[d] for d in sorted(cal_by_day.keys())],
+        },
+        "heart_rate": {
+            "labels": sorted(hr_avg_by_day.keys()),
+            "values": [hr_avg_by_day[d] for d in sorted(hr_avg_by_day.keys())],
+        },
+        "steps": {
+            "labels": sorted(steps_by_day.keys()),
+            "values": [steps_by_day[d] for d in sorted(steps_by_day.keys())],
+        },
+        "sleep": {
+            "labels": sorted(sleep_by_day.keys()),
+            "values": [sleep_by_day[d] for d in sorted(sleep_by_day.keys())],
+        },
+        "workout_types": {
+            "labels": list(type_counts.keys()),
+            "values": list(type_counts.values()),
+        },
+        "recent_workouts": recent,
+    }

backend/routes/gps.py

@@ -0,0 +1,159 @@
+from fastapi import APIRouter, Depends, HTTPException
+from pydantic import BaseModel
+from typing import Optional
+from datetime import datetime, timezone
+from auth import get_user_id
+from db import get_db
+from calories import estimate_calories
+import math
+
+router = APIRouter()
+
+
+class Coordinate(BaseModel):
+    lat: float
+    lng: float
+    timestamp: str
+    altitude: Optional[float] = None
+    speed: Optional[float] = None
+
+
+class RoutePayload(BaseModel):
+    workout_type: str
+    coordinates: list[Coordinate]
+    started_at: str
+    ended_at: str
+    # Sent by the watch so distance/pace/calories are all computed here, not on
+    # the device.
+    avg_heart_rate: Optional[float] = None
+    max_heart_rate: Optional[float] = None
+    # True when the client posts ONLY a route (no separate workout) — the watch
+    # and the web GPS tracker. Tells us to mirror a workout row into watch_data.
+    # Clients that post their own workout (manual entry, Strava/Nike/Garmin
+    # imports) leave this false so the workout isn't counted twice.
+    record_workout: bool = False
+    notes: Optional[str] = None
+
+
+def haversine(lat1, lng1, lat2, lng2) -> float:
+    R = 6371000
+    p = math.pi / 180
+    a = (math.sin((lat2 - lat1) * p / 2) ** 2 +
+         math.cos(lat1 * p) * math.cos(lat2 * p) *
+         math.sin((lng2 - lng1) * p / 2) ** 2)
+    return 2 * R * math.asin(math.sqrt(a))
+
+
+def calc_distance(coords: list[Coordinate]) -> float:
+    total = 0.0
+    for i in range(1, len(coords)):
+        total += haversine(coords[i-1].lat, coords[i-1].lng, coords[i].lat, coords[i].lng)
+    return round(total, 1)
+
+
+def calc_duration(started_at: str, ended_at: str) -> int:
+    try:
+        start = datetime.fromisoformat(started_at.replace("Z", "+00:00"))
+        end   = datetime.fromisoformat(ended_at.replace("Z", "+00:00"))
+        return max(0, int((end - start).total_seconds()))
+    except Exception:
+        return 0
+
+
+def calc_pace(distance_m: float, duration_s: int) -> Optional[str]:
+    if distance_m < 1 or duration_s < 1:
+        return None
+    mins_per_km = (duration_s / 60) / (distance_m / 1000)
+    m = int(mins_per_km)
+    s = int((mins_per_km - m) * 60)
+    return f"{m}:{s:02d} /km"
+
+
+@router.post("/")
+async def save_route(payload: RoutePayload, user_id: str = Depends(get_user_id)):
+    """Save a completed GPS route."""
+    if len(payload.coordinates) < 2:
+        raise HTTPException(status_code=400, detail="Route needs at least 2 coordinates")
+
+    db = await get_db()
+    distance = calc_distance(payload.coordinates)
+    duration = calc_duration(payload.started_at, payload.ended_at)
+    pace = calc_pace(distance, duration)
+    duration_min = round(duration / 60.0, 2)
+
+    calories = estimate_calories(duration_min, payload.workout_type)
+
+    record = {
+        "user_id": user_id,
+        "workout_type": payload.workout_type,
+        "coordinates": [c.model_dump() for c in payload.coordinates],
+        "distance_meters": distance,
+        "duration_seconds": duration,
+        "pace": pace,
+        "calories_burned": calories,
+        "started_at": payload.started_at,
+        "ended_at": payload.ended_at,
+        "notes": payload.notes,
+        "created_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+    result = await db.table("routes").insert(record).execute()
+    if not result.data:
+        raise HTTPException(status_code=500, detail="Route insert failed")
+
+    # Route-only clients (the watch, the web GPS tracker) ask us to also record
+    # the workout — distance, pace and calories all computed here, server-side,
+    # so the device doesn't have to. Clients that post their own workout leave
+    # record_workout false to avoid double-counting.
+    if payload.record_workout:
+        workout_row = {
+            "user_id":          user_id,
+            "type":             "workout",
+            "device":           "gps_route",
+            "timestamp":        payload.started_at,
+            "workout_type":     payload.workout_type,
+            "duration_minutes": duration_min,
+            "avg_heart_rate":   payload.avg_heart_rate,
+            "max_heart_rate":   payload.max_heart_rate,
+            "calories_burned":  calories,
+            "distance_meters":  distance,
+        }
+        try:
+            await db.table("watch_data").insert(workout_row).execute()
+        except Exception as e:
+            # The route saved fine; don't fail the request if the mirror does.
+            print(f"[routes] workout mirror insert failed: {e}")
+
+    return result.data[0]
+
+
+@router.get("/")
+async def get_routes(user_id: str = Depends(get_user_id), limit: int = 20):
+    """Fetch the user's past routes, newest first."""
+    db = await get_db()
+    result = await (
+        db.table("routes")
+        .select("*")
+        .eq("user_id", user_id)
+        .order("started_at", desc=True)
+        .limit(limit)
+        .execute()
+    )
+    return {"routes": result.data}
+
+
+@router.get("/{route_id}")
+async def get_route(route_id: str, user_id: str = Depends(get_user_id)):
+    """Fetch a single route by ID."""
+    db = await get_db()
+    result = await (
+        db.table("routes")
+        .select("*")
+        .eq("id", route_id)
+        .eq("user_id", user_id)
+        .single()
+        .execute()
+    )
+    if not result.data:
+        raise HTTPException(status_code=404, detail="Route not found")
+    return result.data

backend/routes/integrations.py

@@ -0,0 +1,1186 @@
+import base64
+import csv
+import hashlib
+import hmac
+import io
+import json
+import math
+import os
+import urllib.parse
+import xml.etree.ElementTree as ET
+import zipfile
+
+import httpx
+from fastapi import APIRouter, Depends, File, HTTPException, UploadFile
+from fastapi.responses import RedirectResponse
+from auth import get_user_id
+from db import get_db
+from datetime import datetime, timezone, timedelta
+
+router = APIRouter()
+
+STRAVA_CLIENT_ID     = os.getenv("STRAVA_CLIENT_ID", "")
+STRAVA_CLIENT_SECRET = os.getenv("STRAVA_CLIENT_SECRET", "")
+BACKEND_URL          = os.getenv("BACKEND_URL", "http://localhost:8000")
+FRONTEND_URL         = os.getenv("FRONTEND_URL", "http://localhost:5173")
+
+_OAUTH_SECRET = os.getenv("OAUTH_STATE_SECRET", os.urandom(32).hex())
+
+STRAVA_AUTH_URL    = "https://www.strava.com/oauth/authorize"
+STRAVA_TOKEN_URL   = "https://www.strava.com/oauth/token"
+STRAVA_ACTIVITIES  = "https://www.strava.com/api/v3/athlete/activities"
+STRAVA_STREAMS_URL = "https://www.strava.com/api/v3/activities/{id}/streams"
+
+NRC_TYPE_MAP = {
+    "nike.run.gps":       "running",
+    "nike.run.manual":    "running",
+    "nike.run.treadmill": "running",
+    "nike.sport.walk":    "walking",
+    "nike.sport.hike":    "hiking",
+    "nike.sport.cycle":   "cycling",
+    "nike.sport.swim":    "swimming",
+    "run":   "running",
+    "walk":  "walking",
+    "hike":  "hiking",
+    "cycle": "cycling",
+    "swim":  "swimming",
+}
+
+STRAVA_TYPE_MAP = {
+    "Run": "running", "VirtualRun": "running",
+    "Ride": "cycling", "VirtualRide": "cycling", "EBikeRide": "cycling",
+    "Walk": "walking",
+    "Hike": "hiking",
+    "WeightTraining": "weightlifting",
+    "Swim": "swimming",
+    "Workout": "other", "Crossfit": "other", "Elliptical": "other",
+    "RockClimbing": "hiking", "Yoga": "other", "Pilates": "other",
+}
+
+
+@router.get("/status")
+async def integration_status(user_id: str = Depends(get_user_id)):
+    db = await get_db()
+    result = await (
+        db.table("user_integrations")
+        .select("provider")
+        .eq("user_id", user_id)
+        .execute()
+    )
+    return {row["provider"]: True for row in (result.data or [])}
+
+
+def _make_state(user_id: str) -> str:
+    sig = hmac.new(_OAUTH_SECRET.encode(), user_id.encode(), hashlib.sha256).hexdigest()
+    return f"{user_id}.{sig}"
+
+
+def _verify_state(state: str) -> str | None:
+    """Return user_id if state is valid, else None."""
+    try:
+        user_id, sig = state.rsplit(".", 1)
+    except ValueError:
+        return None
+    expected = hmac.new(_OAUTH_SECRET.encode(), user_id.encode(), hashlib.sha256).hexdigest()
+    if hmac.compare_digest(sig, expected):
+        return user_id
+    return None
+
+
+@router.get("/strava/connect")
+async def strava_connect(user_id: str = Depends(get_user_id)):
+    callback = f"{BACKEND_URL}/integrations/strava/callback"
+    url = (
+        f"{STRAVA_AUTH_URL}"
+        f"?client_id={STRAVA_CLIENT_ID}"
+        f"&redirect_uri={callback}"
+        f"&response_type=code"
+        f"&scope=activity:read_all"
+        f"&state={_make_state(user_id)}"
+        f"&approval_prompt=auto"
+    )
+    return {"url": url}
+
+
+@router.get("/strava/callback")
+async def strava_callback(code: str = "", state: str = "", error: str = ""):
+    if error or not code:
+        return RedirectResponse(f"{FRONTEND_URL}/log?error={error or 'access_denied'}")
+
+    user_id = _verify_state(state)
+    if not user_id:
+        return RedirectResponse(f"{FRONTEND_URL}/log?error=invalid_state")
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(STRAVA_TOKEN_URL, data={
+            "client_id":     STRAVA_CLIENT_ID,
+            "client_secret": STRAVA_CLIENT_SECRET,
+            "code":          code,
+            "grant_type":    "authorization_code",
+        })
+        if resp.status_code != 200:
+            return RedirectResponse(f"{FRONTEND_URL}/log?error=token_exchange_failed")
+        data = resp.json()
+
+    db = await get_db()
+    row = {
+        "user_id":       user_id,
+        "provider":      "strava",
+        "access_token":  data["access_token"],
+        "refresh_token": data.get("refresh_token"),
+        "expires_at":    datetime.fromtimestamp(data["expires_at"], tz=timezone.utc).isoformat(),
+        "athlete_id":    str(data.get("athlete", {}).get("id", "")),
+        "updated_at":    datetime.now(timezone.utc).isoformat(),
+    }
+    await db.table("user_integrations").upsert(row, on_conflict="user_id,provider").execute()
+    return RedirectResponse(f"{FRONTEND_URL}/log?connected=strava")
+
+
+@router.post("/strava/sync")
+async def strava_sync(user_id: str = Depends(get_user_id)):
+    """Fetch latest Strava activities. Saves workouts to watch_data and GPS routes to routes table."""
+    db = await get_db()
+
+    integration = await (
+        db.table("user_integrations")
+        .select("*")
+        .eq("user_id", user_id)
+        .eq("provider", "strava")
+        .single()
+        .execute()
+    )
+    if not integration.data:
+        return {"synced": 0, "routes_saved": 0, "error": "Strava not connected"}
+
+    token = await _refresh_strava_token_if_needed(db, integration.data)
+
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.get(
+            STRAVA_ACTIVITIES,
+            headers={"Authorization": f"Bearer {token}"},
+            params={"per_page": 30},
+        )
+        if resp.status_code != 200:
+            return {"synced": 0, "routes_saved": 0, "error": "Failed to fetch Strava activities"}
+        activities = resp.json()
+
+        workout_rows = []
+        route_rows   = []
+
+        for a in activities:
+            start_date = a.get("start_date", "")
+            if not start_date:
+                continue  # skip activities without a date — we can't place them in time
+            workout_type = STRAVA_TYPE_MAP.get(a.get("type", ""), "other")
+
+            # ── Workout record ──────────────────────────────────────────
+            workout_rows.append({
+                "user_id":            user_id,
+                "type":               "workout",
+                "device":             "strava",
+                "timestamp":          start_date,
+                "workout_type":       workout_type,
+                "duration_minutes":   round(a.get("moving_time", 0) / 60, 1),
+                "distance_meters":    a.get("distance"),
+                "calories_burned":    a.get("calories"),
+                "avg_heart_rate":     a.get("average_heartrate"),
+                "max_heart_rate":     a.get("max_heartrate"),
+                "ending_heart_rate":  None,   # not in activity summary; streams would be needed
+                "notes":              a.get("name"),
+            })
+
+            # ── GPS route — only for activities that have a map ─────────
+            if a.get("map", {}).get("summary_polyline"):
+                streams_resp = await client.get(
+                    STRAVA_STREAMS_URL.format(id=a["id"]),
+                    headers={"Authorization": f"Bearer {token}"},
+                    params={"keys": "latlng,time,heartrate", "key_by_type": "true"},
+                )
+                if streams_resp.status_code == 200:
+                    streams   = streams_resp.json()
+                    latlng    = streams.get("latlng",    {}).get("data", [])
+                    times     = streams.get("time",      {}).get("data", [])
+                    hr_stream = streams.get("heartrate", {}).get("data", [])
+
+                    if len(latlng) >= 2:
+                        try:
+                            start_dt = datetime.fromisoformat(start_date.replace("Z", "+00:00"))
+                        except Exception:
+                            continue  # can't build a route without a valid start time
+
+                        elapsed  = a.get("elapsed_time", 0)
+                        ended_dt = start_dt + timedelta(seconds=elapsed)
+
+                        coords = []
+                        for i, pt in enumerate(latlng):
+                            offset_s = times[i] if i < len(times) else 0
+                            ts = (start_dt + timedelta(seconds=offset_s)).isoformat()
+                            c = {"lat": pt[0], "lng": pt[1], "timestamp": ts}
+                            if i < len(hr_stream) and hr_stream[i]:
+                                c["heart_rate"] = hr_stream[i]
+                            coords.append(c)
+
+                        # ending HR = last HR value in stream
+                        ending_hr = None
+                        for val in reversed(hr_stream):
+                            if val:
+                                ending_hr = val
+                                break
+                        # Back-fill ending_heart_rate onto the workout row
+                        workout_rows[-1]["ending_heart_rate"] = ending_hr
+
+                        route_rows.append({
+                            "user_id":          user_id,
+                            "workout_type":     workout_type,
+                            "coordinates":      coords,
+                            "distance_meters":  a.get("distance"),
+                            "duration_seconds": a.get("moving_time"),
+                            "started_at":       start_date,
+                            "ended_at":         ended_dt.isoformat(),
+                            "notes":            a.get("name"),
+                        })
+
+    if workout_rows:
+        await db.table("watch_data").insert(workout_rows).execute()
+    if route_rows:
+        await db.table("routes").insert(route_rows).execute()
+
+    return {
+        "synced":       len(workout_rows),
+        "routes_saved": len(route_rows),
+    }
+
+
+@router.post("/nike/import")
+async def nike_import(
+    file: UploadFile = File(...),
+    user_id: str = Depends(get_user_id),
+):
+    """Parse a Nike Run Club data-export ZIP or JSON and save activities to watch_data."""
+    content = await file.read()
+    activities: list[dict] = []
+
+    filename = (file.filename or "").lower()
+    if filename.endswith(".zip"):
+        try:
+            with zipfile.ZipFile(io.BytesIO(content)) as zf:
+                for name in zf.namelist():
+                    if not name.lower().endswith(".json"):
+                        continue
+                    with zf.open(name) as f:
+                        try:
+                            data = json.load(f)
+                            if isinstance(data, list):
+                                activities.extend(data)
+                            elif isinstance(data, dict) and ("startEpochMs" in data or "type" in data):
+                                activities.append(data)
+                        except Exception:
+                            pass
+        except zipfile.BadZipFile:
+            return {"error": "Invalid ZIP file", "imported": 0, "routes_saved": 0}
+    else:
+        try:
+            data = json.loads(content)
+            activities = data if isinstance(data, list) else [data]
+        except Exception:
+            return {"error": "Invalid JSON file", "imported": 0, "routes_saved": 0}
+
+    workout_rows: list[dict] = []
+    route_rows:   list[dict] = []
+
+    for a in activities:
+        # ── Timestamp ─────────────────────────────────────────────────────
+        start_ms = a.get("startEpochMs") or a.get("start_epoch_ms")
+        if start_ms:
+            start_dt = datetime.fromtimestamp(start_ms / 1000, tz=timezone.utc)
+        else:
+            continue  # skip entries with no timestamp — we can't place them in time
+
+        # ── Duration ──────────────────────────────────────────────────────
+        active_ms   = a.get("activeTime") or a.get("active_duration_ms") or 0
+        duration_min = round(active_ms / 60000, 1) if active_ms else None
+
+        # ── Activity type ─────────────────────────────────────────────────
+        workout_type = NRC_TYPE_MAP.get(a.get("type", "").lower(), "other")
+
+        # ── Tags (NRC stores extra fields here) ───────────────────────────
+        tags = a.get("tags", {})
+
+        def _tag(key: str):
+            return tags.get(key) or tags.get(f"com.nike.{key}")
+
+        # ── Distance ──────────────────────────────────────────────────────
+        distance_m = None
+        dist_raw = _tag("distance") or a.get("distance")
+        if dist_raw is not None:
+            try:
+                if isinstance(dist_raw, dict):
+                    val  = float(dist_raw.get("value", 0))
+                    unit = dist_raw.get("unit", "KM").upper()
+                    distance_m = round(val * 1000 if unit in ("KM", "KILOMETERS") else val * 1609.34, 1)
+                else:
+                    distance_m = round(float(dist_raw) * 1000, 1)  # NRC tags are in km
+            except (ValueError, TypeError):
+                pass
+
+        # ── Calories ──────────────────────────────────────────────────────
+        calories = None
+        cal_raw = _tag("calories") or a.get("calories")
+        if cal_raw is not None:
+            try:
+                calories = float(cal_raw.get("value", 0) if isinstance(cal_raw, dict) else cal_raw)
+            except (ValueError, TypeError):
+                pass
+
+        # ── Heart rate + GPS from metrics ─────────────────────────────────
+        avg_hr = max_hr = ending_hr = None
+        lat_entries: list[dict] = []
+        lng_entries: list[dict] = []
+
+        for metric in a.get("metrics", []):
+            mtype  = metric.get("type", "").upper()
+            values = metric.get("values", [])
+            if not values:
+                continue
+
+            if mtype == "HEART_RATE":
+                hr_vals = [v["value"] for v in values if v.get("value") is not None]
+                if hr_vals:
+                    avg_hr    = round(sum(hr_vals) / len(hr_vals))
+                    max_hr    = int(max(hr_vals))
+                    ending_hr = int(hr_vals[-1])
+            elif mtype == "LATITUDE":
+                lat_entries = values
+            elif mtype == "LONGITUDE":
+                lng_entries = values
+
+        # Also check summaries block for distance/calories if still missing
+        for s in a.get("summaries", []):
+            stype = s.get("metric", "").upper()
+            val   = s.get("value")
+            if val is None:
+                continue
+            if stype == "DISTANCE" and distance_m is None:
+                distance_m = round(float(val) * 1000, 1)
+            elif stype == "CALORIES" and calories is None:
+                calories = float(val)
+
+        # ── Workout row ───────────────────────────────────────────────────
+        workout_rows.append({
+            "user_id":           user_id,
+            "type":              "workout",
+            "device":            "nike_run_club",
+            "timestamp":         start_dt.isoformat(),
+            "workout_type":      workout_type,
+            "duration_minutes":  duration_min,
+            "distance_meters":   distance_m,
+            "calories_burned":   calories,
+            "avg_heart_rate":    avg_hr,
+            "max_heart_rate":    max_hr,
+            "ending_heart_rate": ending_hr,
+            "notes":             _tag("name") or a.get("name") or a.get("title"),
+        })
+
+        # ── GPS route ─────────────────────────────────────────────────────
+        if len(lat_entries) >= 2 and len(lng_entries) >= 2:
+            coords = []
+            for i, lat_entry in enumerate(lat_entries):
+                if i >= len(lng_entries):
+                    break
+                ts_ms = lat_entry.get("startEpochMs") or start_ms
+                coords.append({
+                    "lat":       lat_entry.get("value"),
+                    "lng":       lng_entries[i].get("value"),
+                    "timestamp": datetime.fromtimestamp(ts_ms / 1000, tz=timezone.utc).isoformat(),
+                })
+            if len(coords) >= 2:
+                ended_dt = start_dt + timedelta(milliseconds=active_ms or 0)
+                route_rows.append({
+                    "user_id":          user_id,
+                    "workout_type":     workout_type,
+                    "coordinates":      coords,
+                    "distance_meters":  distance_m,
+                    "duration_seconds": active_ms // 1000 if active_ms else None,
+                    "started_at":       start_dt.isoformat(),
+                    "ended_at":         ended_dt.isoformat(),
+                    "notes":            _tag("name") or a.get("name"),
+                })
+
+    db = await get_db()
+    if workout_rows:
+        await db.table("watch_data").insert(workout_rows).execute()
+    if route_rows:
+        await db.table("routes").insert(route_rows).execute()
+
+    return {"imported": len(workout_rows), "routes_saved": len(route_rows)}
+
+
+async def _refresh_strava_token_if_needed(db, integration: dict) -> str:
+    expires_at = datetime.fromisoformat(integration["expires_at"])
+    if datetime.now(timezone.utc) < expires_at:
+        return integration["access_token"]
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(STRAVA_TOKEN_URL, data={
+            "client_id":     STRAVA_CLIENT_ID,
+            "client_secret": STRAVA_CLIENT_SECRET,
+            "refresh_token": integration["refresh_token"],
+            "grant_type":    "refresh_token",
+        })
+
+    if resp.status_code != 200:
+        raise HTTPException(status_code=502, detail="Failed to refresh Strava token")
+
+    data = resp.json()
+    if "access_token" not in data:
+        raise HTTPException(status_code=502, detail="Strava token response missing access_token")
+
+    await (
+        db.table("user_integrations")
+        .update({
+            "access_token":  data["access_token"],
+            "refresh_token": data.get("refresh_token", integration["refresh_token"]),
+            "expires_at":    datetime.fromtimestamp(data["expires_at"], tz=timezone.utc).isoformat(),
+            "updated_at":    datetime.now(timezone.utc).isoformat(),
+        })
+        .eq("id", integration["id"])
+        .execute()
+    )
+    return data["access_token"]
+
+
+# ── Shared helpers ────────────────────────────────────────────────────────────
+
+def _haversine(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
+    R = 6371000
+    p = math.pi / 180
+    a = (math.sin((lat2 - lat1) * p / 2) ** 2 +
+         math.cos(lat1 * p) * math.cos(lat2 * p) *
+         math.sin((lng2 - lng1) * p / 2) ** 2)
+    return 2 * R * math.asin(math.sqrt(max(0.0, min(1.0, a))))
+
+
+def _classify_activity(name: str) -> str:
+    n = name.lower()
+    if any(w in n for w in ("run", "jog")):           return "running"
+    if any(w in n for w in ("cycl", "bike", "rid")):  return "cycling"
+    if any(w in n for w in ("walk",)):                 return "walking"
+    if any(w in n for w in ("hike", "trail")):         return "hiking"
+    if any(w in n for w in ("swim",)):                 return "swimming"
+    if any(w in n for w in ("weight", "strength", "lift", "gym")): return "weightlifting"
+    return "other"
+
+
+def _strip_ns(root: ET.Element) -> ET.Element:
+    """Remove XML namespace prefixes so we can find tags by local name."""
+    for el in root.iter():
+        if "}" in el.tag:
+            el.tag = el.tag.split("}", 1)[1]
+    return root
+
+
+def _parse_apple_date(s: str) -> datetime:
+    s = s.strip()
+    for fmt in ("%Y-%m-%d %H:%M:%S %z", "%Y-%m-%dT%H:%M:%SZ", "%Y-%m-%dT%H:%M:%S%z"):
+        try:
+            return datetime.strptime(s, fmt)
+        except ValueError:
+            pass
+    raise ValueError(f"Cannot parse Apple Health date: {s!r}")
+
+
+# ── Google Health API (Fitbit wearable data via Google OAuth) ─────────────────
+# Fitbit Web API was deprecated; new apps register at Google Cloud Console.
+# Docs: https://developers.google.com/health/api
+
+GOOGLE_HEALTH_CLIENT_ID     = os.getenv("GOOGLE_HEALTH_CLIENT_ID", "")
+GOOGLE_HEALTH_CLIENT_SECRET = os.getenv("GOOGLE_HEALTH_CLIENT_SECRET", "")
+GOOGLE_AUTH_URL             = "https://accounts.google.com/o/oauth2/v2/auth"
+GOOGLE_TOKEN_URL            = "https://oauth2.googleapis.com/token"
+GOOGLE_HEALTH_BASE          = "https://health.googleapis.com/v4"
+
+GOOGLE_HEALTH_SCOPES = " ".join([
+    "https://www.googleapis.com/auth/googlehealth.activity_and_fitness.readonly",
+    "https://www.googleapis.com/auth/googlehealth.health_metrics_and_measurements.readonly",
+    "https://www.googleapis.com/auth/googlehealth.sleep.readonly",
+])
+
+GOOGLE_EXERCISE_TYPE_MAP = {
+    "RUNNING":           "running",
+    "WALKING":           "walking",
+    "BIKING":            "cycling",
+    "SWIMMING":          "swimming",
+    "HIKING":            "hiking",
+    "YOGA":              "other",
+    "PILATES":           "other",
+    "WORKOUT":           "other",
+    "HIIT":              "other",
+    "WEIGHTLIFTING":     "weightlifting",
+    "STRENGTH_TRAINING": "weightlifting",
+    "OTHER":             "other",
+}
+
+
+def _parse_google_time(t) -> str | None:
+    """Accept RFC 3339 string or proto Timestamp dict {seconds, nanos}."""
+    if isinstance(t, str):
+        return t.replace("Z", "+00:00")
+    if isinstance(t, dict):
+        secs = t.get("seconds") or t.get("epochSeconds")
+        if secs:
+            return datetime.fromtimestamp(int(secs), tz=timezone.utc).isoformat()
+    return None
+
+
+@router.get("/fitbit/connect")
+async def fitbit_connect(user_id: str = Depends(get_user_id)):
+    callback = f"{BACKEND_URL}/integrations/fitbit/callback"
+    url = (
+        f"{GOOGLE_AUTH_URL}"
+        f"?client_id={GOOGLE_HEALTH_CLIENT_ID}"
+        f"&response_type=code"
+        f"&scope={urllib.parse.quote(GOOGLE_HEALTH_SCOPES)}"
+        f"&redirect_uri={urllib.parse.quote(callback, safe='')}"
+        f"&state={_make_state(user_id)}"
+        f"&access_type=offline"
+        f"&prompt=consent"
+    )
+    return {"url": url}
+
+
+@router.get("/fitbit/callback")
+async def fitbit_callback(code: str = "", state: str = "", error: str = ""):
+    if error or not code:
+        return RedirectResponse(f"{FRONTEND_URL}/log?error={error or 'access_denied'}")
+    user_id = _verify_state(state)
+    if not user_id:
+        return RedirectResponse(f"{FRONTEND_URL}/log?error=invalid_state")
+
+    callback = f"{BACKEND_URL}/integrations/fitbit/callback"
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(GOOGLE_TOKEN_URL, data={
+            "code":          code,
+            "grant_type":    "authorization_code",
+            "redirect_uri":  callback,
+            "client_id":     GOOGLE_HEALTH_CLIENT_ID,
+            "client_secret": GOOGLE_HEALTH_CLIENT_SECRET,
+        })
+        if resp.status_code != 200:
+            return RedirectResponse(f"{FRONTEND_URL}/log?error=token_exchange_failed")
+        data = resp.json()
+
+    db = await get_db()
+    await db.table("user_integrations").upsert({
+        "user_id":       user_id,
+        "provider":      "fitbit",
+        "access_token":  data["access_token"],
+        "refresh_token": data.get("refresh_token"),
+        "expires_at":    (datetime.now(timezone.utc) + timedelta(seconds=data.get("expires_in", 3600))).isoformat(),
+        "updated_at":    datetime.now(timezone.utc).isoformat(),
+    }, on_conflict="user_id,provider").execute()
+    return RedirectResponse(f"{FRONTEND_URL}/log?connected=fitbit")
+
+
+@router.post("/fitbit/sync")
+async def fitbit_sync(user_id: str = Depends(get_user_id)):
+    db = await get_db()
+    integration = await (
+        db.table("user_integrations").select("*")
+        .eq("user_id", user_id).eq("provider", "fitbit").single().execute()
+    )
+    if not integration.data:
+        return {"synced": 0, "error": "Fitbit not connected"}
+
+    token   = await _refresh_fitbit_token_if_needed(db, integration.data)
+    after   = (datetime.now(timezone.utc) - timedelta(days=90)).strftime("%Y-%m-%dT%H:%M:%SZ")
+    headers = {"Authorization": f"Bearer {token}"}
+
+    async with httpx.AsyncClient(timeout=30) as client:
+        ex_resp = await client.get(
+            f"{GOOGLE_HEALTH_BASE}/users/me/dataTypes/exercise/dataPoints",
+            headers=headers,
+            params={"filter": f'exercise.interval.startTime >= "{after}"', "pageSize": 100},
+        )
+        sl_resp = await client.get(
+            f"{GOOGLE_HEALTH_BASE}/users/me/dataTypes/sleep/dataPoints",
+            headers=headers,
+            params={"filter": f'sleep.interval.startTime >= "{after}"', "pageSize": 100},
+        )
+
+    exercises = ex_resp.json().get("dataPoints", []) if ex_resp.status_code == 200 else []
+    sleeps    = sl_resp.json().get("dataPoints", []) if sl_resp.status_code == 200 else []
+
+    rows: list[dict] = []
+
+    for pt in exercises:
+        ex  = (pt.get("data") or {}).get("exercise", {})
+        ivl = ex.get("interval", {})
+        start = _parse_google_time(ivl.get("startTime"))
+        if not start:
+            continue
+        try:
+            start_dt = datetime.fromisoformat(start)
+        except ValueError:
+            continue
+
+        dur_min = None
+        active  = ex.get("activeDuration", "")
+        if isinstance(active, str) and active.endswith("s"):
+            try:
+                dur_min = round(float(active[:-1]) / 60, 1)
+            except ValueError:
+                pass
+        if dur_min is None:
+            end_str = _parse_google_time(ivl.get("endTime"))
+            if end_str:
+                try:
+                    dur_min = round((datetime.fromisoformat(end_str) - start_dt).total_seconds() / 60, 1)
+                except ValueError:
+                    pass
+
+        metrics = ex.get("metricsSummary", {})
+        dist_mm = metrics.get("distanceMillimeters")
+        rows.append({
+            "user_id":          user_id,
+            "type":             "workout",
+            "device":           "fitbit",
+            "timestamp":        start_dt.isoformat(),
+            "workout_type":     GOOGLE_EXERCISE_TYPE_MAP.get(ex.get("exerciseType", ""), "other"),
+            "duration_minutes": dur_min,
+            "distance_meters":  round(float(dist_mm) / 1000, 1) if dist_mm else None,
+            "calories_burned":  metrics.get("caloriesKcal"),
+            "avg_heart_rate":   metrics.get("averageHeartRateBeatsPerMinute"),
+        })
+
+    for pt in sleeps:
+        sl    = (pt.get("data") or {}).get("sleep", {})
+        ivl   = sl.get("interval", {})
+        start = _parse_google_time(ivl.get("startTime"))
+        end   = _parse_google_time(ivl.get("endTime"))
+        if not start:
+            continue
+        dur_hrs = None
+        if end:
+            try:
+                s = datetime.fromisoformat(start)
+                e = datetime.fromisoformat(end)
+                dur_hrs = round((e - s).total_seconds() / 3600, 2)
+            except ValueError:
+                pass
+        rows.append({
+            "user_id":     user_id,
+            "type":        "reading",
+            "device":      "fitbit",
+            "timestamp":   start,
+            "sleep_hours": dur_hrs,
+        })
+
+    if rows:
+        await db.table("watch_data").insert(rows).execute()
+    return {
+        "synced":       len([r for r in rows if r["type"] == "workout"]),
+        "sleep_synced": len([r for r in rows if r["type"] == "reading"]),
+    }
+
+
+async def _refresh_fitbit_token_if_needed(db, integration: dict) -> str:
+    expires_at = datetime.fromisoformat(integration["expires_at"])
+    if datetime.now(timezone.utc) < expires_at:
+        return integration["access_token"]
+
+    async with httpx.AsyncClient() as client:
+        resp = await client.post(GOOGLE_TOKEN_URL, data={
+            "grant_type":    "refresh_token",
+            "refresh_token": integration["refresh_token"],
+            "client_id":     GOOGLE_HEALTH_CLIENT_ID,
+            "client_secret": GOOGLE_HEALTH_CLIENT_SECRET,
+        })
+    if resp.status_code != 200:
+        raise HTTPException(status_code=502, detail="Failed to refresh Fitbit token")
+
+    data = resp.json()
+    await (
+        db.table("user_integrations").update({
+            "access_token":  data["access_token"],
+            "refresh_token": data.get("refresh_token", integration["refresh_token"]),
+            "expires_at":    (datetime.now(timezone.utc) + timedelta(seconds=data.get("expires_in", 3600))).isoformat(),
+            "updated_at":    datetime.now(timezone.utc).isoformat(),
+        }).eq("id", integration["id"]).execute()
+    )
+    return data["access_token"]
+
+
+# ── Garmin (GPX / TCX file import) ───────────────────────────────────────────
+
+@router.post("/garmin/import")
+async def garmin_import(file: UploadFile = File(...), user_id: str = Depends(get_user_id)):
+    content  = await file.read()
+    filename = (file.filename or "").lower()
+    if filename.endswith(".gpx"):
+        workout_rows, route_rows = _parse_gpx(content, user_id)
+    elif filename.endswith(".tcx"):
+        workout_rows, route_rows = _parse_tcx(content, user_id)
+    else:
+        return {"error": "Upload a .gpx or .tcx file exported from Garmin Connect.", "imported": 0, "routes_saved": 0}
+
+    db = await get_db()
+    if workout_rows:
+        await db.table("watch_data").insert(workout_rows).execute()
+    if route_rows:
+        await db.table("routes").insert(route_rows).execute()
+    return {"imported": len(workout_rows), "routes_saved": len(route_rows)}
+
+
+def _parse_gpx(content: bytes, user_id: str) -> tuple[list, list]:
+    try:
+        root = _strip_ns(ET.fromstring(content))
+    except ET.ParseError:
+        return [], []
+
+    workout_rows, route_rows = [], []
+
+    for trk in root.findall(".//trk"):
+        name_el = trk.find("name")
+        name    = name_el.text if name_el is not None else None
+        type_el = trk.find("type")
+        type_hint = (type_el.text if type_el is not None else name) or ""
+        workout_type = _classify_activity(type_hint)
+
+        coords, hr_vals = [], []
+        for pt in trk.findall(".//trkpt"):
+            lat, lon = pt.get("lat"), pt.get("lon")
+            time_el  = pt.find("time")
+            if lat is None or lon is None or time_el is None:
+                continue
+            coord: dict = {"lat": float(lat), "lng": float(lon), "timestamp": time_el.text}
+            ele_el = pt.find("ele")
+            if ele_el is not None:
+                try:
+                    coord["altitude"] = float(ele_el.text)
+                except (ValueError, TypeError):
+                    pass
+            hr_el = pt.find(".//hr") or pt.find(".//HeartRateBpm/Value")
+            if hr_el is not None and hr_el.text:
+                try:
+                    v = int(hr_el.text)
+                    hr_vals.append(v)
+                    coord["heart_rate"] = v
+                except (ValueError, TypeError):
+                    pass
+            coords.append(coord)
+
+        if len(coords) < 2:
+            continue
+
+        try:
+            start_dt = datetime.fromisoformat(coords[0]["timestamp"].replace("Z", "+00:00"))
+            end_dt   = datetime.fromisoformat(coords[-1]["timestamp"].replace("Z", "+00:00"))
+        except (ValueError, KeyError):
+            continue
+
+        duration_s = int((end_dt - start_dt).total_seconds())
+        dist_m     = sum(_haversine(coords[i-1]["lat"], coords[i-1]["lng"],
+                                    coords[i]["lat"],   coords[i]["lng"])
+                         for i in range(1, len(coords)))
+
+        workout_rows.append({
+            "user_id":            user_id, "type": "workout", "device": "garmin",
+            "timestamp":          start_dt.isoformat(),
+            "workout_type":       workout_type,
+            "duration_minutes":   round(duration_s / 60, 1),
+            "distance_meters":    round(dist_m, 1),
+            "avg_heart_rate":     round(sum(hr_vals) / len(hr_vals)) if hr_vals else None,
+            "max_heart_rate":     max(hr_vals) if hr_vals else None,
+            "ending_heart_rate":  hr_vals[-1]  if hr_vals else None,
+            "notes":              name,
+        })
+        route_rows.append({
+            "user_id": user_id, "workout_type": workout_type,
+            "coordinates": coords, "distance_meters": round(dist_m, 1),
+            "duration_seconds": duration_s,
+            "started_at": start_dt.isoformat(), "ended_at": end_dt.isoformat(),
+            "notes": name,
+        })
+
+    return workout_rows, route_rows
+
+
+def _parse_tcx(content: bytes, user_id: str) -> tuple[list, list]:
+    try:
+        root = _strip_ns(ET.fromstring(content))
+    except ET.ParseError:
+        return [], []
+
+    workout_rows, route_rows = [], []
+
+    for activity in root.findall(".//Activity"):
+        sport    = activity.get("Sport", "other")
+        workout_type = _classify_activity(sport)
+        id_el    = activity.find("Id")
+        if id_el is None:
+            continue
+        try:
+            start_dt = datetime.fromisoformat(id_el.text.replace("Z", "+00:00"))
+        except (ValueError, AttributeError):
+            continue
+
+        total_time_s = 0
+        total_dist_m = 0.0
+        total_cal    = 0
+        hr_vals, coords = [], []
+
+        for lap in activity.findall(".//Lap"):
+            for tag, target in [("TotalTimeSeconds", None), ("DistanceMeters", None),
+                                  ("Calories", None)]:
+                el = lap.find(tag)
+                if el is not None and el.text:
+                    try:
+                        val = float(el.text)
+                        if tag == "TotalTimeSeconds": total_time_s += int(val)
+                        elif tag == "DistanceMeters": total_dist_m += val
+                        elif tag == "Calories":       total_cal    += int(val)
+                    except (ValueError, TypeError):
+                        pass
+
+            for tp in lap.findall(".//Trackpoint"):
+                time_el = tp.find("Time")
+                pos_el  = tp.find("Position")
+                hr_el   = tp.find(".//HeartRateBpm/Value") or tp.find("HeartRateBpm/Value")
+                if pos_el is not None and time_el is not None:
+                    lat_el = pos_el.find("LatitudeDegrees")
+                    lng_el = pos_el.find("LongitudeDegrees")
+                    if lat_el is not None and lng_el is not None:
+                        try:
+                            coord: dict = {
+                                "lat": float(lat_el.text),
+                                "lng": float(lng_el.text),
+                                "timestamp": time_el.text,
+                            }
+                            if hr_el is not None and hr_el.text:
+                                v = int(hr_el.text)
+                                hr_vals.append(v)
+                                coord["heart_rate"] = v
+                            coords.append(coord)
+                        except (ValueError, TypeError):
+                            pass
+
+        end_dt = start_dt + timedelta(seconds=total_time_s)
+        workout_rows.append({
+            "user_id":            user_id, "type": "workout", "device": "garmin",
+            "timestamp":          start_dt.isoformat(),
+            "workout_type":       workout_type,
+            "duration_minutes":   round(total_time_s / 60, 1),
+            "distance_meters":    round(total_dist_m, 1) if total_dist_m else None,
+            "calories_burned":    total_cal if total_cal else None,
+            "avg_heart_rate":     round(sum(hr_vals) / len(hr_vals)) if hr_vals else None,
+            "max_heart_rate":     max(hr_vals) if hr_vals else None,
+            "ending_heart_rate":  hr_vals[-1]  if hr_vals else None,
+        })
+        if len(coords) >= 2:
+            route_rows.append({
+                "user_id": user_id, "workout_type": workout_type,
+                "coordinates": coords,
+                "distance_meters":  round(total_dist_m, 1) if total_dist_m else None,
+                "duration_seconds": total_time_s,
+                "started_at": start_dt.isoformat(), "ended_at": end_dt.isoformat(),
+            })
+
+    return workout_rows, route_rows
+
+
+# ── Apple Health (XML / ZIP export) ──────────────────────────────────────────
+
+APPLE_WORKOUT_MAP = {
+    "HKWorkoutActivityTypeRunning":                       "running",
+    "HKWorkoutActivityTypeCycling":                       "cycling",
+    "HKWorkoutActivityTypeWalking":                       "walking",
+    "HKWorkoutActivityTypeHiking":                        "hiking",
+    "HKWorkoutActivityTypeSwimming":                      "swimming",
+    "HKWorkoutActivityTypeTraditionalStrengthTraining":   "weightlifting",
+    "HKWorkoutActivityTypeFunctionalStrengthTraining":    "weightlifting",
+    "HKWorkoutActivityTypeHighIntensityIntervalTraining": "other",
+    "HKWorkoutActivityTypeYoga":                          "other",
+    "HKWorkoutActivityTypeCrossTraining":                 "other",
+    "HKWorkoutActivityTypeElliptical":                    "other",
+    "HKWorkoutActivityTypePilates":                       "other",
+}
+
+
+@router.post("/apple/import")
+async def apple_import(file: UploadFile = File(...), user_id: str = Depends(get_user_id)):
+    content  = await file.read()
+    filename = (file.filename or "").lower()
+
+    if filename.endswith(".zip"):
+        try:
+            with zipfile.ZipFile(io.BytesIO(content)) as zf:
+                xml_name = next(
+                    (n for n in zf.namelist() if n.lower().endswith("export.xml")), None
+                )
+                if xml_name is None:
+                    return {"error": "No export.xml found in ZIP.", "imported": 0}
+                xml_bytes = zf.read(xml_name)
+        except zipfile.BadZipFile:
+            return {"error": "Invalid ZIP file.", "imported": 0}
+    elif filename.endswith(".xml"):
+        xml_bytes = content
+    else:
+        return {"error": "Upload export.xml or the ZIP from Health app → Export All Health Data.", "imported": 0}
+
+    workout_rows, reading_rows = _parse_apple_health(xml_bytes, user_id)
+    db = await get_db()
+    if workout_rows:
+        await db.table("watch_data").insert(workout_rows).execute()
+    if reading_rows:
+        await db.table("watch_data").insert(reading_rows).execute()
+    return {"imported": len(workout_rows), "readings_synced": len(reading_rows)}
+
+
+def _parse_apple_health(xml_bytes: bytes, user_id: str) -> tuple[list, list]:
+    try:
+        root = ET.fromstring(xml_bytes)
+    except ET.ParseError:
+        return [], []
+
+    workout_rows: list[dict] = []
+    hr_by_day:    dict[str, list[float]] = {}
+    steps_by_day: dict[str, int]         = {}
+    sleep_by_day: dict[str, float]       = {}
+
+    for w in root.findall("Workout"):
+        activity_type = w.get("workoutActivityType", "")
+        start_str     = w.get("startDate", "")
+        end_str       = w.get("endDate", "")
+        if not start_str:
+            continue
+        try:
+            start_dt = _parse_apple_date(start_str)
+            end_dt   = _parse_apple_date(end_str) if end_str else start_dt
+        except ValueError:
+            continue
+
+        duration_s = int((end_dt - start_dt).total_seconds())
+        calories = avg_hr = max_hr = dist_m = None
+
+        for stat in w.findall("WorkoutStatistics"):
+            stype = stat.get("type", "")
+            if "ActiveEnergyBurned" in stype:
+                try:
+                    calories = float(stat.get("sum") or stat.get("average") or 0) or None
+                except (ValueError, TypeError):
+                    pass
+            elif stype == "HKQuantityTypeIdentifierHeartRate":
+                try:
+                    avg_hr = round(float(stat.get("average") or 0)) or None
+                    max_hr = round(float(stat.get("maximum") or 0)) or None
+                except (ValueError, TypeError):
+                    pass
+            elif "Distance" in stype:
+                try:
+                    val  = float(stat.get("sum") or 0)
+                    unit = stat.get("unit", "km").lower()
+                    dist_m = round(val * (1000 if "km" in unit else 1609.34 if "mi" in unit else 1), 1) or None
+                except (ValueError, TypeError):
+                    pass
+
+        workout_rows.append({
+            "user_id":           user_id, "type": "workout", "device": "apple_health",
+            "timestamp":         start_dt.isoformat(),
+            "workout_type":      APPLE_WORKOUT_MAP.get(activity_type, "other"),
+            "duration_minutes":  round(duration_s / 60, 1),
+            "distance_meters":   dist_m,
+            "calories_burned":   calories,
+            "avg_heart_rate":    avg_hr,
+            "max_heart_rate":    max_hr,
+        })
+
+    for rec in root.findall("Record"):
+        rtype     = rec.get("type", "")
+        start_str = rec.get("startDate", "")
+        if not start_str:
+            continue
+        try:
+            dt  = _parse_apple_date(start_str)
+            day = dt.strftime("%Y-%m-%d")
+        except ValueError:
+            continue
+        val = rec.get("value", "")
+
+        if rtype == "HKQuantityTypeIdentifierHeartRate":
+            try:
+                hr_by_day.setdefault(day, []).append(float(val))
+            except (ValueError, TypeError):
+                pass
+        elif rtype == "HKQuantityTypeIdentifierStepCount":
+            try:
+                steps_by_day[day] = steps_by_day.get(day, 0) + int(float(val))
+            except (ValueError, TypeError):
+                pass
+        elif rtype == "HKCategoryTypeIdentifierSleepAnalysis" and val == "HKCategoryValueSleepAnalysisAsleep":
+            end_str2 = rec.get("endDate", "")
+            if end_str2:
+                try:
+                    end_dt2 = _parse_apple_date(end_str2)
+                    hours = (end_dt2 - dt).total_seconds() / 3600
+                    sleep_by_day[day] = sleep_by_day.get(day, 0) + hours
+                except ValueError:
+                    pass
+
+    reading_rows: list[dict] = []
+    for day in sorted(set(list(hr_by_day) + list(steps_by_day) + list(sleep_by_day))):
+        row: dict = {"user_id": user_id, "type": "reading", "device": "apple_health",
+                     "timestamp": f"{day}T00:00:00+00:00"}
+        hrs = hr_by_day.get(day)
+        if hrs:
+            row["heart_rate"] = round(sum(hrs) / len(hrs))
+        if day in steps_by_day:
+            row["steps"] = steps_by_day[day]
+        if day in sleep_by_day:
+            row["sleep_hours"] = round(sleep_by_day[day], 2)
+        reading_rows.append(row)
+
+    return workout_rows, reading_rows
+
+
+# ── Google Fit (Takeout ZIP import) ──────────────────────────────────────────
+
+@router.post("/google/import")
+async def google_fit_import(file: UploadFile = File(...), user_id: str = Depends(get_user_id)):
+    """
+    Import from Google Takeout → Fit data export.
+    Go to takeout.google.com, select only 'Fit', export as ZIP, then upload here.
+    """
+    content = await file.read()
+    if not (file.filename or "").lower().endswith(".zip"):
+        return {"error": "Upload the ZIP file downloaded from takeout.google.com.", "imported": 0}
+
+    try:
+        zf = zipfile.ZipFile(io.BytesIO(content))
+    except zipfile.BadZipFile:
+        return {"error": "Invalid ZIP file.", "imported": 0}
+
+    workout_rows: list[dict] = []
+    reading_rows: list[dict] = []
+
+    with zf:
+        names = zf.namelist()
+
+        # ── Session JSON files (individual activities) ─────────────────
+        session_files = [n for n in names
+                         if "Activities" in n and n.endswith(".json")]
+        for fname in session_files:
+            try:
+                data = json.loads(zf.read(fname))
+            except (json.JSONDecodeError, KeyError):
+                continue
+            if not isinstance(data, dict):
+                continue
+
+            start_str = data.get("startTime") or data.get("start_time_millis")
+            end_str   = data.get("endTime")   or data.get("end_time_millis")
+            if not start_str:
+                continue
+
+            try:
+                if isinstance(start_str, (int, float)):
+                    start_dt = datetime.fromtimestamp(start_str / 1000, tz=timezone.utc)
+                    end_dt   = datetime.fromtimestamp((end_str or start_str) / 1000, tz=timezone.utc)
+                else:
+                    start_dt = datetime.fromisoformat(start_str.replace("Z", "+00:00"))
+                    end_dt   = datetime.fromisoformat((end_str or start_str).replace("Z", "+00:00"))
+            except (ValueError, TypeError):
+                continue
+
+            activity_type = str(data.get("activityType", data.get("activity_type", 0)))
+            workout_type  = _google_fit_activity_type(activity_type)
+            duration_s    = int((end_dt - start_dt).total_seconds())
+
+            calories = dist_m = avg_hr = None
+            for seg in data.get("activitySegment", {}).get("activityConfidence", []):
+                pass  # presence only; stats are in aggregate fields below
+            try:
+                calories = float(data.get("calories") or data.get("caloriesExpended") or 0) or None
+            except (ValueError, TypeError):
+                pass
+            try:
+                dist_m = float(data.get("distance") or data.get("distanceMeters") or 0) or None
+            except (ValueError, TypeError):
+                pass
+
+            # Heart rate from aggregate key
+            hr_data = data.get("heartRate") or {}
+            try:
+                avg_hr = round(float(hr_data.get("avg") or hr_data.get("average") or 0)) or None
+            except (ValueError, TypeError):
+                pass
+
+            workout_rows.append({
+                "user_id":          user_id, "type": "workout", "device": "google_fit",
+                "timestamp":        start_dt.isoformat(),
+                "workout_type":     workout_type,
+                "duration_minutes": round(duration_s / 60, 1),
+                "distance_meters":  dist_m,
+                "calories_burned":  calories,
+                "avg_heart_rate":   avg_hr,
+            })
+
+        # ── Daily Summary CSV ──────────────────────────────────────────
+        csv_file = next(
+            (n for n in names if "Daily" in n and n.endswith(".csv")), None
+        )
+        if csv_file:
+            try:
+                text = zf.read(csv_file).decode("utf-8-sig")
+                reader = csv.DictReader(io.StringIO(text))
+                for row in reader:
+                    day = (row.get("Date") or "").strip()
+                    if not day:
+                        continue
+                    reading: dict = {"user_id": user_id, "type": "reading",
+                                     "device": "google_fit",
+                                     "timestamp": f"{day}T00:00:00+00:00"}
+                    for col, field in [
+                        ("Step count",               "steps"),
+                        ("Calories (kcal)",          "calories_burned"),
+                        ("Average heart rate (bpm)", "heart_rate"),
+                        ("Sleep duration",           None),
+                    ]:
+                        val = (row.get(col) or "").strip()
+                        if not val:
+                            continue
+                        if col == "Sleep duration":
+                            try:
+                                h, m, s = (val + ":0:0").split(":")[:3]
+                                reading["sleep_hours"] = round(int(h) + int(m) / 60 + int(s) / 3600, 2)
+                            except (ValueError, TypeError):
+                                pass
+                        elif field:
+                            try:
+                                reading[field] = round(float(val), 1)
+                            except (ValueError, TypeError):
+                                pass
+                    reading_rows.append(reading)
+            except Exception:
+                pass
+
+    db = await get_db()
+    if workout_rows:
+        await db.table("watch_data").insert(workout_rows).execute()
+    if reading_rows:
+        await db.table("watch_data").insert(reading_rows).execute()
+    return {"imported": len(workout_rows), "readings_synced": len(reading_rows)}
+
+
+_GOOGLE_ACTIVITY_TYPES: dict[str, str] = {
+    "7": "cycling", "11": "cycling",
+    "1": "other",    "17": "hiking",
+    "37": "running", "38": "running",
+    "39": "running", "93": "running",
+    "41": "other",   "79": "walking",
+    "80": "walking", "45": "other",
+    "97": "weightlifting", "20": "other",
+    "82": "swimming",
+}
+
+
+def _google_fit_activity_type(code: str) -> str:
+    return _GOOGLE_ACTIVITY_TYPES.get(str(code), _classify_activity(code))

backend/routes/user.py

@@ -0,0 +1,148 @@
+from fastapi import APIRouter, Depends
+from auth import get_user_id
+from db import get_db
+from datetime import datetime, timezone, timedelta
+
+router = APIRouter()
+
+
+@router.get("/me")
+async def get_me(user_id: str = Depends(get_user_id)):
+    """Return basic stats for the authenticated user."""
+    db = await get_db()
+
+    readings = await db.table("watch_data").select("id", count="exact").eq("user_id", user_id).eq("type", "reading").execute()
+    workouts = await db.table("watch_data").select("id", count="exact").eq("user_id", user_id).eq("type", "workout").execute()
+    analyses = await db.table("analyses").select("id", count="exact").eq("user_id", user_id).execute()
+
+    return {
+        "user_id": user_id,
+        "readings": readings.count,
+        "workouts": workouts.count,
+        "analyses": analyses.count,
+    }
+
+
+@router.get("/history")
+async def get_history(user_id: str = Depends(get_user_id), limit: int = 20):
+    """Return the user's past AI analyses, newest first — consumed by the phone."""
+    db = await get_db()
+    result = await (
+        db.table("analyses")
+        .select("*")
+        .eq("user_id", user_id)
+        .order("created_at", desc=True)
+        .limit(limit)
+        .execute()
+    )
+    return {"analyses": result.data}
+
+
+@router.get("/summary")
+async def get_summary(user_id: str = Depends(get_user_id)):
+    """Weekly comparison stats, streak, and personal records."""
+    db = await get_db()
+    now = datetime.now(timezone.utc)
+    seven_ago    = now - timedelta(days=7)
+    fourteen_ago = now - timedelta(days=14)
+    sixty_ago    = now - timedelta(days=60)
+
+    # ── Last 14 days for week-over-week comparison ─────────────────────
+    recent = await (
+        db.table("watch_data").select("*")
+        .eq("user_id", user_id)
+        .gte("timestamp", fourteen_ago.isoformat())
+        .execute()
+    )
+    rows = recent.data or []
+    seven_str = seven_ago.isoformat()
+    this_week = [r for r in rows if (r.get("timestamp") or "") >= seven_str]
+    last_week = [r for r in rows if (r.get("timestamp") or "") <  seven_str]
+
+    def week_stats(week_rows):
+        workouts = [r for r in week_rows if r.get("type") == "workout"]
+        readings = [r for r in week_rows if r.get("type") == "reading"]
+        hrs   = [r.get("avg_heart_rate") or r.get("heart_rate") for r in week_rows
+                 if r.get("avg_heart_rate") or r.get("heart_rate")]
+        steps = [r.get("steps") or 0 for r in readings]
+        sleep = [r.get("sleep_hours") for r in readings if r.get("sleep_hours")]
+        return {
+            "workouts":    len(workouts),
+            "distance_km": round(sum(r.get("distance_meters") or 0 for r in workouts) / 1000, 1),
+            "calories":    round(sum(r.get("calories_burned") or 0 for r in workouts)),
+            "avg_hr":      round(sum(hrs) / len(hrs)) if hrs else 0,
+            "total_steps": sum(steps),
+            "avg_sleep":   round(sum(sleep) / len(sleep), 1) if sleep else 0,
+        }
+
+    # ── Streak (consecutive days with a workout) ───────────────────────
+    streak_res = await (
+        db.table("watch_data").select("timestamp")
+        .eq("user_id", user_id).eq("type", "workout")
+        .gte("timestamp", sixty_ago.isoformat())
+        .execute()
+    )
+    active_days = {(r.get("timestamp") or "")[:10] for r in (streak_res.data or []) if r.get("timestamp")}
+    streak, check = 0, now.date()
+    while str(check) in active_days:
+        streak += 1
+        check -= timedelta(days=1)
+
+    # ── Personal records ───────────────────────────────────────────────
+    w_all = await (db.table("watch_data").select("distance_meters,calories_burned,heart_rate,steps,timestamp")
+                   .eq("user_id", user_id).execute())
+    r_all = await (db.table("routes").select("distance_meters,duration_seconds")
+                   .eq("user_id", user_id).execute())
+
+    w_rows = w_all.data or []
+    workout_rows  = [r for r in w_rows if r.get("distance_meters") or r.get("calories_burned")]
+    reading_rows  = [r for r in w_rows if r.get("heart_rate") or r.get("steps")]
+
+    max_dist_m = max((r.get("distance_meters") or 0 for r in workout_rows), default=0)
+    max_cal    = max((r.get("calories_burned") or 0 for r in workout_rows), default=0)
+    max_hr     = max((r.get("heart_rate") or 0 for r in reading_rows), default=0)
+
+    steps_by_day: dict = {}
+    for r in reading_rows:
+        day = (r.get("timestamp") or "")[:10]
+        if day:
+            steps_by_day[day] = steps_by_day.get(day, 0) + (r.get("steps") or 0)
+    max_steps_day = max(steps_by_day.values(), default=0)
+
+    best_pace_str, best_pace_val = None, float("inf")
+    for r in (r_all.data or []):
+        dist, dur = r.get("distance_meters") or 0, r.get("duration_seconds") or 0
+        if dist > 500 and dur > 0:
+            pace = (dur / 60) / (dist / 1000)
+            if pace < best_pace_val:
+                best_pace_val = pace
+                m, s = int(pace), int((pace % 1) * 60)
+                best_pace_str = f"{m}:{s:02d}"
+
+    return {
+        "streak":    streak,
+        "this_week": week_stats(this_week),
+        "last_week": week_stats(last_week),
+        "records": {
+            "longest_km":   round(max_dist_m / 1000, 1) if max_dist_m else None,
+            "max_calories": round(max_cal)  if max_cal  else None,
+            "max_hr":       round(max_hr)   if max_hr   else None,
+            "max_steps":    max_steps_day   if max_steps_day else None,
+            "best_pace":    best_pace_str,
+        },
+    }
+
+
+@router.get("/data")
+async def get_watch_data(user_id: str = Depends(get_user_id), limit: int = 100):
+    """Return raw watch data for the user, newest first."""
+    db = await get_db()
+    result = await (
+        db.table("watch_data")
+        .select("*")
+        .eq("user_id", user_id)
+        .order("timestamp", desc=True)
+        .limit(limit)
+        .execute()
+    )
+    return {"data": result.data}

backend/routes/watch.py

@@ -0,0 +1,49 @@
+from fastapi import APIRouter, Depends, HTTPException
+from auth import get_user_id
+from db import get_db
+from models.watch import WatchSyncPayload
+from calories import estimate_calories
+
+router = APIRouter()
+
+
+@router.post("/sync")
+async def sync_watch_data(payload: WatchSyncPayload, user_id: str = Depends(get_user_id)):
+    """Receive a batch of watch readings and workouts from the phone/watch app."""
+    db = await get_db()
+    rows = []
+
+    for reading in payload.readings:
+        rows.append({
+            "user_id": user_id,
+            "type": "reading",
+            "device": payload.device,
+            **reading.model_dump(),
+        })
+
+    # The watch no longer computes calories (it can't know the user's weight), so
+    # we estimate them here. Only for watch-originated workouts — manual web
+    # entries intentionally leave calories blank unless the user types them.
+    is_watch = (payload.device or "").startswith("fitness_watch")
+
+    for workout in payload.workouts:
+        row = {
+            "user_id": user_id,
+            "type": "workout",
+            "device": payload.device,
+            **workout.model_dump(),
+        }
+        if is_watch and row.get("calories_burned") is None:
+            row["calories_burned"] = estimate_calories(
+                row.get("duration_minutes") or 0, row.get("workout_type"))
+        rows.append(row)
+
+    if not rows:
+        raise HTTPException(status_code=400, detail="No data provided")
+
+    for row in rows:
+        if row.get("timestamp"):
+            row["timestamp"] = row["timestamp"].isoformat()
+
+    await db.table("watch_data").insert(rows).execute()
+    return {"synced": len(rows)}