diff --git a/pilot/dashboard_data.py b/pilot/dashboard_data.py
new file mode 100644
index 0000000..7377ca2
--- /dev/null
+++ b/pilot/dashboard_data.py
@@ -0,0 +1,302 @@
+"""pragent pilot — dashboard read-only query layer.
+
+Three functions: overview / repo_summary / pr_summary. Each opens the SQLite
+feedback DB via `feedback.init`, runs the queries it needs, and returns plain
+dicts/lists. NEVER writes — that's the dashboard_server's job (via the Gitea
+contents API). This module is what the dashboard_server's templates render.
+
+All three functions are tolerant of a missing or empty DB: they return the
+shaped dict with zeros/empty lists rather than crashing. The dashboard is a
+read-only view; the pilot can boot with no feedback DB and the dashboard
+should still load.
+
+Cost note: `total_cost_usd` is hardcoded to 0.0. Per-review `usage:cost` is
+not in the feedback SQLite — only the raw `review` / `inline_finding` rows
+are stored there. The equivalent-cost calc lives in `ai_review._render_collapsible_usage`
+and only knows about the latest review's tokens. Surfacing a rolled-up dollar
+figure without per-row telemetry would be guessing, so we don't.
+"""
+from __future__ import annotations
+
+import datetime
+import os
+import sqlite3
+
+from pilot import feedback
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _empty_overview() -> dict:
+    return {
+        "total_reviews": 0,
+        "total_findings": 0,
+        "total_repos": 0,
+        "last_30d_reviews": 0,
+        "daily": [{"date": _iso_date(i), "count": 0} for i in range(7)],
+        "top_repos": [],
+        "total_cost_usd": 0.0,
+    }
+
+
+def _empty_repo_summary(repo: str) -> dict:
+    return {
+        "repo": repo,
+        "total_runs": 0,
+        "last_run_ts": 0,
+        "runs_by_day": [],
+        "findings_by_severity": {"critical": 0, "high": 0, "medium": 0, "low": 0},
+        "top_findings": [],
+        # NOTE: review rows don't carry a `model` column in the schema today,
+        # so we have nothing to aggregate. When that lands, replace this
+        # empty list with a `SELECT model, COUNT(*) …` over `review`.
+        "models_used": [],
+    }
+
+
+def _empty_pr_summary(repo: str, pr: int) -> dict:
+    return {
+        "repo": repo,
+        "pr": pr,
+        "head_sha": "",
+        "posted_at": 0,
+        "review_id_gitea": None,
+        "body_comment_id": None,
+        "findings": [],
+        # usage isn't on the review row today; ai_review.py renders it
+        # in-memory at review time. Leave empty.
+        "usage": {},
+    }
+
+
+def _iso_date(days_ago: int) -> str:
+    """Return YYYY-MM-DD for `days_ago` days before today (UTC)."""
+    d = datetime.datetime.now(datetime.timezone.utc).date() - datetime.timedelta(days=days_ago)
+    return d.isoformat()
+
+
+def _open_or_none(db_path: str) -> sqlite3.Connection | None:
+    """Open the DB if it exists and looks like a feedback DB. Else None.
+
+    Tolerates missing files (fresh container) and a schema-less file (the
+    operator dropped a stray DB at the path). Returns a connection with
+    Row factory set so callers can use `row["col"]`.
+    """
+    if not db_path or not os.path.exists(db_path):
+        return None
+    try:
+        conn = feedback.init(db_path)
+    except sqlite3.DatabaseError:
+        return None
+    return conn
+
+
+# ---------------------------------------------------------------------------
+# Public API
+# ---------------------------------------------------------------------------
+
+
+def overview(db_path: str) -> dict:
+    """Top-of-page summary: totals + 7-bucket daily sparkline + top 5 repos."""
+    conn = _open_or_none(db_path)
+    if conn is None:
+        return _empty_overview()
+    try:
+        cur = conn.execute("SELECT COUNT(*) FROM review")
+        total_reviews = cur.fetchone()[0]
+        cur = conn.execute("SELECT COUNT(*) FROM inline_finding")
+        total_findings = cur.fetchone()[0]
+        cur = conn.execute("SELECT COUNT(DISTINCT repo) FROM review")
+        total_repos = cur.fetchone()[0]
+
+        # Last 30d window — reviews AND findings posted within the window.
+        ts_30d_ago = int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400
+        cur = conn.execute("SELECT COUNT(*) FROM review WHERE posted_at >= ?", (ts_30d_ago,))
+        last_30d_reviews = cur.fetchone()[0]
+
+        # 7-bucket daily sparkline, oldest first. Bucket key is UTC date.
+        cur = conn.execute(
+            "SELECT posted_at FROM review WHERE posted_at >= ?",
+            (int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 7 * 86400,),
+        )
+        buckets: dict[str, int] = {_iso_date(i): 0 for i in range(7)}
+        for (ts,) in cur.fetchall():
+            d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
+            if d in buckets:
+                buckets[d] += 1
+        daily = [{"date": _iso_date(i), "count": buckets[_iso_date(i)]} for i in range(7)]
+
+        # Top 5 repos by run count, descending. last_seen is the most recent
+        # review timestamp on that repo.
+        cur = conn.execute(
+            "SELECT repo, COUNT(*) AS runs, MAX(posted_at) AS last_seen "
+            "FROM review GROUP BY repo ORDER BY runs DESC, last_seen DESC LIMIT 5"
+        )
+        top_repos = [
+            {"repo": row[0], "run_count": row[1], "last_seen": int(row[2])}
+            for row in cur.fetchall()
+        ]
+
+        return {
+            "total_reviews": total_reviews,
+            "total_findings": total_findings,
+            "total_repos": total_repos,
+            "last_30d_reviews": last_30d_reviews,
+            "daily": daily,
+            "top_repos": top_repos,
+            "total_cost_usd": 0.0,
+        }
+    finally:
+        conn.close()
+
+
+def repo_summary(db_path: str, repo: str) -> dict:
+    """Per-repo drill-down: runs by day, severity histogram, top findings."""
+    conn = _open_or_none(db_path)
+    if conn is None:
+        return _empty_repo_summary(repo)
+    try:
+        cur = conn.execute(
+            "SELECT COUNT(*), MAX(posted_at) FROM review WHERE repo = ?", (repo,)
+        )
+        row = cur.fetchone()
+        total_runs = row[0] or 0
+        last_run_ts = int(row[1]) if row[1] else 0
+
+        # runs_by_day for the last 30 days, oldest first; zero-buckets included.
+        cur = conn.execute(
+            "SELECT posted_at FROM review WHERE repo = ? AND posted_at >= ?",
+            (repo, int(datetime.datetime.now(datetime.timezone.utc).timestamp()) - 30 * 86400),
+        )
+        buckets: dict[str, int] = {}
+        for d in range(30):
+            buckets[_iso_date(d)] = 0  # newest-day mapped to 0; we'll iterate
+        # Re-key: build oldest-first, days_ago goes 29..0
+        oldest_first = {}
+        for d in range(30):
+            oldest_first[_iso_date(29 - d)] = 0
+        for (ts,) in cur.fetchall():
+            d = datetime.datetime.fromtimestamp(ts, tz=datetime.timezone.utc).date().isoformat()
+            if d in oldest_first:
+                oldest_first[d] += 1
+        runs_by_day = [{"date": k, "count": v} for k, v in oldest_first.items()]
+
+        # findings_by_severity — case-insensitive match; bucket unknown as 'low'.
+        cur = conn.execute(
+            "SELECT severity, COUNT(*) FROM inline_finding WHERE repo = ? GROUP BY severity",
+            (repo,),
+        )
+        fbs = {"critical": 0, "high": 0, "medium": 0, "low": 0}
+        for sev, n in cur.fetchall():
+            k = (sev or "").strip().lower()
+            if k not in fbs:
+                k = "low"
+            fbs[k] += n
+
+        # top_findings — top 5 posthashes by occurrence count, joined with
+        # vote rollups via feedback.findings_with_votes.
+        cur = conn.execute(
+            "SELECT f.path, f.line, MAX(f.severity) AS severity, MAX(f.problem) AS problem, "
+            "COUNT(*) AS occurrences, "
+            "COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
+            "COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
+            "MAX(ts.resolved) AS resolved, "
+            "COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id IN "
+            "  (SELECT id FROM inline_finding WHERE posthash = f.posthash AND repo = f.repo AND path = f.path AND line = f.line)), 0) AS reply_count "
+            "FROM inline_finding f "
+            "LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
+            "LEFT JOIN thread_state ts ON ts.finding_id = f.id "
+            "WHERE f.repo = ? "
+            "GROUP BY f.posthash, f.repo, f.path, f.line "
+            "ORDER BY occurrences DESC, upvotes DESC LIMIT 5",
+            (repo,),
+        )
+        top_findings = [
+            {
+                "path": r[0],
+                "line": r[1],
+                "severity": r[2],
+                "problem": r[3],
+                "occurrences": r[4],
+                "upvotes": int(r[5] or 0),
+                "downvotes": int(r[6] or 0),
+                "resolved": int(r[7] or 0),
+                "reply_count": int(r[8] or 0),
+            }
+            for r in cur.fetchall()
+        ]
+
+        return {
+            "repo": repo,
+            "total_runs": total_runs,
+            "last_run_ts": last_run_ts,
+            "runs_by_day": runs_by_day,
+            "findings_by_severity": fbs,
+            "top_findings": top_findings,
+            "models_used": [],  # see _empty_repo_summary NOTE
+        }
+    finally:
+        conn.close()
+
+
+def pr_summary(db_path: str, repo: str, pr: int) -> dict:
+    """Per-PR view: meta + every finding the bot ever posted on that PR."""
+    conn = _open_or_none(db_path)
+    if conn is None:
+        return _empty_pr_summary(repo, pr)
+    try:
+        cur = conn.execute(
+            "SELECT head_sha, posted_at, review_id_gitea, body_comment_id "
+            "FROM review WHERE repo = ? AND pr = ? ORDER BY posted_at DESC LIMIT 1",
+            (repo, pr),
+        )
+        row = cur.fetchone()
+        if row is None:
+            return _empty_pr_summary(repo, pr)
+        head_sha, posted_at, review_id_gitea, body_comment_id = row
+
+        cur = conn.execute(
+            "SELECT f.path, f.line, f.severity, f.problem, f.fix, f.suggestion, "
+            "COALESCE(SUM(CASE WHEN rct.content = '+1' THEN 1 ELSE 0 END), 0) AS upvotes, "
+            "COALESCE(SUM(CASE WHEN rct.content = '-1' THEN 1 ELSE 0 END), 0) AS downvotes, "
+            "MAX(ts.resolved) AS resolved, "
+            "COALESCE((SELECT COUNT(*) FROM reply WHERE finding_id = f.id), 0) AS reply_count "
+            "FROM inline_finding f "
+            "LEFT JOIN reaction rct ON rct.comment_id = f.comment_id "
+            "LEFT JOIN thread_state ts ON ts.finding_id = f.id "
+            "WHERE f.repo = ? AND f.pr = ? "
+            "GROUP BY f.id "
+            "ORDER BY f.path, f.line",
+            (repo, pr),
+        )
+        findings = [
+            {
+                "path": r[0],
+                "line": r[1],
+                "severity": r[2],
+                "problem": r[3],
+                "fix": r[4],
+                "suggestion": r[5],
+                "upvotes": int(r[6] or 0),
+                "downvotes": int(r[7] or 0),
+                "resolved": int(r[8] or 0),
+                "reply_count": int(r[9] or 0),
+            }
+            for r in cur.fetchall()
+        ]
+
+        return {
+            "repo": repo,
+            "pr": pr,
+            "head_sha": head_sha,
+            "posted_at": int(posted_at),
+            "review_id_gitea": review_id_gitea,
+            "body_comment_id": body_comment_id,
+            "findings": findings,
+            "usage": {},
+        }
+    finally:
+        conn.close()
\ No newline at end of file
diff --git a/tests/pilot/test_dashboard_data.py b/tests/pilot/test_dashboard_data.py
new file mode 100644
index 0000000..fe2ba06
--- /dev/null
+++ b/tests/pilot/test_dashboard_data.py
@@ -0,0 +1,249 @@
+"""Tests for pilot/dashboard_data.py — read-only query layer over the feedback SQLite.
+
+Covers: empty-DB fallbacks (no crash on missing/empty DB), overview rollups,
+per-repo drill-down (findings by severity, top findings, runs by day), and
+the per-PR view. The dashboard never writes — only reads.
+"""
+import os
+import sys
+import tempfile
+import time
+import unittest
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.abspath(os.path.join(HERE, "..", ".."))
+sys.path.insert(0, os.path.join(HERE, "..", ".."))  # so `from pilot import …` works
+
+from pilot import dashboard_data, feedback
+
+
+def _seed_repo(conn, *, repo: str, prs: int, findings_per_pr: int, day_offset: int = 0):
+    """Seed one repo with `prs` PRs each with `findings_per_pr` findings.
+
+    All timestamps cluster on (now - day_offset days). Returns list of review ids.
+    """
+    base = int(time.time()) - day_offset * 86400
+    rids = []
+    for n in range(prs):
+        rid = feedback.record_review(
+            conn, repo=repo, pr=n + 1, head_sha=f"sha{n}",
+            review_id_gitea=1000 + n, body_comment_id=2000 + n,
+            posted_at=base + n * 60,
+        )
+        rids.append(rid)
+        for k in range(findings_per_pr):
+            feedback.record_inline_finding(
+                conn, review_id=rid, repo=repo, pr=n + 1,
+                path=f"src/file_{k}.py", line=k + 1,
+                severity=["critical", "high", "medium", "low"][k % 4],
+                problem=f"problem {k}",
+                fix=f"fix {k}", suggestion=f"suggestion {k}",
+                comment_id=3000 + n * 10 + k,
+                posted_at=base + n * 60,
+            )
+    return rids
+
+
+class TestEmptyDB(unittest.TestCase):
+    def test_missing_file_returns_zero_dict(self):
+        with tempfile.TemporaryDirectory() as d:
+            missing = f"{d}/nope.db"
+            data = dashboard_data.overview(missing)
+            self.assertEqual(data["total_reviews"], 0)
+            self.assertEqual(data["total_findings"], 0)
+            self.assertEqual(data["total_repos"], 0)
+            self.assertEqual(data["last_30d_reviews"], 0)
+            self.assertEqual(len(data["daily"]), 7)
+            self.assertEqual(data["top_repos"], [])
+            self.assertEqual(data["total_cost_usd"], 0.0)
+
+    def test_missing_file_repo_summary_safe(self):
+        with tempfile.TemporaryDirectory() as d:
+            data = dashboard_data.repo_summary(f"{d}/nope.db", "o/r")
+            self.assertEqual(data["repo"], "o/r")
+            self.assertEqual(data["total_runs"], 0)
+            self.assertEqual(data["runs_by_day"], [])
+            for sev in ("critical", "high", "medium", "low"):
+                self.assertEqual(data["findings_by_severity"][sev], 0)
+            self.assertEqual(data["top_findings"], [])
+            self.assertEqual(data["models_used"], [])
+
+    def test_missing_file_pr_summary_safe(self):
+        with tempfile.TemporaryDirectory() as d:
+            data = dashboard_data.pr_summary(f"{d}/nope.db", "o/r", 1)
+            self.assertEqual(data["repo"], "o/r")
+            self.assertEqual(data["pr"], 1)
+            self.assertEqual(data["findings"], [])
+            self.assertEqual(data["usage"], {})
+
+
+class TestEmptyButExistingDB(unittest.TestCase):
+    """`init` creates the schema — DB exists but has no rows."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = f"{self.tmp.name}/f.db"
+        feedback.init(self.db)
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_overview_is_zero(self):
+        data = dashboard_data.overview(self.db)
+        self.assertEqual(data["total_reviews"], 0)
+        self.assertEqual(data["total_findings"], 0)
+        self.assertEqual(data["total_repos"], 0)
+
+    def test_repo_summary_is_zero(self):
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        self.assertEqual(data["total_runs"], 0)
+        self.assertEqual(data["findings_by_severity"], {"critical": 0, "high": 0, "medium": 0, "low": 0})
+
+    def test_pr_summary_is_zero(self):
+        data = dashboard_data.pr_summary(self.db, "o/r", 1)
+        self.assertEqual(data["findings"], [])
+
+
+class TestOverview(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = f"{self.tmp.name}/f.db"
+        self.conn = feedback.init(self.db)
+        _seed_repo(self.conn, repo="alpha/one", prs=3, findings_per_pr=2)
+        _seed_repo(self.conn, repo="beta/two", prs=1, findings_per_pr=4)
+        self.conn.close()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_totals(self):
+        data = dashboard_data.overview(self.db)
+        self.assertEqual(data["total_reviews"], 4)
+        self.assertEqual(data["total_findings"], 6 + 4)  # 3*2 + 1*4 = 10
+        self.assertEqual(data["total_repos"], 2)
+        self.assertEqual(data["total_cost_usd"], 0.0)
+
+    def test_top_repos_sorted_by_run_count(self):
+        data = dashboard_data.overview(self.db)
+        repos = [r["repo"] for r in data["top_repos"]]
+        # alpha/one has 3 runs, beta/two has 1.
+        self.assertEqual(repos[0], "alpha/one")
+        self.assertEqual(data["top_repos"][0]["run_count"], 3)
+        self.assertEqual(data["top_repos"][1]["run_count"], 1)
+        # last_seen is a unix timestamp int.
+        for r in data["top_repos"]:
+            self.assertIsInstance(r["last_seen"], int)
+
+    def test_daily_buckets_are_7(self):
+        data = dashboard_data.overview(self.db)
+        self.assertEqual(len(data["daily"]), 7)
+        for b in data["daily"]:
+            self.assertIn("date", b)
+            self.assertIn("count", b)
+
+    def test_last_30d_reviews(self):
+        data = dashboard_data.overview(self.db)
+        self.assertEqual(data["last_30d_reviews"], 4)
+
+
+class TestRepoSummary(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = f"{self.tmp.name}/f.db"
+        self.conn = feedback.init(self.db)
+        # 4 PRs with 2 findings each → 8 findings, severity cycle [c,h,m,l,c,h,m,l]
+        _seed_repo(self.conn, repo="o/r", prs=4, findings_per_pr=2)
+        # Add some reactions so top_findings has signal.
+        rows = self.conn.execute(
+            "SELECT id, comment_id FROM inline_finding WHERE repo=? ORDER BY id LIMIT 3",
+            ("o/r",),
+        ).fetchall()
+        for r in rows:
+            feedback.record_reaction(self.conn, comment_id=r["comment_id"], user="u", content="+1")
+        self.conn.close()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_basic_shape(self):
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        self.assertEqual(data["repo"], "o/r")
+        self.assertEqual(data["total_runs"], 4)
+        self.assertIsInstance(data["last_run_ts"], int)
+
+    def test_findings_by_severity(self):
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        fbs = data["findings_by_severity"]
+        # 4 PRs × 2 findings; per-PR severities are [critical, high].
+        # (k in range(2) → k=0 critical, k=1 high for every PR.)
+        self.assertEqual(fbs["critical"], 4)
+        self.assertEqual(fbs["high"], 4)
+        self.assertEqual(fbs["medium"], 0)
+        self.assertEqual(fbs["low"], 0)
+
+    def test_runs_by_day_is_list(self):
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        self.assertIsInstance(data["runs_by_day"], list)
+        for r in data["runs_by_day"]:
+            self.assertIn("date", r)
+            self.assertIn("count", r)
+
+    def test_top_findings_structure(self):
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        self.assertGreater(len(data["top_findings"]), 0)
+        first = data["top_findings"][0]
+        for k in ("path", "line", "severity", "problem", "occurrences", "upvotes", "downvotes", "resolved", "reply_count"):
+            self.assertIn(k, first)
+
+    def test_models_used_is_empty_list_with_note(self):
+        # The schema has no `model` column on review — the dashboard can't show
+        # model usage from this DB today. We document that via an empty list.
+        data = dashboard_data.repo_summary(self.db, "o/r")
+        self.assertEqual(data["models_used"], [])
+
+
+class TestPRSummary(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.db = f"{self.tmp.name}/f.db"
+        self.conn = feedback.init(self.db)
+        rid = feedback.record_review(
+            self.conn, repo="o/r", pr=42, head_sha="abc",
+            review_id_gitea=9001, body_comment_id=8001,
+            posted_at=1700000000,
+        )
+        for k in range(3):
+            feedback.record_inline_finding(
+                self.conn, review_id=rid, repo="o/r", pr=42,
+                path=f"src/x_{k}.py", line=k + 10,
+                severity=["critical", "high", "low"][k],
+                problem=f"p{k}", fix=f"f{k}", suggestion=f"s{k}",
+                comment_id=7000 + k,
+            )
+        self.conn.close()
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def test_meta(self):
+        data = dashboard_data.pr_summary(self.db, "o/r", 42)
+        self.assertEqual(data["repo"], "o/r")
+        self.assertEqual(data["pr"], 42)
+        self.assertEqual(data["head_sha"], "abc")
+        self.assertEqual(data["review_id_gitea"], 9001)
+        self.assertEqual(data["body_comment_id"], 8001)
+        self.assertEqual(data["posted_at"], 1700000000)
+        # usage is empty because the schema has no usage column.
+        self.assertEqual(data["usage"], {})
+
+    def test_findings(self):
+        data = dashboard_data.pr_summary(self.db, "o/r", 42)
+        self.assertEqual(len(data["findings"]), 3)
+        for f in data["findings"]:
+            for k in ("path", "line", "severity", "problem", "fix", "suggestion",
+                      "upvotes", "downvotes", "resolved", "reply_count"):
+                self.assertIn(k, f)
+
+
+if __name__ == "__main__":
+    unittest.main()
\ No newline at end of file