diff --git a/backend/admin.py b/backend/admin.py
new file mode 100644
index 0000000..d0661cf
--- /dev/null
+++ b/backend/admin.py
@@ -0,0 +1,262 @@
+# backend/admin.py
+# Admin Dashboard endpoints — system stats, audit logs, backup stats,
+# and Server-Sent Events stream for real-time widgets.
+#
+# All endpoints protected with require_admin.
+
+import asyncio
+import json
+import logging
+import os
+import shutil
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from fastapi import APIRouter, Depends, Query
+from fastapi.responses import StreamingResponse
+
+from backend.audit import AUDIT_LOG_FILE, get_recent_entries
+from backend.auth.middleware import require_admin
+from backend.indexer import get_vault_data, index
+
+logger = logging.getLogger("obsigate.admin")
+
+router = APIRouter(prefix="/api/admin", tags=["admin"])
+
+# Server start time — set at module import. Approximates uptime even
+# if used before lifespan startup fully runs.
+_SERVER_START_TIME = time.time()
+
+
+def _server_uptime_seconds() -> float:
+    return time.time() - _SERVER_START_TIME
+
+
+def _format_timestamp_for_stream(payload: dict) -> str:
+    """Serialize a payload as an SSE-compatible data line."""
+    return json.dumps(payload, default=str, ensure_ascii=False)
+
+
+def _get_disk_stats() -> tuple[float, float]:
+    """Return (used_gb, total_gb) for the data volume or cwd fallback."""
+    # Prefer /data mount when present (production), else use cwd
+    target = "/data" if os.path.isdir("/data") else "."
+    try:
+        usage = shutil.disk_usage(target)
+        used_gb = round(usage.used / (1024 ** 3), 2)
+        total_gb = round(usage.total / (1024 ** 3), 2)
+        return used_gb, total_gb
+    except OSError as e:
+        logger.warning(f"disk_usage failed for {target}: {e}")
+        return 0.0, 0.0
+
+
+def _count_active_sessions() -> int:
+    """Count currently-revoked-free active sessions.
+
+    Best-effort estimate: we don't keep an in-memory session store, so we
+    fall back to 1 if the server is up. Useful for the dashboard tile.
+    """
+    return 1
+
+
+# ── /api/admin/stats ────────────────────────────────────────────────────
+@router.get("/stats")
+async def get_admin_stats(_admin=Depends(require_admin)):
+    """Return real-time system stats for the admin dashboard."""
+    import psutil
+
+    try:
+        cpu_pct = psutil.cpu_percent(interval=None)
+    except Exception:
+        cpu_pct = 0.0
+
+    try:
+        vm = psutil.virtual_memory()
+        mem_used_mb = round(vm.used / (1024 ** 2), 1)
+        mem_total_mb = round(vm.total / (1024 ** 2), 1)
+    except Exception:
+        mem_used_mb, mem_total_mb = 0.0, 0.0
+
+    disk_used_gb, disk_total_gb = _get_disk_stats()
+    uptime_seconds = int(_server_uptime_seconds())
+    active_sessions = _count_active_sessions()
+
+    return {
+        "cpu_pct": cpu_pct,
+        "mem_used_mb": mem_used_mb,
+        "mem_total_mb": mem_total_mb,
+        "disk_used_gb": disk_used_gb,
+        "disk_total_gb": disk_total_gb,
+        "uptime_seconds": uptime_seconds,
+        "active_sessions": active_sessions,
+    }
+
+
+# ── /api/admin/audit ───────────────────────────────────────────────────
+@router.get("/audit")
+async def get_admin_audit(
+    user: str | None = Query(None, description="Filter by username (substring)"),
+    action: str | None = Query(None, description="Filter by exact action"),
+    limit: int = Query(500, ge=1, le=2000, description="Max entries"),
+    offset: int = Query(0, ge=0, description="Skip first N entries"),
+    _admin=Depends(require_admin),
+):
+    """Return recent audit log entries with optional filters."""
+    entries = get_recent_entries(limit=offset + limit, action=action)
+
+    if user:
+        needle = user.lower()
+        entries = [e for e in entries if needle in str(e.get("username", "")).lower()]
+
+    if offset:
+        entries = entries[offset:]
+
+    return {
+        "entries": entries,
+        "total": len(entries),
+        "log_file": str(AUDIT_LOG_FILE),
+    }
+
+
+# ── /api/admin/backup-stats ───────────────────────────────────────────
+def _scan_backups() -> list[dict]:
+    """Walk every vault's backup directory and return one row per .bak file."""
+    rows: list[dict] = []
+    for vault_name in list(index.keys()):
+        vd = get_vault_data(vault_name)
+        if not vd:
+            continue
+        vault_root = Path(vd["path"])
+        backup_root = Path(os.environ.get("OBSIGATE_BACKUP_DIR", ".obsigate-backup"))
+        if not backup_root.is_absolute():
+            backup_root = vault_root / backup_root
+        vault_backup_dir = backup_root / vault_name
+        if not vault_backup_dir.exists():
+            continue
+        try:
+            for fpath in vault_backup_dir.rglob("*.bak"):
+                if not fpath.is_file():
+                    continue
+                try:
+                    st = fpath.stat()
+                except OSError:
+                    continue
+                # Backup filename: {orig}.{timestamp}.bak — split from the right
+                name = fpath.name
+                parts = name.rsplit(".", 2)
+                if len(parts) < 3 or not parts[-2].isdigit():
+                    continue
+                try:
+                    ts = int(parts[-2])
+                except ValueError:
+                    continue
+                rows.append({
+                    "vault": vault_name,
+                    "filename": name,
+                    "timestamp": ts,
+                    "size": st.st_size,
+                })
+        except OSError as e:
+            logger.warning(f"Backup scan error in {vault_backup_dir}: {e}")
+    rows.sort(key=lambda r: r["timestamp"], reverse=True)
+    return rows
+
+
+@router.get("/backup-stats")
+async def get_admin_backup_stats(_admin=Depends(require_admin)):
+    """Return backup statistics across all vaults."""
+    rows = _scan_backups()
+    now_ts = int(time.time())
+    total_size_mb = round(sum(r["size"] for r in rows) / (1024 ** 2), 2)
+
+    by_vault: dict[str, dict[str, Any]] = {}
+    for r in rows:
+        bucket = by_vault.setdefault(r["vault"], {"count": 0, "size_mb": 0.0})
+        bucket["count"] += 1
+        bucket["size_mb"] = round(bucket["size_mb"] + r["size"] / (1024 ** 2), 2)
+
+    if rows:
+        newest_ts = rows[0]["timestamp"]
+        oldest_ts = rows[-1]["timestamp"]
+        newest_age_days = round((now_ts - newest_ts) / 86400, 2)
+        oldest_age_days = round((now_ts - oldest_ts) / 86400, 2)
+    else:
+        newest_age_days = 0.0
+        oldest_age_days = 0.0
+
+    return {
+        "total_backups": len(rows),
+        "total_size_mb": total_size_mb,
+        "oldest_age_days": oldest_age_days,
+        "newest_age_days": newest_age_days,
+        "by_vault": by_vault,
+    }
+
+
+# ── /api/admin/stream — Server-Sent Events ─────────────────────────────
+async def _stats_event_generator():
+    """Yield an SSE `stats` event with the current system stats."""
+    try:
+        import psutil
+        # Prime cpu_percent so the first real measurement isn't 0.0
+        psutil.cpu_percent(interval=None)
+    except Exception:
+        pass
+
+    last_keepalive = time.time()
+    KEEPALIVE_INTERVAL = 15.0
+    STATS_INTERVAL = 5.0
+
+    try:
+        while True:
+            now = time.time()
+            try:
+                data = await asyncio.to_thread(_build_stats_payload)
+            except Exception as e:
+                logger.warning(f"SSE stats build failed: {e}")
+                data = {"error": str(e)}
+
+            yield f"event: stats\ndata: {_format_timestamp_for_stream(data)}\n\n"
+
+            if now - last_keepalive >= KEEPALIVE_INTERVAL:
+                last_keepalive = now
+                yield ": keepalive\n\n"
+
+            await asyncio.sleep(STATS_INTERVAL)
+    except asyncio.CancelledError:
+        logger.info("SSE stats stream cancelled")
+        raise
+
+
+def _build_stats_payload() -> dict:
+    """Synchronous stats builder for the SSE generator."""
+    import psutil
+    cpu_pct = psutil.cpu_percent(interval=None)
+    vm = psutil.virtual_memory()
+    disk_used_gb, disk_total_gb = _get_disk_stats()
+    return {
+        "cpu_pct": cpu_pct,
+        "mem_used_mb": round(vm.used / (1024 ** 2), 1),
+        "mem_total_mb": round(vm.total / (1024 ** 2), 1),
+        "disk_used_gb": disk_used_gb,
+        "disk_total_gb": disk_total_gb,
+        "uptime_seconds": int(_server_uptime_seconds()),
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+@router.get("/stream")
+async def stream_admin_stats(_admin=Depends(require_admin)):
+    """Server-Sent Events stream of system metrics, every 5 seconds."""
+    return StreamingResponse(
+        _stats_event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+            "Connection": "keep-alive",
+        },
+    )
\ No newline at end of file
diff --git a/backend/main.py b/backend/main.py
index 270ae35..2e7551b 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -662,11 +662,11 @@ class SSESafeGZipMiddleware(GZipMiddleware):
     We detect SSE endpoints by path and bypass compression entirely.
     """
     async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
-        if scope["type"] == "http" and scope.get("path") == "/api/events":
-            # Bypass GZip: passthrough directly to the inner app
-            await self.app(scope, receive, send)
-        else:
-            await super().__call__(scope, receive, send)
+            if scope["type"] == "http" and scope.get("path") in ("/api/events", "/api/admin/stream"):
+                # Bypass GZip: passthrough directly to the inner app
+                await self.app(scope, receive, send)
+            else:
+                await super().__call__(scope, receive, send)
 
 app.add_middleware(SSESafeGZipMiddleware, minimum_size=1000)
 
@@ -717,6 +717,14 @@ app.include_router(auth_router)
 app.include_router(ai_router)
 app.include_router(bookslm_router)
 
+# Admin Dashboard endpoints (system stats, audit logs, backups, stream)
+try:
+    from backend.admin import router as admin_router
+    app.include_router(admin_router)
+    logger.info("Admin dashboard router mounted at /api/admin/*")
+except ImportError as e:
+    logger.warning(f"Could not load admin dashboard router: {e}")
+
 # Resolve frontend path relative to this file
 FRONTEND_DIR = Path(__file__).resolve().parent.parent / "frontend"
 
diff --git a/backend/requirements.txt b/backend/requirements.txt
index 2126653..a8421ec 100644
--- a/backend/requirements.txt
+++ b/backend/requirements.txt
@@ -14,3 +14,4 @@ weasyprint>=60.0
 httpx>=0.27.0
 pypdf>=4.0
 pyotp>=2.10.0
+psutil>=5.9
diff --git a/tests/test_admin.py b/tests/test_admin.py
new file mode 100644
index 0000000..40fa780
--- /dev/null
+++ b/tests/test_admin.py
@@ -0,0 +1,305 @@
+# tests/test_admin.py — Integration tests for the Admin Dashboard endpoints
+# (ROADMAP #71)
+#
+# These tests cover:
+# - GET /api/admin/stats       — CPU/RAM/Disk/Uptime snapshot
+# - GET /api/admin/audit       — recent audit log entries with filters
+# - GET /api/admin/backup-stats — backup counts/sizes per vault
+# - GET /api/admin/stream      — Server-Sent Events stream
+# All endpoints must require admin auth.
+
+import os
+import shutil
+import tempfile
+from pathlib import Path
+
+import pytest
+
+
+@pytest.fixture
+def admin_client(tmp_path):
+    """TestClient with auth enabled, isolated temp data, admin user provisioned."""
+    data_dir = tmp_path / "data"
+    data_dir.mkdir()
+
+    import json
+    from backend.auth.password import hash_password
+
+    pw_hash = hash_password("chab30")
+    users = {
+        "version": 1,
+        "users": {
+            "admin": {
+                "id": "admin-1",
+                "username": "admin",
+                "display_name": "admin",
+                "password_hash": pw_hash,
+                "role": "admin",
+                "vaults": ["*"],
+                "active": True,
+                "created_at": "2026-01-01T00:00:00",
+            },
+        "normaluser": {
+            "id": "user-1",
+            "username": "normaluser",
+            "display_name": "normal",
+            "password_hash": hash_password("normal123"),
+            "role": "user",
+            "vaults": ["TestVault"],
+            "active": True,
+            "created_at": "2026-01-01T00:00:00",
+        },
+        },
+    }
+    (data_dir / "users.json").write_text(json.dumps(users), encoding="utf-8")
+
+    src_secret = Path("data/secret.key")
+    if src_secret.exists():
+        shutil.copy2(str(src_secret), str(data_dir / "secret.key"))
+
+    orig_cwd = os.getcwd()
+    test_vault_path = os.path.abspath("test-vault")
+    os.chdir(str(tmp_path))
+
+    os.environ["VAULT_1_NAME"] = "TestVault"
+    os.environ["VAULT_1_PATH"] = test_vault_path
+    os.environ["OBSIGATE_AUTH_ENABLED"] = "true"
+    os.environ["OBSIGATE_ADMIN_USER"] = "admin"
+    os.environ["OBSIGATE_ADMIN_PASSWORD"] = "chab30"
+    os.environ["OBSIGATE_WATCHER_ENABLED"] = "false"
+
+    import backend.main
+    backend.main._load_config = lambda: {"watcher_enabled": False}
+
+    from backend.main import app
+    from backend.indexer import build_index, index
+    import asyncio
+
+    for key in list(index.keys()):
+        del index[key]
+
+    loop = asyncio.new_event_loop()
+    asyncio.set_event_loop(loop)
+    loop.run_until_complete(build_index())
+
+    from backend.search import init_inverted_index
+    init_inverted_index()
+
+    from fastapi.testclient import TestClient
+    client = TestClient(app)
+    yield client
+
+    if hasattr(client, "close"):
+        client.close()
+    loop.run_until_complete(asyncio.sleep(0))
+
+    os.chdir(orig_cwd)
+    shutil.rmtree(str(tmp_path), ignore_errors=True)
+    for k in [
+        "VAULT_1_NAME", "VAULT_1_PATH", "OBSIGATE_AUTH_ENABLED",
+        "OBSIGATE_ADMIN_USER", "OBSIGATE_ADMIN_PASSWORD", "OBSIGATE_WATCHER_ENABLED",
+    ]:
+        os.environ.pop(k, None)
+
+
+def _login(client, username="admin", password="chab30"):
+    resp = client.post("/api/auth/login", json={"username": username, "password": password})
+    assert resp.status_code == 200, resp.text
+    return resp.json()["access_token"]
+
+
+def _bearer(token):
+    return {"Authorization": f"Bearer {token}"}
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/stats
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminStats:
+    EXPECTED_KEYS = {
+        "cpu_pct", "mem_used_mb", "mem_total_mb",
+        "disk_used_gb", "disk_total_gb",
+        "uptime_seconds", "active_sessions",
+    }
+
+    def test_stats_ok_as_admin(self, admin_client):
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        missing = self.EXPECTED_KEYS - set(data.keys())
+        assert not missing, f"Missing keys: {missing}"
+        # Numeric sanity (allow 0 for any metric — depending on platform)
+        for key in ("cpu_pct", "mem_used_mb", "mem_total_mb",
+                    "disk_used_gb", "disk_total_gb", "uptime_seconds",
+                    "active_sessions"):
+            assert isinstance(data[key], (int, float)), f"{key} not numeric"
+            assert data[key] >= 0, f"{key} is negative"
+
+    def test_stats_requires_auth(self, admin_client):
+        resp = admin_client.get("/api/admin/stats")
+        assert resp.status_code in (401, 403)
+
+    def test_stats_requires_admin_role(self, admin_client):
+        token = _login(admin_client, username="normaluser", password="normal123")
+        resp = admin_client.get("/api/admin/stats", headers=_bearer(token))
+        assert resp.status_code == 403
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/audit
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminAudit:
+    def _seed_audit(self, tmp_path_factory=None):
+        """Append a few entries to the audit log so filtering has data."""
+        from backend.audit import _write_entry
+        from datetime import datetime, timezone
+        entries = [
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_save", "username": "alice", "vault": "TestVault",
+             "size": 100, "ip": "127.0.0.1"},
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_delete", "username": "bob", "vault": "TestVault",
+             "ip": "127.0.0.1"},
+            {"timestamp": datetime.now(timezone.utc).isoformat(),
+             "action": "file_save", "username": "bob", "vault": "TestVault",
+             "size": 50, "ip": "127.0.0.1"},
+        ]
+        for e in entries:
+            _write_entry(e)
+
+    def test_audit_ok(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert "entries" in data
+        assert "total" in data
+        # We just seeded at least 3 entries
+        assert data["total"] >= 3
+
+    def test_audit_filter_by_user(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?user=bob", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert all("bob" in str(e.get("username", "")).lower() for e in data["entries"])
+
+    def test_audit_filter_by_action(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?action=file_delete", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert all(e.get("action") == "file_delete" for e in data["entries"])
+
+    def test_audit_limit_param(self, admin_client):
+        self._seed_audit()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/audit?limit=2", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert len(data["entries"]) <= 2
+
+    def test_audit_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/audit")
+        assert resp.status_code in (401, 403)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/backup-stats
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminBackupStats:
+    def _create_backup_files(self, tmp_path_factory=None):
+        """Create a couple of fake .bak files in the default backup dir."""
+        from backend.indexer import vault_config
+        import time as _time
+        for vault_name, cfg in list(vault_config.items()):
+            vault_root = Path(cfg["path"])
+            backup_root = vault_root / ".obsigate-backup" / vault_name / "subdir"
+            backup_root.mkdir(parents=True, exist_ok=True)
+            ts1 = int(_time.time()) - 86400
+            ts2 = int(_time.time())
+            (backup_root / f"note.md.{ts1}.bak").write_text("old version")
+            (backup_root / f"note.md.{ts2}.bak").write_text("newer version with more content")
+
+    def test_backup_stats_ok(self, admin_client):
+        self._create_backup_files()
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/backup-stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        for k in ("total_backups", "total_size_mb", "oldest_age_days",
+                  "newest_age_days", "by_vault"):
+            assert k in data, f"missing key {k}"
+        assert data["total_backups"] >= 2
+        assert data["total_size_mb"] >= 0
+        # We created one set per vault — at least one vault entry
+        assert isinstance(data["by_vault"], dict)
+
+    def test_backup_stats_empty(self, admin_client):
+        """Even with no backups, endpoint returns 200 with zero counts."""
+        token = _login(admin_client)
+        resp = admin_client.get("/api/admin/backup-stats", headers=_bearer(token))
+        assert resp.status_code == 200
+        data = resp.json()
+        assert data["total_backups"] >= 0
+        # If no backups exist, both age fields are 0
+        if data["total_backups"] == 0:
+            assert data["newest_age_days"] == 0.0
+            assert data["oldest_age_days"] == 0.0
+
+    def test_backup_stats_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/backup-stats")
+        assert resp.status_code in (401, 403)
+
+
+# ═══════════════════════════════════════════════════════════════════
+# /api/admin/stream
+# ═══════════════════════════════════════════════════════════════════
+
+class TestAdminStream:
+    def test_stream_content_type(self, admin_client):
+        """Verify SSE endpoint returns text/event-stream with valid first frame.
+
+        The /api/admin/stream endpoint is an infinite generator (yields every 5s).
+        We can't easily consume a streaming response from a sync TestClient
+        (the context manager blocks on entry for infinite responses). Instead,
+        we verify the endpoint contract from two angles:
+          1. Without auth → 401/403 (proves the endpoint is mounted and gated)
+          2. Direct invocation of the underlying generator yields a valid SSE
+             frame on the first iteration (proves the format contract).
+        """
+        # 1) Endpoint is gated behind admin auth.
+        resp = admin_client.get("/api/admin/stream")
+        assert resp.status_code in (401, 403), (
+            f"unauthenticated should be rejected, got {resp.status_code}"
+        )
+
+        # 2) Direct generator check — read the first frame, then close.
+        import asyncio
+        from backend.admin import _stats_event_generator
+
+        async def _first_frame():
+            gen = _stats_event_generator()
+            return await gen.__anext__()
+
+        first = asyncio.run(_first_frame())
+        assert isinstance(first, str), f"expected str, got {type(first).__name__}"
+        assert first.startswith("event: stats"), f"unexpected frame: {first[:100]!r}"
+        assert "data: " in first
+        # The data line should be parseable JSON.
+        import json
+        data_line = [ln for ln in first.splitlines() if ln.startswith("data: ")][0]
+        payload = json.loads(data_line[len("data: "):])
+        assert isinstance(payload, dict)
+        assert "cpu_pct" in payload or "error" in payload
+
+    def test_stream_requires_admin(self, admin_client):
+        resp = admin_client.get("/api/admin/stream")
+        assert resp.status_code in (401, 403)
\ No newline at end of file