diff --git a/Cargo.lock b/Cargo.lock
index c460cd2..645a166 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -632,6 +632,7 @@ dependencies = [
  "anyhow",
  "axum",
  "futures",
+ "getrandom 0.4.3",
  "google-gmail1",
  "lazy_static",
  "regex",
@@ -641,6 +642,7 @@ dependencies = [
  "sqlx",
  "tempfile",
  "tokio",
+ "tower",
  "yup-oauth2",
 ]
 
diff --git a/Cargo.toml b/Cargo.toml
index ea5789b..caea790 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -9,6 +9,9 @@ edition = "2021"
 anyhow = "1.0"
 axum = "0.8"
 futures = "0.3"
+# CSPRNG for the viewer's per-process CSRF token (already in the tree
+# transitively via the TLS stack; this only makes it a direct dependency).
+getrandom = "0.4"
 google-gmail1 = "7.0"
 lazy_static = "1.5"
 regex = "1.13"
@@ -23,3 +26,5 @@ yup-oauth2 = "12"
 
 [dev-dependencies]
 tempfile = "3"
+# tower::ServiceExt::oneshot for exercising the axum router in endpoint tests.
+tower = { version = "0.5", features = ["util"] }
diff --git a/src/bin/web.rs b/src/bin/web.rs
index 626868c..204a289 100644
--- a/src/bin/web.rs
+++ b/src/bin/web.rs
@@ -1,15 +1,23 @@
-//! Local web viewer for gmail-stats (Phase 1 of issue #11).
+//! Local web viewer for gmail-stats (issue #11 Phase 1, issue #28 Phase B).
 //!
-//! Serves one embedded HTML page and one JSON endpoint over `./stats.db`,
-//! strictly read-only, bound to 127.0.0.1 only.
+//! Serves the embedded UI plus read-only JSON endpoints over `./stats.db`,
+//! bound to 127.0.0.1 only. Phase B adds observe-only ingestion state:
+//! `GET /api/status` (db readiness, active/last run, flock probe, viewer-side
+//! rate/ETA, CSRF token slot) and `GET /api/runs` (run history). The viewer
+//! performs no database writes of any kind; the only file it touches is the
+//! ingest lockfile, probed without truncation via open + try-lock + release.
 //!
 //! Usage: `cargo run --bin web [port]` (default port 7878), then open the
-//! printed URL. Run from the repo root so `./stats.db` resolves.
+//! printed URL. Run from the repo root so `./stats.db` resolves, or point
+//! GMAIL_STATS_DB at the database.
 
-use std::time::{Duration, SystemTime, UNIX_EPOCH};
+use std::collections::VecDeque;
+use std::path::{Path, PathBuf};
+use std::sync::{Arc, Mutex};
+use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
 
 use axum::{
-    extract::{Request, State},
+    extract::{Query, Request, State},
     http::{header, HeaderValue, StatusCode},
     middleware::{self, Next},
     response::{Html, IntoResponse, Response},
@@ -17,8 +25,10 @@ use axum::{
     Json, Router,
 };
 use serde_json::json;
-use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
-use sqlx::SqlitePool;
+use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions, SqliteRow};
+use sqlx::{Row, SqlitePool};
+
+use gmail_stats::ingest;
 
 // CSS/JS are separate same-origin assets, not inline blocks: under the
 // `default-src 'self'` CSP below, `'self'` only allowlists same-origin URLs —
@@ -28,6 +38,28 @@ const APP_CSS: &str = include_str!("../../web/app.css");
 const APP_JS: &str = include_str!("../../web/app.js");
 const DEFAULT_PORT: u16 = 7878;
 
+/// How far back the in-memory rate window looks. Old observations age out so
+/// the displayed rate tracks the recent pace, not the whole run's average.
+const RATE_WINDOW: Duration = Duration::from_secs(30);
+/// Minimum span between oldest and newest observation before a rate is
+/// reported at all; two samples a few ms apart would just be noise.
+const RATE_MIN_SPAN: Duration = Duration::from_millis(500);
+
+/// Shared state for the router. Everything is read-only against the database;
+/// the rate window lives purely in viewer memory and is never persisted.
+struct AppState {
+    pool: SqlitePool,
+    db_path: PathBuf,
+    /// Per-process CSRF token (32 bytes of CSPRNG output, hex-encoded),
+    /// minted at startup and served via /api/status. No endpoint consumes it
+    /// yet — the first state-changing routes (Phase C) will require it in a
+    /// custom header, and shipping the slot now means clients can already
+    /// learn it the intended way (same-origin GET only; a cross-origin page
+    /// cannot read this response).
+    csrf_token: String,
+    rate: Mutex<RateWindow>,
+}
+
 #[tokio::main]
 async fn main() -> anyhow::Result<()> {
     let port: u16 = match std::env::args().nth(1) {
@@ -36,30 +68,10 @@ async fn main() -> anyhow::Result<()> {
             .map_err(|_| anyhow::anyhow!("invalid port: {arg}"))?,
         None => DEFAULT_PORT,
     };
+    let db_path =
+        PathBuf::from(std::env::var("GMAIL_STATS_DB").unwrap_or_else(|_| "./stats.db".to_string()));
 
-    // Read-only at the connection level, not by convention: the viewer must
-    // not be able to touch a mid-scan DB even via a bug.
-    let options = SqliteConnectOptions::new()
-        .filename("./stats.db")
-        .read_only(true)
-        .pragma("query_only", "ON")
-        .create_if_missing(false)
-        .busy_timeout(Duration::from_secs(5));
-
-    // Lazy connect so the server still starts (and can serve the friendly
-    // setup page) when stats.db doesn't exist yet.
-    let pool = SqlitePoolOptions::new()
-        .max_connections(2)
-        .connect_lazy_with(options);
-
-    let app = Router::new()
-        .route("/", get(index))
-        .route("/app.css", get(app_css))
-        .route("/app.js", get(app_js))
-        .route("/api/summary", get(summary))
-        .fallback(not_found)
-        .layer(middleware::from_fn(host_guard_and_security_headers))
-        .with_state(pool);
+    let app = build_router(Arc::new(AppState::new(&db_path)));
 
     // Loopback only, structurally. The port is the only knob.
     let listener = tokio::net::TcpListener::bind(("127.0.0.1", port)).await?;
@@ -71,6 +83,49 @@ async fn main() -> anyhow::Result<()> {
     Ok(())
 }
 
+impl AppState {
+    fn new(db_path: &Path) -> Self {
+        // Read-only at the connection level, not by convention: the viewer
+        // must not be able to touch a mid-scan DB even via a bug.
+        let options = SqliteConnectOptions::new()
+            .filename(db_path)
+            .read_only(true)
+            .pragma("query_only", "ON")
+            .create_if_missing(false)
+            .busy_timeout(Duration::from_secs(5));
+
+        // Lazy connect so the server still starts (and can serve the friendly
+        // onboarding page) when stats.db doesn't exist yet.
+        let pool = SqlitePoolOptions::new()
+            .max_connections(2)
+            .connect_lazy_with(options);
+
+        let mut token_bytes = [0u8; 32];
+        getrandom::fill(&mut token_bytes).expect("operating system CSPRNG unavailable");
+        let csrf_token: String = token_bytes.iter().map(|b| format!("{b:02x}")).collect();
+
+        AppState {
+            pool,
+            db_path: db_path.to_path_buf(),
+            csrf_token,
+            rate: Mutex::new(RateWindow::default()),
+        }
+    }
+}
+
+fn build_router(state: Arc<AppState>) -> Router {
+    Router::new()
+        .route("/", get(index))
+        .route("/app.css", get(app_css))
+        .route("/app.js", get(app_js))
+        .route("/api/summary", get(summary))
+        .route("/api/status", get(status))
+        .route("/api/runs", get(runs))
+        .fallback(not_found)
+        .layer(middleware::from_fn(host_guard_and_security_headers))
+        .with_state(state)
+}
+
 /// DNS-rebinding guard plus defense-in-depth headers on every response.
 async fn host_guard_and_security_headers(req: Request, next: Next) -> Response {
     let host_allowed = req
@@ -122,8 +177,8 @@ async fn not_found() -> StatusCode {
     StatusCode::NOT_FOUND
 }
 
-async fn summary(State(pool): State<SqlitePool>) -> Response {
-    match build_summary(&pool).await {
+async fn summary(State(state): State<Arc<AppState>>) -> Response {
+    match build_summary(&state.pool).await {
         Ok(body) => Json(body).into_response(),
         Err(err) => error_response(&err),
     }
@@ -161,18 +216,351 @@ async fn build_summary(pool: &SqlitePool) -> Result<serde_json::Value, sqlx::Err
         .map(|(sender, mails_sent)| json!({ "sender": sender, "mails_sent": mails_sent }))
         .collect();
 
-    let generated_at_unix = SystemTime::now()
-        .duration_since(UNIX_EPOCH)
-        .map(|d| d.as_secs())
-        .unwrap_or(0);
-
     Ok(json!({
         "total_messages": total_messages,
         "senders": senders,
-        "generated_at_unix": generated_at_unix,
+        "generated_at_unix": now_unix(),
     }))
 }
 
+// ---------------------------------------------------------------------------
+// GET /api/status — the single cheap poll target for "watch from the browser"
+// ---------------------------------------------------------------------------
+
+/// How a database-level error should be interpreted.
+enum DbErrorKind {
+    /// The file (or a table the query needs) doesn't exist.
+    MissingFile,
+    MissingTable,
+    Busy,
+    Other,
+}
+
+fn classify_db_error(err: &sqlx::Error) -> DbErrorKind {
+    let lower = err.to_string().to_lowercase();
+    if lower.contains("unable to open database file") {
+        DbErrorKind::MissingFile
+    } else if lower.contains("no such table") {
+        DbErrorKind::MissingTable
+    } else if lower.contains("database is locked") || lower.contains("database is busy") {
+        DbErrorKind::Busy
+    } else {
+        DbErrorKind::Other
+    }
+}
+
+async fn status(State(state): State<Arc<AppState>>) -> Response {
+    match build_status(&state).await {
+        Ok(body) => Json(body).into_response(),
+        Err(err) => error_response(&err),
+    }
+}
+
+/// Assemble the status document. Missing database, missing tables, and empty
+/// tables are all *data* here (they drive onboarding), never errors; only
+/// busy/unexpected failures propagate to the caller as HTTP errors.
+async fn build_status(state: &AppState) -> Result<serde_json::Value, sqlx::Error> {
+    let lock_held = probe_ingest_lock(&state.db_path);
+
+    // Database readiness. `unable to open` ⇒ the file is missing (the pool is
+    // read-only and never creates it); `no such table` ⇒ the file exists but
+    // was never initialized — both are onboarding states, not errors.
+    let db: &str =
+        match sqlx::query_scalar::<_, i64>("SELECT COUNT(DISTINCT mail_id) FROM seen_mails")
+            .fetch_one(&state.pool)
+            .await
+        {
+            Ok(0) => "empty",
+            Ok(_) => "ready",
+            Err(err) => match classify_db_error(&err) {
+                DbErrorKind::MissingFile => {
+                    return Ok(status_document(
+                        state, "missing", None, None, false, lock_held,
+                    ));
+                }
+                DbErrorKind::MissingTable => "empty",
+                DbErrorKind::Busy | DbErrorKind::Other => return Err(err),
+            },
+        };
+
+    // Run rows. An old database without ingest_runs simply has no runs; the
+    // ingester creates the table on its next start.
+    let (active_run, last_run, mixed_sources) = match fetch_runs_view(&state.pool).await {
+        Ok(view) => view,
+        Err(err) => match classify_db_error(&err) {
+            DbErrorKind::MissingFile | DbErrorKind::MissingTable => (None, None, false),
+            DbErrorKind::Busy | DbErrorKind::Other => return Err(err),
+        },
+    };
+
+    Ok(status_document(
+        state,
+        db,
+        active_run,
+        last_run,
+        mixed_sources,
+        lock_held,
+    ))
+}
+
+async fn fetch_runs_view(
+    pool: &SqlitePool,
+) -> Result<(Option<SqliteRow>, Option<SqliteRow>, bool), sqlx::Error> {
+    let active = sqlx::query(
+        "SELECT * FROM ingest_runs \
+         WHERE state NOT IN ('done', 'failed', 'cancelled', 'abandoned') \
+         ORDER BY run_id DESC LIMIT 1",
+    )
+    .fetch_optional(pool)
+    .await?;
+    let last = sqlx::query(
+        "SELECT * FROM ingest_runs \
+         WHERE state IN ('done', 'failed', 'cancelled', 'abandoned') \
+         ORDER BY run_id DESC LIMIT 1",
+    )
+    .fetch_optional(pool)
+    .await?;
+    // Interim mixed-sources caution (removed by Phase D's exact dedupe): true
+    // once more than one source has actually contributed rows — completed, or
+    // got far enough to add new messages before stopping.
+    let mixed: i64 = sqlx::query_scalar(
+        "SELECT COUNT(DISTINCT source) FROM ingest_runs \
+         WHERE state = 'done' OR messages_new > 0",
+    )
+    .fetch_one(pool)
+    .await?;
+    Ok((active, last, mixed >= 2))
+}
+
+fn status_document(
+    state: &AppState,
+    db: &str,
+    active_run: Option<SqliteRow>,
+    last_run: Option<SqliteRow>,
+    mixed_sources: bool,
+    lock_held: Option<bool>,
+) -> serde_json::Value {
+    // Viewer-derived rate/ETA from successive observations of the active
+    // run's counters. Pure viewer memory: nothing here is written anywhere.
+    let mut rate_per_sec = None;
+    let mut eta_seconds = None;
+    {
+        let mut window = state.rate.lock().unwrap_or_else(|e| e.into_inner());
+        match &active_run {
+            Some(row) => {
+                let run_id: i64 = row.try_get("run_id").unwrap_or(0);
+                let messages_seen: i64 = row.try_get("messages_seen").unwrap_or(0);
+                let bytes_done: Option<i64> = row.try_get("bytes_done").unwrap_or(None);
+                let (msg_rate, byte_rate) =
+                    window.observe(Instant::now(), run_id, messages_seen, bytes_done);
+                rate_per_sec = msg_rate.map(|r| (r * 10.0).round() / 10.0);
+                let bytes_total: Option<i64> = row.try_get("bytes_total").unwrap_or(None);
+                let total_estimate: Option<i64> = row.try_get("total_estimate").unwrap_or(None);
+                eta_seconds = compute_eta(
+                    messages_seen,
+                    bytes_done,
+                    bytes_total,
+                    total_estimate,
+                    msg_rate,
+                    byte_rate,
+                );
+            }
+            None => window.reset(),
+        }
+    }
+
+    json!({
+        "db": db,
+        "now_unix": now_unix(),
+        "ingest_lock_held": lock_held,
+        "active_run": active_run.as_ref().map(run_to_json),
+        "last_run": last_run.as_ref().map(run_to_json),
+        // Phase B never spawns anything, so no run is ever owned by the
+        // viewer; the field exists so clients already branch on it.
+        "owns_active_run": false,
+        "mixed_sources": mixed_sources,
+        "rate_per_sec": rate_per_sec,
+        "eta_seconds": eta_seconds,
+        "csrf_token": state.csrf_token,
+    })
+}
+
+/// Serialize one ingest_runs row for the API. resume_token and
+/// mbox_fingerprint are internal resume bookkeeping and stay out of the
+/// payload; everything else is display data (rendered via textContent only on
+/// the client — error text and paths are not trusted).
+fn run_to_json(row: &SqliteRow) -> serde_json::Value {
+    fn i(row: &SqliteRow, col: &str) -> Option<i64> {
+        row.try_get::<Option<i64>, _>(col).unwrap_or(None)
+    }
+    fn s(row: &SqliteRow, col: &str) -> Option<String> {
+        row.try_get::<Option<String>, _>(col).unwrap_or(None)
+    }
+    json!({
+        "run_id": i(row, "run_id"),
+        "source": s(row, "source"),
+        "state": s(row, "state"),
+        "pid": i(row, "pid"),
+        "started_at_unix": i(row, "started_at_unix"),
+        "updated_at_unix": i(row, "updated_at_unix"),
+        "finished_at_unix": i(row, "finished_at_unix"),
+        "messages_seen": i(row, "messages_seen").unwrap_or(0),
+        "messages_new": i(row, "messages_new").unwrap_or(0),
+        "total_estimate": i(row, "total_estimate"),
+        "bytes_total": i(row, "bytes_total"),
+        "bytes_done": i(row, "bytes_done"),
+        "mbox_path": s(row, "mbox_path"),
+        "error_kind": s(row, "error_kind"),
+        "error": s(row, "error"),
+    })
+}
+
+/// Probe the ingest lock without disturbing it: open the lockfile exactly the
+/// way the ingester does (create-if-missing, never truncate — see
+/// `ingest::acquire_ingest_lock`), try a non-blocking exclusive flock, and
+/// release immediately. Held ⇒ an ingester is alive right now, whoever
+/// started it. Returns None if the probe itself failed (e.g. permissions),
+/// which the client treats as "unknown".
+fn probe_ingest_lock(db_path: &Path) -> Option<bool> {
+    let lock_path = ingest::ingest_lock_path(db_path);
+    let file = std::fs::OpenOptions::new()
+        .create(true)
+        .truncate(false)
+        .write(true)
+        .open(&lock_path)
+        .ok()?;
+    match rustix::fs::flock(&file, rustix::fs::FlockOperation::NonBlockingLockExclusive) {
+        Ok(()) => {
+            // Explicit unlock before the fd drops, so the probe's hold time
+            // is as close to zero as possible.
+            let _ = rustix::fs::flock(&file, rustix::fs::FlockOperation::Unlock);
+            Some(false)
+        }
+        Err(rustix::io::Errno::WOULDBLOCK) => Some(true),
+        Err(_) => None,
+    }
+}
+
+/// Sliding window of (time, counters) observations for the active run, kept
+/// only in viewer memory. Rates come from the span between the oldest and
+/// newest sample still inside RATE_WINDOW.
+#[derive(Default)]
+struct RateWindow {
+    run_id: Option<i64>,
+    samples: VecDeque<(Instant, i64, Option<i64>)>,
+}
+
+impl RateWindow {
+    fn reset(&mut self) {
+        self.run_id = None;
+        self.samples.clear();
+    }
+
+    /// Record an observation and return (messages/sec, bytes/sec) over the
+    /// current window, if enough history exists to be meaningful.
+    fn observe(
+        &mut self,
+        now: Instant,
+        run_id: i64,
+        messages_seen: i64,
+        bytes_done: Option<i64>,
+    ) -> (Option<f64>, Option<f64>) {
+        if self.run_id != Some(run_id) {
+            self.reset();
+            self.run_id = Some(run_id);
+        }
+        self.samples.push_back((now, messages_seen, bytes_done));
+        while let Some(&(t, _, _)) = self.samples.front() {
+            if now.duration_since(t) > RATE_WINDOW && self.samples.len() > 2 {
+                self.samples.pop_front();
+            } else {
+                break;
+            }
+        }
+
+        let (first_t, first_msgs, first_bytes) = *self.samples.front().unwrap();
+        let span = now.duration_since(first_t);
+        if span < RATE_MIN_SPAN {
+            return (None, None);
+        }
+        let secs = span.as_secs_f64();
+        let msg_rate = ((messages_seen - first_msgs).max(0) as f64) / secs;
+        let byte_rate = match (bytes_done, first_bytes) {
+            (Some(b), Some(fb)) => Some(((b - fb).max(0) as f64) / secs),
+            _ => None,
+        };
+        (Some(msg_rate), byte_rate)
+    }
+}
+
+/// Estimated seconds to completion. Prefers byte progress (exact for imports)
+/// and falls back to the message-count estimate (scans). None when there is
+/// no total or no forward progress in the window.
+fn compute_eta(
+    messages_seen: i64,
+    bytes_done: Option<i64>,
+    bytes_total: Option<i64>,
+    total_estimate: Option<i64>,
+    msg_rate: Option<f64>,
+    byte_rate: Option<f64>,
+) -> Option<i64> {
+    if let (Some(done), Some(total), Some(rate)) = (bytes_done, bytes_total, byte_rate) {
+        if rate > 0.0 {
+            return Some((((total - done).max(0)) as f64 / rate).ceil() as i64);
+        }
+    }
+    if let (Some(total), Some(rate)) = (total_estimate, msg_rate) {
+        if rate > 0.0 {
+            return Some((((total - messages_seen).max(0)) as f64 / rate).ceil() as i64);
+        }
+    }
+    None
+}
+
+// ---------------------------------------------------------------------------
+// GET /api/runs — run history
+// ---------------------------------------------------------------------------
+
+async fn runs(
+    State(state): State<Arc<AppState>>,
+    Query(params): Query<std::collections::HashMap<String, String>>,
+) -> Response {
+    let limit: i64 = params
+        .get("limit")
+        .and_then(|v| v.parse::<i64>().ok())
+        .unwrap_or(20)
+        .clamp(1, 100);
+    match build_runs(&state.pool, limit).await {
+        Ok(body) => Json(body).into_response(),
+        Err(err) => error_response(&err),
+    }
+}
+
+async fn build_runs(pool: &SqlitePool, limit: i64) -> Result<serde_json::Value, sqlx::Error> {
+    let rows = match sqlx::query("SELECT * FROM ingest_runs ORDER BY run_id DESC LIMIT ?")
+        .bind(limit)
+        .fetch_all(pool)
+        .await
+    {
+        Ok(rows) => rows,
+        // A database that predates ingest_runs (or doesn't exist yet) simply
+        // has no run history; this endpoint powers a strip, not a diagnosis.
+        Err(err) => match classify_db_error(&err) {
+            DbErrorKind::MissingFile | DbErrorKind::MissingTable => Vec::new(),
+            DbErrorKind::Busy | DbErrorKind::Other => return Err(err),
+        },
+    };
+    let runs: Vec<serde_json::Value> = rows.iter().map(run_to_json).collect();
+    Ok(json!({ "runs": runs, "now_unix": now_unix() }))
+}
+
+fn now_unix() -> u64 {
+    SystemTime::now()
+        .duration_since(UNIX_EPOCH)
+        .map(|d| d.as_secs())
+        .unwrap_or(0)
+}
+
 fn error_response(err: &sqlx::Error) -> Response {
     let detail = err.to_string();
     let lower = detail.to_lowercase();
@@ -206,6 +594,11 @@ fn error_response(err: &sqlx::Error) -> Response {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use axum::body::Body;
+    use axum::http::Request as HttpRequest;
+    use sqlx::sqlite::SqliteJournalMode;
+    use sqlx::{Connection, SqliteConnection};
+    use tower::ServiceExt;
 
     async fn memory_pool() -> SqlitePool {
         // One connection max: each sqlite::memory: connection is a separate
@@ -276,4 +669,351 @@ mod tests {
         assert_eq!(senders[0]["sender"], "");
         assert_eq!(senders[0]["mails_sent"], 5);
     }
+
+    // -- Phase B endpoint tests ---------------------------------------------
+
+    /// Build a real on-disk database with Phase A's actual DDL (via
+    /// `ingest::migrate`), so the tests exercise the same schema the ingester
+    /// produces.
+    async fn migrated_db(dir: &tempfile::TempDir) -> PathBuf {
+        let db_path = dir.path().join("stats.db");
+        let options = SqliteConnectOptions::new()
+            .filename(&db_path)
+            .create_if_missing(true)
+            .journal_mode(SqliteJournalMode::Wal);
+        let mut conn = SqliteConnection::connect_with(&options).await.unwrap();
+        ingest::migrate(&mut conn).await.unwrap();
+        conn.close().await.ok();
+        db_path
+    }
+
+    async fn exec(db_path: &Path, sql: &str) {
+        let options = SqliteConnectOptions::new()
+            .filename(db_path)
+            .journal_mode(SqliteJournalMode::Wal);
+        let mut conn = SqliteConnection::connect_with(&options).await.unwrap();
+        sqlx::raw_sql(sqlx::AssertSqlSafe(sql.to_string()))
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        conn.close().await.ok();
+    }
+
+    async fn get_json(app: Router, uri: &str) -> (StatusCode, serde_json::Value) {
+        let response = app
+            .oneshot(
+                HttpRequest::builder()
+                    .uri(uri)
+                    .header(header::HOST, "127.0.0.1:7878")
+                    .body(Body::empty())
+                    .unwrap(),
+            )
+            .await
+            .unwrap();
+        let status = response.status();
+        let bytes = axum::body::to_bytes(response.into_body(), 1 << 20)
+            .await
+            .unwrap();
+        let value = serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
+        (status, value)
+    }
+
+    fn app_for(db_path: &Path) -> Router {
+        build_router(Arc::new(AppState::new(db_path)))
+    }
+
+    #[tokio::test]
+    async fn status_with_missing_db_is_200_and_onboardable() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = dir.path().join("does-not-exist.db");
+        let (status, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(
+            status,
+            StatusCode::OK,
+            "status must never 5xx on missing db"
+        );
+        assert_eq!(body["db"], "missing");
+        assert!(body["active_run"].is_null());
+        assert!(body["last_run"].is_null());
+        assert_eq!(body["owns_active_run"], false);
+        assert_eq!(body["mixed_sources"], false);
+        // CSRF token slot: 32 CSPRNG bytes, hex-encoded.
+        let token = body["csrf_token"].as_str().unwrap();
+        assert_eq!(token.len(), 64);
+        assert!(token.chars().all(|c| c.is_ascii_hexdigit()));
+        // The read-only pool must not have created the file as a side effect.
+        assert!(!db_path.exists(), "viewer must not create the database");
+    }
+
+    #[tokio::test]
+    async fn status_with_empty_db_reports_empty() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        let (status, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["db"], "empty");
+        assert!(body["active_run"].is_null());
+        assert_eq!(body["ingest_lock_held"], false);
+    }
+
+    #[tokio::test]
+    async fn status_without_ingest_runs_table_treats_runs_as_none() {
+        // A pre-Phase-A database: baseline tables with data, no ingest_runs.
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = dir.path().join("old.db");
+        let options = SqliteConnectOptions::new()
+            .filename(&db_path)
+            .create_if_missing(true);
+        let mut conn = SqliteConnection::connect_with(&options).await.unwrap();
+        sqlx::query("CREATE TABLE seen_mails (mail_id string)")
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        sqlx::query("CREATE TABLE senders (sender string, mails_sent int)")
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        sqlx::query("INSERT INTO seen_mails VALUES ('m1')")
+            .execute(&mut conn)
+            .await
+            .unwrap();
+        conn.close().await.ok();
+
+        let (status, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["db"], "ready");
+        assert!(body["active_run"].is_null());
+        assert!(body["last_run"].is_null());
+
+        let (status, body) = get_json(app_for(&db_path), "/api/runs").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["runs"].as_array().unwrap().len(), 0);
+    }
+
+    #[tokio::test]
+    async fn status_with_active_and_finished_runs() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        exec(&db_path, "INSERT INTO seen_mails VALUES ('m1')").await;
+        exec(
+            &db_path,
+            "INSERT INTO ingest_runs \
+             (run_id, source, state, started_at_unix, updated_at_unix, finished_at_unix, \
+              messages_seen, messages_new, error_kind, error) \
+             VALUES (1, 'gmail_api', 'failed', 100, 150, 150, 10, 5, \
+                     'policy_enforced', 'Error 400: policy_enforced')",
+        )
+        .await;
+        exec(
+            &db_path,
+            "INSERT INTO ingest_runs \
+             (run_id, source, state, started_at_unix, updated_at_unix, \
+              messages_seen, messages_new, bytes_total, bytes_done, mbox_path) \
+             VALUES (2, 'mbox', 'running', 200, 205, 1000, 900, 50000, 25000, '/tmp/a.mbox')",
+        )
+        .await;
+
+        let (status, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["db"], "ready");
+        let active = &body["active_run"];
+        assert_eq!(active["run_id"], 2);
+        assert_eq!(active["source"], "mbox");
+        assert_eq!(active["state"], "running");
+        assert_eq!(active["messages_seen"], 1000);
+        assert_eq!(active["bytes_total"], 50000);
+        assert_eq!(active["bytes_done"], 25000);
+        let last = &body["last_run"];
+        assert_eq!(last["run_id"], 1);
+        assert_eq!(last["state"], "failed");
+        assert_eq!(last["error_kind"], "policy_enforced");
+        assert_eq!(body["owns_active_run"], false);
+        // First observation of this run: no rate window yet.
+        assert!(body["rate_per_sec"].is_null());
+        assert!(body["now_unix"].as_u64().unwrap() > 0);
+    }
+
+    #[tokio::test]
+    async fn status_mixed_sources_flags_dual_source_databases() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        exec(&db_path, "INSERT INTO seen_mails VALUES ('m1')").await;
+        exec(
+            &db_path,
+            "INSERT INTO ingest_runs (run_id, source, state, started_at_unix, updated_at_unix, \
+             messages_new) VALUES (1, 'gmail_api', 'done', 100, 150, 10)",
+        )
+        .await;
+        let (_, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(body["mixed_sources"], false);
+
+        exec(
+            &db_path,
+            "INSERT INTO ingest_runs (run_id, source, state, started_at_unix, updated_at_unix, \
+             messages_new) VALUES (2, 'mbox', 'done', 200, 250, 3)",
+        )
+        .await;
+        let (_, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(body["mixed_sources"], true);
+    }
+
+    #[tokio::test]
+    async fn status_lock_probe_detects_a_running_ingester_without_truncating() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        let lock_path = ingest::ingest_lock_path(&db_path);
+        // Pre-existing lockfile content must survive the probe (no truncate).
+        std::fs::write(&lock_path, b"sentinel").unwrap();
+
+        let (_, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(body["ingest_lock_held"], false);
+
+        // A live ingester holds the flock; the probe must see it and must not
+        // steal or break it.
+        let held = ingest::acquire_ingest_lock(&db_path).unwrap();
+        let (_, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(body["ingest_lock_held"], true);
+        drop(held);
+
+        let (_, body) = get_json(app_for(&db_path), "/api/status").await;
+        assert_eq!(body["ingest_lock_held"], false);
+        assert_eq!(
+            std::fs::read(&lock_path).unwrap(),
+            b"sentinel",
+            "probe must never truncate the lockfile"
+        );
+    }
+
+    #[tokio::test]
+    async fn runs_endpoint_orders_and_limits() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        for id in 1..=5 {
+            exec(
+                &db_path,
+                &format!(
+                    "INSERT INTO ingest_runs (run_id, source, state, started_at_unix, \
+                     updated_at_unix) VALUES ({id}, 'mbox', 'done', {id}00, {id}50)"
+                ),
+            )
+            .await;
+        }
+        let (status, body) = get_json(app_for(&db_path), "/api/runs?limit=3").await;
+        assert_eq!(status, StatusCode::OK);
+        let runs = body["runs"].as_array().unwrap();
+        assert_eq!(runs.len(), 3);
+        // Newest first.
+        assert_eq!(runs[0]["run_id"], 5);
+        assert_eq!(runs[1]["run_id"], 4);
+        assert_eq!(runs[2]["run_id"], 3);
+
+        // Default limit, garbage limit, and missing db all behave.
+        let (_, body) = get_json(app_for(&db_path), "/api/runs").await;
+        assert_eq!(body["runs"].as_array().unwrap().len(), 5);
+        let (_, body) = get_json(app_for(&db_path), "/api/runs?limit=bogus").await;
+        assert_eq!(body["runs"].as_array().unwrap().len(), 5);
+        let missing = dir.path().join("nope.db");
+        let (status, body) = get_json(app_for(&missing), "/api/runs").await;
+        assert_eq!(status, StatusCode::OK);
+        assert_eq!(body["runs"].as_array().unwrap().len(), 0);
+    }
+
+    #[tokio::test]
+    async fn new_routes_keep_security_headers_and_emit_no_cors() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        for uri in ["/api/status", "/api/runs"] {
+            let response = app_for(&db_path)
+                .oneshot(
+                    HttpRequest::builder()
+                        .uri(uri)
+                        .header(header::HOST, "127.0.0.1:7878")
+                        .header(header::ORIGIN, "https://evil.example")
+                        .body(Body::empty())
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            let headers = response.headers();
+            assert_eq!(
+                headers.get(header::CONTENT_SECURITY_POLICY).unwrap(),
+                "default-src 'self'",
+                "{uri} lost the CSP"
+            );
+            assert_eq!(
+                headers.get(header::X_CONTENT_TYPE_OPTIONS).unwrap(),
+                "nosniff"
+            );
+            assert_eq!(headers.get(header::REFERRER_POLICY).unwrap(), "no-referrer");
+            assert!(
+                headers.get(header::ACCESS_CONTROL_ALLOW_ORIGIN).is_none(),
+                "{uri} must never emit CORS headers"
+            );
+            assert!(headers.get(header::ACCESS_CONTROL_ALLOW_METHODS).is_none());
+        }
+    }
+
+    #[tokio::test]
+    async fn new_routes_respect_the_host_guard() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        for uri in ["/api/status", "/api/runs"] {
+            let response = app_for(&db_path)
+                .oneshot(
+                    HttpRequest::builder()
+                        .uri(uri)
+                        .header(header::HOST, "attacker.example")
+                        .body(Body::empty())
+                        .unwrap(),
+                )
+                .await
+                .unwrap();
+            assert_eq!(response.status(), StatusCode::FORBIDDEN, "{uri}");
+        }
+    }
+
+    #[tokio::test]
+    async fn csrf_token_is_stable_per_process_state() {
+        let dir = tempfile::tempdir().unwrap();
+        let db_path = migrated_db(&dir).await;
+        let state = Arc::new(AppState::new(&db_path));
+        let (_, a) = get_json(build_router(state.clone()), "/api/status").await;
+        let (_, b) = get_json(build_router(state.clone()), "/api/status").await;
+        assert_eq!(a["csrf_token"], b["csrf_token"]);
+        // And two distinct viewer processes mint distinct tokens.
+        let other = Arc::new(AppState::new(&db_path));
+        assert_ne!(state.csrf_token, other.csrf_token);
+    }
+
+    #[test]
+    fn rate_window_computes_rates_and_resets_across_runs() {
+        let mut window = RateWindow::default();
+        let t0 = Instant::now();
+        // First sample: no span yet.
+        assert_eq!(window.observe(t0, 1, 100, Some(1000)), (None, None));
+        // Two seconds later, 80 more messages and 8000 more bytes.
+        let (msg, bytes) = window.observe(t0 + Duration::from_secs(2), 1, 180, Some(9000));
+        assert_eq!(msg, Some(40.0));
+        assert_eq!(bytes, Some(4000.0));
+
+        // ETA from bytes: (50000-9000)/4000 = 10.25 → 11s.
+        let eta = compute_eta(180, Some(9000), Some(50_000), None, msg, bytes);
+        assert_eq!(eta, Some(11));
+        // ETA from message estimate when no byte totals exist.
+        let eta = compute_eta(180, None, None, Some(580), msg, None);
+        assert_eq!(eta, Some(10));
+        // No forward progress ⇒ no ETA rather than a division blowup.
+        let (msg, _) = window.observe(t0 + Duration::from_secs(3), 1, 180, Some(9000));
+        assert!(eta_is_none_when_rate_zero(msg));
+
+        // A new run id resets the window.
+        assert_eq!(
+            window.observe(t0 + Duration::from_secs(4), 2, 5, None),
+            (None, None)
+        );
+    }
+
+    fn eta_is_none_when_rate_zero(msg_rate: Option<f64>) -> bool {
+        compute_eta(180, None, None, Some(580), msg_rate.map(|_| 0.0), None).is_none()
+    }
 }
diff --git a/src/lib.rs b/src/lib.rs
new file mode 100644
index 0000000..2975970
--- /dev/null
+++ b/src/lib.rs
@@ -0,0 +1,10 @@
+//! Shared library surface for the gmail_stats binaries.
+//!
+//! The scanner/importer binary (`src/main.rs`) and the web viewer
+//! (`src/bin/web.rs`) both need the ingestion coordination primitives —
+//! lockfile naming, the `ingest_runs` schema, migrations — so they live here
+//! rather than being duplicated. The viewer only ever uses them read-only
+//! (plus a non-destructive flock probe); all writes stay in the ingester.
+
+pub mod ingest;
+pub mod mbox;
diff --git a/src/main.rs b/src/main.rs
index 05b6315..ec20601 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -18,10 +18,8 @@ use tokio::sync::{mpsc, Mutex};
 use tokio::task;
 use tokio::time::{Interval, MissedTickBehavior};
 
-mod ingest;
-mod mbox;
-
-use ingest::{seen_mail, SeenMail, WriteMsg};
+use gmail_stats::ingest::{self, seen_mail, SeenMail, WriteMsg};
+use gmail_stats::mbox;
 
 type GmailHub =
     Gmail<hyper_rustls::HttpsConnector<hyper_util::client::legacy::connect::HttpConnector>>;
diff --git a/web/app.css b/web/app.css
index f42752e..45cf7e6 100644
--- a/web/app.css
+++ b/web/app.css
@@ -125,3 +125,132 @@ tr.hidden-row td.act { opacity: 1; }
   border-radius: 6px;
   padding: 0.25rem 0.4rem;
 }
+
+/* --- Phase B: observe-only ingestion UI ---------------------------------- */
+
+.page-head { display: flex; align-items: baseline; gap: 0.75rem; flex-wrap: wrap; }
+
+.pill {
+  display: inline-flex;
+  align-items: center;
+  gap: 0.4rem;
+  background: var(--card);
+  border: 1px solid var(--accent);
+  border-radius: 999px;
+  padding: 0.15rem 0.7rem;
+  font-size: 0.85rem;
+  white-space: nowrap;
+}
+.pill .dot {
+  width: 0.5rem;
+  height: 0.5rem;
+  border-radius: 50%;
+  background: var(--accent);
+  animation: pulse 1.2s ease-in-out infinite;
+}
+@keyframes pulse { 50% { opacity: 0.35; } }
+
+.badge {
+  font-size: 0.8rem;
+  color: var(--muted);
+  text-transform: uppercase;
+  letter-spacing: 0.04em;
+}
+.chip {
+  display: inline-block;
+  border-radius: 999px;
+  border: 1px solid var(--border);
+  padding: 0.05rem 0.55rem;
+  font-size: 0.8rem;
+}
+.chip-running, .chip-starting { border-color: var(--accent); color: var(--accent); }
+.chip-awaiting_auth { border-color: #d97706; color: #d97706; }
+.chip-done { border-color: #16a34a; color: #16a34a; }
+.chip-failed, .chip-abandoned { border-color: #dc2626; color: #dc2626; }
+.chip-cancelled, .chip-other { color: var(--muted); }
+
+#ingest-panel summary {
+  cursor: pointer;
+  display: flex;
+  align-items: center;
+  gap: 0.6rem;
+  list-style-position: inside;
+}
+.run-body { margin-top: 0.75rem; }
+.run-meta { display: flex; gap: 1rem; flex-wrap: wrap; margin: 0.5rem 0 0.25rem; }
+
+.bar {
+  height: 0.5rem;
+  border-radius: 999px;
+  background: var(--bg);
+  border: 1px solid var(--border);
+  overflow: hidden;
+  position: relative;
+}
+.bar-fill {
+  height: 100%;
+  background: var(--accent);
+  border-radius: 999px;
+  transition: width 0.6s ease;
+}
+.bar.indeterminate .bar-fill {
+  position: absolute;
+  animation: sweep 1.4s ease-in-out infinite;
+}
+@keyframes sweep {
+  0% { left: -30%; }
+  100% { left: 100%; }
+}
+
+.warn { color: #d97706; }
+.warn-banner {
+  background: color-mix(in srgb, #d97706 12%, var(--card));
+  border: 1px solid #d97706;
+  border-radius: 8px;
+  padding: 0.6rem 0.9rem;
+  margin: 0 0 1rem;
+  font-size: 0.9rem;
+}
+#generated-at.stale { opacity: 0.45; }
+
+.cards { display: flex; gap: 0.75rem; flex-wrap: wrap; margin: 1rem 0; }
+.card {
+  flex: 1 1 18rem;
+  background: var(--bg);
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 0.9rem 1rem;
+}
+.card h3 { margin: 0 0 0.5rem; font-size: 1rem; }
+.card ol { margin: 0.5rem 0; padding-left: 1.2rem; }
+.cmd { display: flex; gap: 0.5rem; align-items: stretch; }
+.cmd pre { flex: 1 1 auto; margin: 0.5rem 0; background: var(--card); }
+.cmd .copy-btn { align-self: center; white-space: nowrap; }
+.callout {
+  border-left: 3px solid #d97706;
+  padding-left: 0.6rem;
+  font-size: 0.9rem;
+  color: var(--muted);
+}
+
+#history { margin-top: 1.5rem; }
+.history-title { font-size: 1rem; margin: 0 0 0.5rem; color: var(--muted); }
+#history-list { list-style: none; margin: 0; padding: 0; }
+#history-list li {
+  display: flex;
+  gap: 0.75rem;
+  align-items: baseline;
+  flex-wrap: wrap;
+  background: var(--card);
+  border: 1px solid var(--border);
+  border-radius: 8px;
+  padding: 0.45rem 0.75rem;
+  margin-bottom: 0.4rem;
+  font-size: 0.9rem;
+}
+.history-error {
+  flex-basis: 100%;
+  color: #dc2626;
+  font-size: 0.85rem;
+  overflow-wrap: anywhere;
+}
diff --git a/web/app.js b/web/app.js
index d99f70a..d70e25f 100644
--- a/web/app.js
+++ b/web/app.js
@@ -1,13 +1,29 @@
 "use strict";
 
-// Sender strings come from arbitrary From: headers and are attacker-controlled.
-// This page only ever renders them via textContent — never innerHTML.
+// Sender strings come from arbitrary From: headers and are attacker-controlled;
+// run error text and mbox paths from /api/status are equally untrusted. This
+// page only ever renders dynamic strings via textContent — never innerHTML.
 
 const PAGE_SIZES = [25, 50, 100, 500];
 const DEFAULT_PAGE_SIZE = 50;
 const HIDDEN_KEY = "gmail-stats.hiddenSenders";
 const PAGE_SIZE_KEY = "gmail-stats.pageSize";
 
+// Ingestion status polling cadence (issue #26): 1s while a run is active,
+// decaying to 10s when idle, paused entirely while the tab is hidden.
+const STATUS_POLL_ACTIVE_MS = 1000;
+const STATUS_POLL_IDLE_MAX_MS = 10000;
+// During an active run the summary re-fetches every ~3-5s so the table fills
+// in live without hammering the DB.
+const SUMMARY_REFRESH_MS = 3500;
+// A heartbeat older than this is displayed as "stalled?" (display-only).
+const STALL_AFTER_S = 15;
+
+const RUN_STATES = [
+  "starting", "awaiting_auth", "running",
+  "done", "failed", "cancelled", "abandoned",
+];
+
 const views = ["loading", "setup", "error", "stats"];
 
 const state = {
@@ -17,6 +33,21 @@ const state = {
   pageSize: loadPageSize(),
   showHidden: false,
   hidden: loadHidden(),
+  view: "loading",
+};
+
+// Everything the observe-only ingestion UI knows, all viewer-side.
+const ingest = {
+  // False once /api/status turns out not to exist (the static Pages demo, or
+  // any non-viewer host): every piece of ingestion UI hides and polling stops.
+  available: true,
+  status: null,
+  runsNow: 0,
+  delay: STATUS_POLL_ACTIVE_MS,
+  timer: null,
+  lastSummaryAt: 0,
+  runsKey: null,
+  wasActive: false,
 };
 
 function loadPageSize() {
@@ -48,6 +79,7 @@ function saveHidden() {
 }
 
 function showView(name) {
+  state.view = name;
   for (const id of views) {
     document.getElementById(id).classList.toggle("hidden", id !== name);
   }
@@ -93,7 +125,376 @@ function senderRow(row, isHidden) {
   return tr;
 }
 
+// --- ingestion status helpers ---------------------------------------------
+
+function activeRun() {
+  return (ingest.available && ingest.status && ingest.status.active_run) || null;
+}
+
+function ingestActive() {
+  const s = ingest.available ? ingest.status : null;
+  return Boolean(s && (s.active_run || s.ingest_lock_held === true));
+}
+
+// Onboarding replaces the stats view when the database is missing/empty and
+// nothing is running. Only the live viewer (with /api/status) can know this.
+function shouldOnboard() {
+  const s = ingest.available ? ingest.status : null;
+  return Boolean(s && (s.db === "missing" || s.db === "empty") && !ingestActive());
+}
+
+function sourceLabel(source) {
+  if (source === "mbox") return "Takeout import";
+  if (source === "gmail_api") return "Gmail scan";
+  return source ? String(source) : "Ingestion";
+}
+
+function formatDuration(totalSeconds) {
+  const s = Math.max(0, Math.round(Number(totalSeconds)));
+  if (s < 90) return `${s}s`;
+  const m = Math.round(s / 60);
+  if (m < 90) return `${m} min`;
+  return `${Math.floor(m / 60)} h ${m % 60} min`;
+}
+
+function relativeTime(nowUnix, thenUnix) {
+  const d = Math.max(0, Number(nowUnix) - Number(thenUnix));
+  if (d < 60) return "just now";
+  if (d < 3600) return `${Math.round(d / 60)} min ago`;
+  if (d < 172800) return `${Math.round(d / 3600)} h ago`;
+  return `${Math.round(d / 86400)} d ago`;
+}
+
+function stateChip(el, runState) {
+  const known = RUN_STATES.includes(runState) ? runState : "other";
+  el.className = `chip chip-${known}`;
+  el.textContent =
+    runState === "awaiting_auth" ? "awaiting authorization" : String(runState);
+}
+
+/// Fraction complete for a run, or null when there is no usable total.
+function runProgress(run) {
+  if (!run) return null;
+  const bytesTotal = Number(run.bytes_total);
+  if (bytesTotal > 0 && run.bytes_done != null) {
+    return Math.min(1, Math.max(0, Number(run.bytes_done) / bytesTotal));
+  }
+  const total = Number(run.total_estimate);
+  if (total > 0) {
+    return Math.min(1, Math.max(0, Number(run.messages_seen) / total));
+  }
+  return null;
+}
+
+function hideIngestUi() {
+  document.getElementById("ingest-pill").classList.add("hidden");
+  document.getElementById("ingest-panel").classList.add("hidden");
+  document.getElementById("history").classList.add("hidden");
+  document.getElementById("mixed-banner").classList.add("hidden");
+}
+
+function renderPill() {
+  const pill = document.getElementById("ingest-pill");
+  const run = activeRun();
+  if (!ingestActive()) {
+    pill.classList.add("hidden");
+    return;
+  }
+  const parts = [];
+  if (run) {
+    parts.push(run.source === "mbox" ? "Importing" : "Scanning");
+    parts.push(`${Number(run.messages_seen || 0).toLocaleString()} messages`);
+    const rate = ingest.status.rate_per_sec;
+    if (rate != null) parts.push(`${Number(rate).toLocaleString()}/s`);
+    const progress = runProgress(run);
+    if (progress != null) parts.push(`${Math.round(progress * 100)}%`);
+  } else {
+    // Flock held but no row yet: an ingester is just starting.
+    parts.push("Ingester starting…");
+  }
+  document.getElementById("pill-text").textContent = parts.join(" · ");
+  pill.classList.remove("hidden");
+}
+
+function renderPanel() {
+  const panel = document.getElementById("ingest-panel");
+  if (!ingestActive()) {
+    panel.classList.add("hidden");
+    return;
+  }
+  const run = activeRun();
+  const status = ingest.status;
+
+  document.getElementById("run-source").textContent =
+    run ? sourceLabel(run.source) : "Ingestion";
+  stateChip(document.getElementById("run-state"), run ? run.state : "starting");
+
+  const counts = document.getElementById("run-counts");
+  counts.textContent = run
+    ? `${Number(run.messages_seen || 0).toLocaleString()} seen · ` +
+      `${Number(run.messages_new || 0).toLocaleString()} new`
+    : "Waiting for the first progress report…";
+
+  const rateEl = document.getElementById("run-rate");
+  rateEl.textContent =
+    status.rate_per_sec != null
+      ? `${Number(status.rate_per_sec).toLocaleString()} msg/s`
+      : "";
+
+  const etaEl = document.getElementById("run-eta");
+  etaEl.textContent =
+    status.eta_seconds != null ? `~${formatDuration(status.eta_seconds)} left` : "";
+
+  const bar = document.getElementById("run-bar");
+  const fill = document.getElementById("run-bar-fill");
+  const progress = runProgress(run);
+  bar.classList.remove("hidden");
+  if (progress != null) {
+    bar.classList.remove("indeterminate");
+    fill.style.width = `${(progress * 100).toFixed(1)}%`;
+  } else {
+    // No total yet: indeterminate "counting…" sweep.
+    bar.classList.add("indeterminate");
+    fill.style.width = "30%";
+  }
+
+  const stalled = document.getElementById("run-stalled");
+  const heartbeatAge =
+    run && status.now_unix ? Number(status.now_unix) - Number(run.updated_at_unix) : 0;
+  if (run && heartbeatAge > STALL_AFTER_S) {
+    stalled.textContent =
+      `No heartbeat for ${formatDuration(heartbeatAge)} — the ingester may be stalled.`;
+    stalled.classList.remove("hidden");
+  } else {
+    stalled.classList.add("hidden");
+  }
+
+  // Phase B never owns a run (owns_active_run is always false): every run was
+  // started from a terminal, and that's where it can be cancelled.
+  document.getElementById("run-origin").textContent = status.owns_active_run
+    ? ""
+    : "Started from the terminal — cancel it with Ctrl-C there.";
+
+  panel.classList.remove("hidden");
+}
+
+function historyItem(run, nowUnix) {
+  const li = document.createElement("li");
+
+  const source = document.createElement("span");
+  source.className = "badge";
+  source.textContent = sourceLabel(run.source);
+
+  const chip = document.createElement("span");
+  stateChip(chip, run.state);
+
+  const counts = document.createElement("span");
+  counts.textContent =
+    `${Number(run.messages_seen || 0).toLocaleString()} seen · ` +
+    `${Number(run.messages_new || 0).toLocaleString()} new`;
+
+  const when = document.createElement("span");
+  when.className = "muted";
+  const stamp = run.finished_at_unix || run.updated_at_unix;
+  when.textContent = stamp ? relativeTime(nowUnix, stamp) : "";
+
+  li.append(source, chip, counts, when);
+
+  if (run.error) {
+    const error = document.createElement("div");
+    error.className = "history-error";
+    // error_kind and error are ingester-written but rendered as inert text.
+    error.textContent = run.error_kind ? `${run.error_kind}: ${run.error}` : run.error;
+    li.appendChild(error);
+  }
+  return li;
+}
+
+function renderHistory(runs, nowUnix) {
+  const section = document.getElementById("history");
+  const list = document.getElementById("history-list");
+  list.replaceChildren();
+  if (!ingest.available || runs.length === 0) {
+    section.classList.add("hidden");
+    return;
+  }
+  for (const run of runs) {
+    list.appendChild(historyItem(run, nowUnix));
+  }
+  section.classList.remove("hidden");
+}
+
+async function fetchRuns() {
+  if (!ingest.available) return;
+  let response;
+  try {
+    response = await fetch("/api/runs?limit=20");
+  } catch (err) {
+    return;
+  }
+  if (!response.ok) return;
+  let body;
+  try {
+    body = await response.json();
+  } catch (err) {
+    return;
+  }
+  renderHistory(Array.isArray(body.runs) ? body.runs : [], body.now_unix || 0);
+}
+
+// The main view (loading/setup/error/stats) is owned by the summary flow, but
+// status changes can flip it: onboarding appears when the DB is missing/empty
+// and idle, and leaves as soon as a run starts or data exists.
+function reconcileMainView() {
+  if (shouldOnboard()) {
+    if (state.view === "stats" || state.view === "setup" || state.view === "loading") {
+      showView("setup");
+    }
+    return;
+  }
+  if (state.view === "setup") {
+    // Onboarding no longer applies (a run started, or data appeared).
+    if (state.data) {
+      update();
+    } else {
+      showView("loading");
+    }
+  }
+}
+
+function renderIngest() {
+  if (!ingest.available) {
+    hideIngestUi();
+    return;
+  }
+  renderPill();
+  renderPanel();
+  document
+    .getElementById("mixed-banner")
+    .classList.toggle("hidden", !(ingest.status && ingest.status.mixed_sources === true));
+  reconcileMainView();
+}
+
+function scheduleStatusPoll(delayMs) {
+  if (!ingest.available) return;
+  if (ingest.timer) clearTimeout(ingest.timer);
+  ingest.timer = setTimeout(pollStatus, delayMs);
+}
+
+function runsKeyOf(status) {
+  if (!status) return "none";
+  const a = status.active_run;
+  const l = status.last_run;
+  return [
+    a ? `${a.run_id}:${a.state}` : "-",
+    l ? `${l.run_id}:${l.state}` : "-",
+  ].join("|");
+}
+
+async function pollStatus() {
+  ingest.timer = null;
+  if (!ingest.available || document.hidden) return;
+
+  let response;
+  try {
+    response = await fetch("/api/status");
+  } catch (err) {
+    // Server unreachable (restart?): hide the ingestion UI, keep probing at
+    // the idle cadence in case it comes back.
+    hideIngestUi();
+    ingest.delay = STATUS_POLL_IDLE_MAX_MS;
+    scheduleStatusPoll(ingest.delay);
+    return;
+  }
+  if (response.status === 404) {
+    // The endpoint doesn't exist at all: static hosting (the Pages demo).
+    // Hide every piece of ingestion UI and never poll again.
+    ingest.available = false;
+    hideIngestUi();
+    return;
+  }
+  if (!response.ok) {
+    // busy 503 and friends: keep whatever is on screen, try again shortly.
+    scheduleStatusPoll(Math.min(STATUS_POLL_IDLE_MAX_MS, ingest.delay * 2));
+    return;
+  }
+  let body;
+  try {
+    body = await response.json();
+  } catch (err) {
+    // 200 with a non-JSON body: an SPA-style static host, not the viewer.
+    ingest.available = false;
+    hideIngestUi();
+    return;
+  }
+
+  ingest.status = body;
+  renderIngest();
+
+  const key = runsKeyOf(body);
+  if (key !== ingest.runsKey) {
+    ingest.runsKey = key;
+    fetchRuns();
+  }
+
+  const active = ingestActive();
+  if (active && Date.now() - ingest.lastSummaryAt > SUMMARY_REFRESH_MS) {
+    refreshSummary();
+  }
+  if (ingest.wasActive && !active) {
+    // Run just finished: pull the final numbers immediately.
+    refreshSummary();
+  }
+  ingest.wasActive = active;
+
+  ingest.delay = active
+    ? STATUS_POLL_ACTIVE_MS
+    : Math.min(STATUS_POLL_IDLE_MAX_MS, Math.max(ingest.delay, STATUS_POLL_ACTIVE_MS) * 2);
+  scheduleStatusPoll(ingest.delay);
+}
+
+// --- summary ----------------------------------------------------------------
+
+// The single site that talks to /api/summary (the demo build rewrites this
+// exact call to read a static summary.json instead).
+async function fetchSummary() {
+  const response = await fetch("/api/summary");
+  const body = await response.json();
+  return { response, body };
+}
+
+/// Silent refresh during active runs: replaces the data but preserves every
+/// piece of client state (search, page, page size, hidden senders — all
+/// client-side already). A busy 503 keeps the previous table and dims the
+/// freshness stamp instead of blanking anything.
+async function refreshSummary() {
+  ingest.lastSummaryAt = Date.now();
+  let response;
+  let body;
+  try {
+    ({ response, body } = await fetchSummary());
+  } catch (err) {
+    return;
+  }
+  const generated = document.getElementById("generated-at");
+  if (!response.ok) {
+    if (body && body.error === "busy") {
+      generated.classList.add("stale");
+    }
+    return;
+  }
+  generated.classList.remove("stale");
+  state.data = body;
+  if (state.view === "stats" || state.view === "loading" || state.view === "setup") {
+    update();
+  }
+}
+
 function update() {
+  if (shouldOnboard()) {
+    showView("setup");
+    return;
+  }
   const data = state.data;
   if (!data) return;
   const all = Array.isArray(data.senders) ? data.senders : [];
@@ -153,15 +554,21 @@ async function load() {
   let response;
   let body;
   try {
-    response = await fetch("/api/summary");
-    body = await response.json();
+    ({ response, body } = await fetchSummary());
   } catch (err) {
     showError("Could not reach the server. Is it still running?");
     return;
   }
   if (!response.ok) {
     if (body && body.error === "missing_db") {
-      showView("setup");
+      // No data yet. If an ingester is already running, stay on the loading
+      // view — the progress panel is visible and the summary refresh will
+      // take over as soon as the first rows land. Otherwise: onboarding.
+      if (ingestActive()) {
+        showView("loading");
+      } else {
+        showView("setup");
+      }
     } else if (body && body.error === "busy") {
       showError(body.message || "Database busy — retry shortly.");
     } else {
@@ -171,6 +578,7 @@ async function load() {
   }
   state.data = body;
   state.page = 1;
+  ingest.lastSummaryAt = Date.now();
   update();
 }
 
@@ -206,4 +614,42 @@ document.getElementById("next-page").addEventListener("click", () => {
   state.page += 1;
   update();
 });
+
+// Copy buttons on the onboarding cards copy the adjacent CLI command.
+for (const btn of document.querySelectorAll(".copy-btn")) {
+  btn.addEventListener("click", () => {
+    const pre = btn.parentElement.querySelector("pre");
+    const text = pre ? pre.textContent : "";
+    const done = () => {
+      btn.textContent = "Copied";
+      setTimeout(() => { btn.textContent = "Copy"; }, 1500);
+    };
+    if (navigator.clipboard && navigator.clipboard.writeText) {
+      navigator.clipboard.writeText(text).then(done, () => selectText(pre));
+    } else {
+      selectText(pre);
+    }
+  });
+}
+
+function selectText(el) {
+  if (!el) return;
+  const range = document.createRange();
+  range.selectNodeContents(el);
+  const selection = window.getSelection();
+  selection.removeAllRanges();
+  selection.addRange(range);
+}
+
+// Pause status polling while the tab is hidden; poll immediately on return.
+document.addEventListener("visibilitychange", () => {
+  if (document.hidden) {
+    if (ingest.timer) clearTimeout(ingest.timer);
+    ingest.timer = null;
+  } else if (ingest.available) {
+    pollStatus();
+  }
+});
+
 load();
+pollStatus();
diff --git a/web/index.html b/web/index.html
index e920a98..ed07cda 100644
--- a/web/index.html
+++ b/web/index.html
@@ -12,23 +12,53 @@
 </head>
 <body>
 <main>
-  <h1>gmail-stats <span>· local viewer</span></h1>
+  <header class="page-head">
+    <h1>gmail-stats <span>· local viewer</span></h1>
+    <span id="ingest-pill" class="pill hidden"><span class="dot" aria-hidden="true"></span><span id="pill-text"></span></span>
+  </header>
 
   <section id="loading">
     <p class="muted">Loading&hellip;</p>
   </section>
 
+  <!-- Onboarding: shown when the database is missing or empty and nothing is
+       running. Phase B is observe-only, so these cards show the exact CLI
+       commands to copy; launch buttons arrive in Phase C. -->
   <section id="setup" class="notice hidden">
-    <h2>No stats database found</h2>
-    <p>The viewer reads <code>./stats.db</code> from the directory it was started in,
-       but the database (or its tables) doesn't exist yet.</p>
-    <p>Create the tables, then run the scanner from the repo root:</p>
-    <pre>$ sqlite3 stats.db
-sqlite&gt; CREATE TABLE seen_mails (mail_id string);
-sqlite&gt; CREATE TABLE senders (sender string, mails_sent int);
-sqlite&gt; .quit
-$ cargo run</pre>
-    <p>When the scan finishes (or while it runs), come back here.</p>
+    <h2>No mail data yet</h2>
+    <p>Get your Gmail data into <code>stats.db</code> one of two ways, running the
+       command from the repo root in a terminal. Once a run starts, this page
+       notices it and fills in live.</p>
+    <div class="cards">
+      <div class="card">
+        <h3>Scan Gmail (OAuth)</h3>
+        <p>Reads message metadata over the Gmail API. Needs a Google Cloud OAuth
+           client (<code>credentials.json</code> in the repo root) plus a one-time
+           browser consent.</p>
+        <div class="cmd">
+          <pre>cargo run</pre>
+          <button class="copy-btn" type="button">Copy</button>
+        </div>
+        <p class="callout">Using Advanced Protection? Google blocks this method
+           outright (<code>Error 400: policy_enforced</code>) — use the Takeout
+           import instead.</p>
+      </div>
+      <div class="card">
+        <h3>Import a Google Takeout mailbox</h3>
+        <ol>
+          <li>At <code>takeout.google.com</code>, export only <strong>Mail</strong>
+              (mbox format).</li>
+          <li>Wait for the export email — Google typically takes
+              <strong>about 2 days</strong> to prepare it for Advanced Protection
+              accounts.</li>
+          <li>Download, unzip, and import the <code>.mbox</code> file:</li>
+        </ol>
+        <div class="cmd">
+          <pre>cargo run -- import "path/to/All mail Including Spam and Trash.mbox"</pre>
+          <button class="copy-btn" type="button">Copy</button>
+        </div>
+      </div>
+    </div>
     <button id="retry-setup" type="button">Retry</button>
   </section>
 
@@ -38,7 +68,31 @@ <h2>Something went wrong</h2>
     <button id="retry-error" type="button">Retry</button>
   </section>
 
+  <!-- Observe-only progress panel for the active ingestion run. -->
+  <section id="ingest-panel" class="notice hidden">
+    <details open>
+      <summary>
+        <span id="run-source" class="badge"></span>
+        <span id="run-state" class="chip"></span>
+      </summary>
+      <div class="run-body">
+        <div class="bar hidden" id="run-bar"><div class="bar-fill" id="run-bar-fill"></div></div>
+        <p class="run-meta">
+          <span id="run-counts"></span>
+          <span id="run-rate" class="muted"></span>
+          <span id="run-eta" class="muted"></span>
+        </p>
+        <p class="warn hidden" id="run-stalled"></p>
+        <p class="muted" id="run-origin"></p>
+      </div>
+    </details>
+  </section>
+
   <section id="stats" class="hidden">
+    <div id="mixed-banner" class="warn-banner hidden">
+      Counts may include duplicates across scan + import: both sources have
+      written to this database, and exact cross-source dedupe hasn't run yet.
+    </div>
     <div class="tiles">
       <div class="tile">
         <div class="label">Total messages</div>
@@ -78,6 +132,13 @@ <h2>Something went wrong</h2>
     </nav>
     <p class="muted" id="generated-at"></p>
   </section>
+
+  <!-- Run history strip, fed by /api/runs; hidden entirely in static/demo
+       hosting where the status endpoints don't exist. -->
+  <section id="history" class="hidden">
+    <h2 class="history-title">Recent runs</h2>
+    <ul id="history-list"></ul>
+  </section>
 </main>
 </body>
 </html>