diff --git a/components/db/lobbying.ts b/components/db/lobbying.ts
index fabae0159..53b428733 100644
--- a/components/db/lobbying.ts
+++ b/components/db/lobbying.ts
@@ -276,19 +276,98 @@ export type BillRow = {
 async function fetchLobbyingBillSummaries(
   court: number
 ): Promise<Record<string, BillSummaryEntry>> {
-  const snap = await getDoc(
-    doc(firestore, LOBBYING_STATS_COLLECTION, `billSummaries_${court}`)
+  const snap = await getDocs(
+    collection(
+      firestore,
+      LOBBYING_STATS_COLLECTION,
+      `billSummaries_${court}`,
+      "bills"
+    )
   )
-  if (!snap.exists()) return {}
-  const raw = snap.data() as { data?: string }
-  if (!raw.data) return {}
-  return JSON.parse(raw.data) as Record<string, BillSummaryEntry>
+  const result: Record<string, BillSummaryEntry> = {}
+  snap.docs.forEach(d => {
+    result[d.id] = d.data() as BillSummaryEntry
+  })
+  return result
 }
 
 export function useLobbyingBillSummaries(court: number) {
   return useAsync(fetchLobbyingBillSummaries, [court])
 }
 
+// ── Client / firm summaries (precomputed server-side; see writer.py and
+// seedLobbyingStats.ts). Replaces client-side derivation over
+// useLobbyingAllRegistrants(), which only fetches the first 2,000 of
+// 25,000+ registrant docs and was silently showing an incomplete list. ────
+
+export type ClientSummaryFirm = {
+  entityName: string
+  entityNameNorm: string
+  compensation: number | null
+}
+
+export type ClientSummaryRow = {
+  clientName: string
+  clientNameNorm: string
+  totalCompensation: number | null
+  registrantCount: number
+  firms: ClientSummaryFirm[]
+}
+
+export type FirmSummaryRow = {
+  entityName: string
+  entityNameNorm: string
+  regType: string
+  years: number[]
+  clientCount: number
+}
+
+async function fetchClientSummaries(): Promise<ClientSummaryRow[]> {
+  const snap = await getDocs(
+    collection(
+      firestore,
+      LOBBYING_STATS_COLLECTION,
+      "clientSummaries",
+      "clients"
+    )
+  )
+  return snap.docs.map(d => d.data() as ClientSummaryRow)
+}
+
+async function fetchClientSummary(
+  clientNameNorm: string
+): Promise<ClientSummaryRow | undefined> {
+  const snap = await getDoc(
+    doc(
+      firestore,
+      LOBBYING_STATS_COLLECTION,
+      "clientSummaries",
+      "clients",
+      encodeURIComponent(clientNameNorm)
+    )
+  )
+  return snap.exists() ? (snap.data() as ClientSummaryRow) : undefined
+}
+
+async function fetchFirmSummaries(): Promise<FirmSummaryRow[]> {
+  const snap = await getDocs(
+    collection(firestore, LOBBYING_STATS_COLLECTION, "firmSummaries", "firms")
+  )
+  return snap.docs.map(d => d.data() as FirmSummaryRow)
+}
+
+export function useLobbyingClientSummaries() {
+  return useAsync(fetchClientSummaries, [])
+}
+
+export function useLobbyingClientSummary(clientNameNorm: string) {
+  return useAsync(fetchClientSummary, [clientNameNorm])
+}
+
+export function useLobbyingFirmSummaries() {
+  return useAsync(fetchFirmSummaries, [])
+}
+
 export function useLobbyingBillRows(courts: number[]) {
   const courtsKey = courts.join(",")
   return useAsync(
diff --git a/components/lobbying/LobbyingBillCard.tsx b/components/lobbying/LobbyingBillCard.tsx
index a22c713e1..96b90f4b6 100644
--- a/components/lobbying/LobbyingBillCard.tsx
+++ b/components/lobbying/LobbyingBillCard.tsx
@@ -2,9 +2,13 @@ import React from "react"
 import { useTranslation } from "next-i18next"
 import { useLobbyingFilingsForBill } from "components/db/lobbying"
 import { LobbyingFilingsTable } from "./LobbyingFilingsTable"
+import { LobbyingPaginationBar } from "./LobbyingPaginationBar"
+import { usePagination } from "./usePagination"
 import { MAPLE_COLORS } from "./chartTheme"
 import { normalizePosition } from "./LobbyingPositionChip"
 
+const PAGE_SIZE = 10
+
 interface LobbyingBillCardProps {
   court: number
   billId: string
@@ -22,6 +26,10 @@ export const LobbyingBillCard: React.FC<LobbyingBillCardProps> = ({
     status,
     error
   } = useLobbyingFilingsForBill(court, billId)
+  const { page, setPage, pageItems, totalPages, totalItems } = usePagination(
+    filings ?? [],
+    PAGE_SIZE
+  )
 
   if (status === "loading" || status === "not-requested") {
     return (
@@ -60,10 +68,10 @@ export const LobbyingBillCard: React.FC<LobbyingBillCardProps> = ({
   return (
     <div style={cardStyle} className={className}>
       <div style={headerStyle}>
-        <span style={titleStyle}>{t("lobbying:titles.overview")}</span>
-        <span style={countStyle}>
-          {t("lobbying:billCard.filingCount_other", { count: total })}
-        </span>
+        <span style={titleStyle}>{t("lobbying:billCard.title")}</span>
+        <a href={explorerHref} style={viewAllLinkStyle}>
+          {t("lobbying:billCard.viewAll")}
+        </a>
       </div>
 
       <PositionBar counts={counts} total={total} />
@@ -87,17 +95,21 @@ export const LobbyingBillCard: React.FC<LobbyingBillCardProps> = ({
       </div>
 
       <LobbyingFilingsTable
-        filings={filings}
+        filings={pageItems}
         showBill={false}
         showClient
         showFirm
         showAmount={false}
-        maxRows={5}
+        bordered
+      />
+      <LobbyingPaginationBar
+        page={page}
+        totalPages={totalPages}
+        totalItems={totalItems}
+        pageSize={PAGE_SIZE}
+        onPage={setPage}
+        itemLabel={t("lobbying:billCard.filingsLabel")}
       />
-
-      <a href={explorerHref} style={viewAllLinkStyle}>
-        {t("lobbying:billCard.viewAll")}
-      </a>
     </div>
   )
 }
@@ -211,11 +223,6 @@ const titleStyle: React.CSSProperties = {
   letterSpacing: "0.06em"
 }
 
-const countStyle: React.CSSProperties = {
-  fontSize: 12,
-  color: MAPLE_COLORS.textMuted
-}
-
 const barContainerStyle: React.CSSProperties = {
   display: "flex",
   height: 8,
@@ -245,10 +252,9 @@ const legendItemStyle: React.CSSProperties = {
 }
 
 const viewAllLinkStyle: React.CSSProperties = {
-  display: "block",
-  fontSize: 13,
+  fontSize: 12,
   fontWeight: 600,
   color: MAPLE_COLORS.primary,
   textDecoration: "none",
-  marginTop: "0.5rem"
+  whiteSpace: "nowrap"
 }
diff --git a/components/lobbying/LobbyingFilingsTable.tsx b/components/lobbying/LobbyingFilingsTable.tsx
index 103fdb329..1366e484a 100644
--- a/components/lobbying/LobbyingFilingsTable.tsx
+++ b/components/lobbying/LobbyingFilingsTable.tsx
@@ -14,6 +14,7 @@ interface LobbyingFilingsTableProps {
   showActivity?: boolean
   maxRows?: number
   onViewAll?: () => void
+  bordered?: boolean
 }
 
 export const LobbyingFilingsTable: React.FC<LobbyingFilingsTableProps> = ({
@@ -24,7 +25,8 @@ export const LobbyingFilingsTable: React.FC<LobbyingFilingsTableProps> = ({
   showAmount = true,
   showActivity = false,
   maxRows,
-  onViewAll
+  onViewAll,
+  bordered = false
 }) => {
   const { t } = useTranslation("lobbying")
   const rows = maxRows ? filings.slice(0, maxRows) : filings
@@ -39,7 +41,17 @@ export const LobbyingFilingsTable: React.FC<LobbyingFilingsTableProps> = ({
   }
 
   return (
-    <div>
+    <div
+      style={
+        bordered
+          ? {
+              border: "1px solid var(--bs-border-color)",
+              borderRadius: 6,
+              overflow: "hidden"
+            }
+          : undefined
+      }
+    >
       <Table hover responsive size="sm" style={tableStyle}>
         <thead>
           <tr style={theadRowStyle}>
@@ -71,8 +83,30 @@ export const LobbyingFilingsTable: React.FC<LobbyingFilingsTableProps> = ({
                   )}
                 </td>
               )}
-              {showClient && <td style={cellStyle}>{f.clientName}</td>}
-              {showFirm && <td style={cellStyle}>{f.entityName}</td>}
+              {showClient && (
+                <td style={cellStyle}>
+                  <a
+                    href={`/lobbying/clients/${encodeURIComponent(
+                      f.clientNameNorm
+                    )}`}
+                    style={{ color: MAPLE_COLORS.primary }}
+                  >
+                    {f.clientName}
+                  </a>
+                </td>
+              )}
+              {showFirm && (
+                <td style={cellStyle}>
+                  <a
+                    href={`/lobbying/firms/${encodeURIComponent(
+                      f.entityNameNorm
+                    )}`}
+                    style={{ color: MAPLE_COLORS.primary }}
+                  >
+                    {f.entityName}
+                  </a>
+                </td>
+              )}
               {showActivity && (
                 <td style={{ ...cellStyle, color: MAPLE_COLORS.textMuted }}>
                   {f.activityTitle || "—"}
@@ -116,8 +150,10 @@ export const LobbyingFilingsTable: React.FC<LobbyingFilingsTableProps> = ({
 
 const tableStyle: React.CSSProperties = {
   fontSize: 13,
-  color: MAPLE_COLORS.textBody
-}
+  color: MAPLE_COLORS.textBody,
+  "--bs-table-bg": "var(--bs-gray-100)",
+  "--bs-table-hover-bg": "rgba(15, 23, 42, 0.06)"
+} as React.CSSProperties
 
 const theadRowStyle: React.CSSProperties = {
   fontSize: 11,
diff --git a/components/lobbying/LobbyingPaginationBar.tsx b/components/lobbying/LobbyingPaginationBar.tsx
index 7c6254a45..18f75512d 100644
--- a/components/lobbying/LobbyingPaginationBar.tsx
+++ b/components/lobbying/LobbyingPaginationBar.tsx
@@ -1,4 +1,4 @@
-import React from "react"
+import React, { useEffect, useState } from "react"
 import { useTranslation } from "next-i18next"
 import { MAPLE_COLORS } from "./chartTheme"
 
@@ -8,6 +8,7 @@ interface Props {
   totalItems: number
   pageSize: number
   onPage: (p: number) => void
+  itemLabel?: string
 }
 
 export function LobbyingPaginationBar({
@@ -15,19 +16,41 @@ export function LobbyingPaginationBar({
   totalPages,
   totalItems,
   pageSize,
-  onPage
+  onPage,
+  itemLabel
 }: Props) {
   const { t } = useTranslation("lobbying")
+  const [pageInput, setPageInput] = useState(String(page))
+
+  // Keep the typed value in sync when the page changes externally (Prev/
+  // Next, or a filter change resetting to page 1) — but not while the user
+  // is actively typing a replacement value.
+  useEffect(() => {
+    setPageInput(String(page))
+  }, [page])
+
   if (totalPages <= 1) return null
   const start = (page - 1) * pageSize + 1
   const end = Math.min(page * pageSize, totalItems)
 
+  function commitPageInput() {
+    const parsed = parseInt(pageInput, 10)
+    if (Number.isNaN(parsed)) {
+      setPageInput(String(page))
+      return
+    }
+    const clamped = Math.min(Math.max(parsed, 1), totalPages)
+    setPageInput(String(clamped))
+    if (clamped !== page) onPage(clamped)
+  }
+
   return (
     <div style={wrapStyle}>
       <span style={{ color: MAPLE_COLORS.textMuted, fontSize: 13 }}>
         {start}–{end} of {totalItems}
+        {itemLabel ? ` ${itemLabel}` : ""}
       </span>
-      <div style={{ display: "flex", gap: "0.25rem" }}>
+      <div style={{ display: "flex", gap: "0.25rem", alignItems: "center" }}>
         <button
           onClick={() => onPage(page - 1)}
           disabled={page === 1}
@@ -35,14 +58,22 @@ export function LobbyingPaginationBar({
         >
           {t("pagination.prev")}
         </button>
-        <span
-          style={{
-            ...btnStyle(false),
-            cursor: "default",
-            color: MAPLE_COLORS.textMuted
-          }}
-        >
-          {page} / {totalPages}
+        <span style={pageJumpStyle}>
+          <input
+            type="text"
+            inputMode="numeric"
+            value={pageInput}
+            onChange={e => setPageInput(e.target.value.replace(/[^0-9]/g, ""))}
+            onBlur={commitPageInput}
+            onKeyDown={e => {
+              if (e.key === "Enter") {
+                e.currentTarget.blur()
+              }
+            }}
+            aria-label={t("pagination.pageNumber")}
+            style={pageInputStyle}
+          />
+          <span style={{ color: MAPLE_COLORS.textMuted }}> / {totalPages}</span>
         </span>
         <button
           onClick={() => onPage(page + 1)}
@@ -75,3 +106,24 @@ const btnStyle = (disabled: boolean): React.CSSProperties => ({
   cursor: disabled ? "default" : "pointer",
   opacity: disabled ? 0.5 : 1
 })
+
+// Deliberately not button-shaped (no border/background) — this is an
+// editable field, not a static label, so it shouldn't visually compete with
+// the actual Prev/Next buttons on either side.
+const pageJumpStyle: React.CSSProperties = {
+  display: "flex",
+  alignItems: "center",
+  fontSize: 13,
+  padding: "0.3rem 0.35rem"
+}
+
+const pageInputStyle: React.CSSProperties = {
+  width: "2.2rem",
+  textAlign: "center",
+  border: "none",
+  borderBottom: `1px solid ${MAPLE_COLORS.borderDefault}`,
+  background: "transparent",
+  color: MAPLE_COLORS.textBody,
+  fontSize: 13,
+  padding: "0.1rem 0.2rem"
+}
diff --git a/firestore.rules b/firestore.rules
index df5659ac5..9e0bb6b8f 100644
--- a/firestore.rules
+++ b/firestore.rules
@@ -114,6 +114,27 @@ service cloud.firestore {
     match /lobbyingMeta/{id} {
       allow read: if true;
       allow write: if false;
+
+      // billSummaries_{court} docs store per-bill counts as small docs in
+      // this subcollection (not as a single JSON blob) to avoid Firestore's
+      // 1MB per-document/field size limit as a session's filings accumulate.
+      match /bills/{billId} {
+        allow read: if true;
+        allow write: if false;
+      }
+
+      // clientSummaries/firmSummaries store one small doc per client/firm
+      // (same reasoning as billSummaries above) so the clients and firms
+      // browse pages can read a precomputed rollup instead of scanning the
+      // full, capped-at-2000 lobbyingRegistrants query client-side.
+      match /clients/{clientId} {
+        allow read: if true;
+        allow write: if false;
+      }
+      match /firms/{firmId} {
+        allow read: if true;
+        allow write: if false;
+      }
     }
     match /transcriptions/{tid} {
       // public, read-only
diff --git a/lobbying-scraper/scrape.py b/lobbying-scraper/scrape.py
index 55fa793c6..1f4ed23a4 100644
--- a/lobbying-scraper/scrape.py
+++ b/lobbying-scraper/scrape.py
@@ -38,7 +38,9 @@
 from writer import (
     BACKFILL_DOC,
     BACKFILL_URLS_COLLECTION,
+    PROCESSED_URLS_COLLECTION,
     SCRAPER_DOC,
+    SUMMARY_CACHE_COLLECTION,
     compute_stats,
     write_filings,
     write_registrant,
@@ -46,36 +48,70 @@
 
 
 # ── Cursor helpers ────────────────────────────────────────────────────────────
+#
+# Weekly-mode cursor state lives in subcollections under SCRAPER_DOC, one small
+# doc per URL, mirroring the backfill cursor below. An earlier version stored
+# the entire processed-URL history and summary cache as two fields on a single
+# document; that doc grew past Firestore's 1MB limit once run against the full
+# corpus, silently failing (and thus skipping) every registrant processed
+# after the limit was hit. Per-URL docs have no such ceiling.
 
 
-def _load_live_cursor(db: firestore.Client) -> tuple[set[str], dict[str, list[str]]]:
-    """Return (processedDiscUrls, summaryDiscCache) from the live scraper doc."""
-    doc = db.document(SCRAPER_DOC).get()
-    data = doc.to_dict() or {}
+def _url_hash(url: str) -> str:
+    return hashlib.sha256(url.encode()).hexdigest()[:40]
+
+
+def _is_processed(db: firestore.Client, disc_url: str) -> bool:
+    h = _url_hash(disc_url)
     return (
-        set(data.get("processedDiscUrls", [])),
-        data.get("summaryDiscCache", {}),
+        db.document(SCRAPER_DOC)
+        .collection(PROCESSED_URLS_COLLECTION)
+        .document(h)
+        .get()
+        .exists
     )
 
 
-def _save_live_cursor(
-    db: firestore.Client,
-    processed: set[str],
-    cache: dict[str, list[str]],
+def _mark_processed(db: firestore.Client, disc_url: str) -> None:
+    h = _url_hash(disc_url)
+    db.document(SCRAPER_DOC).collection(PROCESSED_URLS_COLLECTION).document(h).set(
+        {"url": disc_url, "processedAt": datetime.now(tz=timezone.utc).isoformat()}
+    )
+
+
+def _get_cached_disc_urls(db: firestore.Client, summary_url: str) -> list[str] | None:
+    """Cached disclosure URLs for a registrant's summary page, or None if unseen.
+
+    Only consulted for prior years — the current year is always refetched live
+    since its disclosures can still change.
+    """
+    h = _url_hash(summary_url)
+    doc = db.document(SCRAPER_DOC).collection(SUMMARY_CACHE_COLLECTION).document(h).get()
+    if not doc.exists:
+        return None
+    return doc.to_dict().get("discUrls", [])
+
+
+def _cache_disc_urls(
+    db: firestore.Client, summary_url: str, disc_urls: list[str]
 ) -> None:
-    db.document(SCRAPER_DOC).set(
-        {"processedDiscUrls": list(processed), "summaryDiscCache": cache},
-        merge=True,
+    h = _url_hash(summary_url)
+    db.document(SCRAPER_DOC).collection(SUMMARY_CACHE_COLLECTION).document(h).set(
+        {
+            "summaryUrl": summary_url,
+            "discUrls": disc_urls,
+            "cachedAt": datetime.now(tz=timezone.utc).isoformat(),
+        }
     )
 
 
 def _is_backfill_processed(db: firestore.Client, disc_url: str) -> bool:
-    h = hashlib.sha256(disc_url.encode()).hexdigest()[:40]
+    h = _url_hash(disc_url)
     return db.document(BACKFILL_DOC).collection(BACKFILL_URLS_COLLECTION).document(h).get().exists
 
 
 def _mark_backfill_processed(db: firestore.Client, disc_url: str) -> None:
-    h = hashlib.sha256(disc_url.encode()).hexdigest()[:40]
+    h = _url_hash(disc_url)
     db.document(BACKFILL_DOC).collection(BACKFILL_URLS_COLLECTION).document(h).set(
         {"url": disc_url, "processedAt": datetime.now(tz=timezone.utc).isoformat()}
     )
@@ -118,7 +154,7 @@ def run_weekly(
 ) -> int:
     """Incremental weekly check. Returns number of new disclosures processed."""
     current_year = datetime.now(tz=timezone.utc).year
-    processed, cache = _load_live_cursor(db) if db is not None else (set(), {})
+    use_cursor = db is not None and not dry_run
 
     session = make_session()
     new_count = 0
@@ -137,33 +173,33 @@ def run_weekly(
         print(f"  {len(summary_urls)} registrants on portal")
 
         for summary_url in summary_urls:
-            # Use cached disc URLs for prior years; always re-check current year
-            disc_urls = cache.get(summary_url)
-            if disc_urls is None or year == current_year:
+            # Prior years: trust the cache if we have one. Current year:
+            # always refetch live, since its disclosures can still change.
+            disc_urls = None
+            if year != current_year and use_cursor:
+                disc_urls = _get_cached_disc_urls(db, summary_url)
+
+            if disc_urls is None:
                 try:
                     meta = fetch_disclosure_meta(session, summary_url)
                     disc_urls = meta.disclosure_urls
-                    cache[summary_url] = disc_urls
-                    if not dry_run:
-                        _save_live_cursor(db, processed, cache)
+                    if use_cursor:
+                        _cache_disc_urls(db, summary_url, disc_urls)
                 except Exception as e:
                     print(f"  failed to fetch summary {summary_url}: {e}", file=sys.stderr)
                     continue
 
-            new_disc_urls = [u for u in disc_urls if u not in processed]
-            if not new_disc_urls:
-                continue
-
-            for disc_url in new_disc_urls:
+            for disc_url in disc_urls:
+                if use_cursor and _is_processed(db, disc_url):
+                    continue
                 try:
                     comp_n, filing_n = process_disclosure(
                         db, session, summary_url, disc_url, year, dry_run=dry_run
                     )
-                    processed.add(disc_url)
                     new_count += 1
                     print(f"  processed: {comp_n} clients, {filing_n} filings")
-                    if not dry_run:
-                        _save_live_cursor(db, processed, cache)
+                    if use_cursor:
+                        _mark_processed(db, disc_url)
                 except Exception as e:
                     print(f"  failed to process {disc_url}: {e}", file=sys.stderr)
 
@@ -171,17 +207,21 @@ def run_weekly(
 
 
 # ── Historical backfill ───────────────────────────────────────────────────────
-
-
-def _completed_years(db: "firestore.Client") -> set[int]:
-    data = db.document(BACKFILL_DOC).get().to_dict() or {}
-    return set(data.get("completedYears", []))
-
-
-def _mark_year_complete(db: "firestore.Client", year: int) -> None:
-    db.document(BACKFILL_DOC).set(
-        {"completedYears": firestore.ArrayUnion([year])}, merge=True
-    )
+#
+# Correctness here relies entirely on the per-URL cursor (_is_backfill_processed
+# / _mark_backfill_processed below) — every disclosure URL is checked and
+# marked individually, so re-running a backfill is always safe and complete.
+#
+# An earlier version also tracked a per-year "completedYears" flag as a
+# fast-path to skip re-listing a year's registrants at all. That flag was
+# permanent once set, which is wrong for the current (still-accruing) year:
+# a backfill run partway through the year would mark it complete after
+# finding whatever existed at that moment, and every later run would then
+# skip it forever — silently missing every disclosure filed afterward. There
+# is no reliable way to tell "genuinely finished" apart from "happened to be
+# a quiet moment" for a year that's still in progress, so the flag is gone;
+# each run always re-lists every requested year's registrants (one cheap
+# HTTP request per year) and leans on the per-URL cursor for correctness.
 
 
 def run_backfill(
@@ -190,18 +230,15 @@ def run_backfill(
     limit: int | None = None,
     dry_run: bool = False,
 ) -> int:
-    """Full historical backfill using the subcollection cursor. Resumable."""
+    """Full historical backfill using the per-URL subcollection cursor.
+
+    Always resumable and safe to re-run: every disclosure URL is checked
+    individually against the cursor, so no year is ever skipped wholesale.
+    """
     session = make_session()
     total_new = 0
 
-    done = _completed_years(db) if db is not None and not dry_run else set()
-    if done:
-        print(f"Skipping already-completed years: {sorted(done)}")
-
     for year in years:
-        if year in done:
-            continue
-
         print(f"\n── {year} ──")
         try:
             summary_urls = fetch_summary_links(session, year)
@@ -240,8 +277,6 @@ def run_backfill(
                 print(f"  [{i+1}/{len(summary_urls)}] {year_new} new disclosures so far")
 
         print(f"  {year} complete: {year_new} new disclosures")
-        if db is not None and not dry_run and not limit:
-            _mark_year_complete(db, year)
 
     return total_new
 
diff --git a/lobbying-scraper/tests/test_scrape.py b/lobbying-scraper/tests/test_scrape.py
new file mode 100644
index 000000000..4e8d1557a
--- /dev/null
+++ b/lobbying-scraper/tests/test_scrape.py
@@ -0,0 +1,322 @@
+"""Unit tests for the weekly/backfill cursor logic in scrape.py.
+
+Uses a tiny in-memory fake standing in for firestore.Client — real enough to
+exercise document/subcollection reads and writes statefully across calls
+(unlike a plain MagicMock, which can't easily simulate "write now, read back
+later"), without needing a live database or emulator.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from portal import DisclosureDetail, DisclosureMeta
+import scrape
+
+
+# ── Fake Firestore ────────────────────────────────────────────────────────────
+
+
+class _FakeSnapshot:
+    def __init__(self, data):
+        self._data = data
+
+    @property
+    def exists(self):
+        return self._data is not None
+
+    def to_dict(self):
+        return self._data
+
+
+class _FakeDocRef:
+    def __init__(self, store, path):
+        self._store = store
+        self._path = path
+
+    def get(self):
+        return _FakeSnapshot(self._store.get(self._path))
+
+    def set(self, data, merge=False):
+        if merge and self._path in self._store:
+            self._store[self._path] = {**self._store[self._path], **data}
+        else:
+            self._store[self._path] = dict(data)
+
+    def collection(self, name):
+        return _FakeCollectionRef(self._store, f"{self._path}/{name}")
+
+
+class _FakeCollectionRef:
+    def __init__(self, store, path):
+        self._store = store
+        self._path = path
+
+    def document(self, doc_id):
+        return _FakeDocRef(self._store, f"{self._path}/{doc_id}")
+
+
+class FakeFirestore:
+    """Minimal stand-in for firestore.Client: document()/collection() only."""
+
+    def __init__(self):
+        self.store: dict[str, dict] = {}
+
+    def document(self, path):
+        return _FakeDocRef(self.store, path)
+
+    def collection(self, name):
+        return _FakeCollectionRef(self.store, name)
+
+
+# ── Helpers ───────────────────────────────────────────────────────────────────
+
+
+def _meta(summary_url: str, disc_urls: list[str]) -> DisclosureMeta:
+    return DisclosureMeta(
+        entity_name=f"Entity for {summary_url}",
+        year=2024,
+        reg_type="Employer",
+        disclosure_urls=disc_urls,
+    )
+
+
+# ── run_backfill: no year is ever skipped wholesale ──────────────────────────
+
+
+def test_backfill_relists_every_year_on_every_run():
+    """No completedYears-style gate: fetch_summary_links must be called for
+    every requested year, on every run, regardless of what a prior run did."""
+    db = FakeFirestore()
+    summary_links = {2020: ["https://x/summary/a"], 2021: [], 2022: []}
+    meta_by_url = {"https://x/summary/a": _meta("https://x/summary/a", [])}
+
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, year: summary_links.get(year, []),
+    ) as fetch_links, patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: meta_by_url[url],
+    ):
+        scrape.run_backfill(db, years=[2020, 2021, 2022])
+        scrape.run_backfill(db, years=[2020, 2021, 2022])
+
+    # 3 years x 2 runs = 6 calls, none skipped by a "completed" flag.
+    assert fetch_links.call_count == 6
+    called_years = sorted(c.args[1] for c in fetch_links.call_args_list)
+    assert called_years == [2020, 2020, 2021, 2021, 2022, 2022]
+
+
+def test_backfill_does_not_write_completed_years_anywhere():
+    """The old completedYears field must never be written by the new code."""
+    db = FakeFirestore()
+    summary_links = {2024: ["https://x/summary/a"]}
+    meta_by_url = {
+        "https://x/summary/a": _meta("https://x/summary/a", ["https://x/disc/1"])
+    }
+
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, year: summary_links.get(year, []),
+    ), patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: meta_by_url[url],
+    ), patch(
+        "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+    ), patch(
+        "scrape.write_registrant", return_value=None
+    ), patch(
+        "scrape.write_filings", return_value=0
+    ):
+        scrape.run_backfill(db, years=[2024])
+
+    for path, data in db.store.items():
+        assert "completedYears" not in data, f"stale field written at {path}"
+
+
+def test_backfill_regression_partial_year_then_real_data_appears():
+    """The exact bug scenario: a backfill run mid-year finds nothing for the
+    current year (quiet moment), and a later run for the same year finds real
+    disclosures. The second run must NOT skip the year — it must process the
+    newly-appeared data. (Previously: the first pass would mark the year
+    'complete' with 0 disclosures, and the second run would skip it forever.)
+    """
+    db = FakeFirestore()
+    summary_url = "https://x/summary/late-filer"
+    disc_url = "https://x/disc/late-filer-1"
+
+    # First run: this year has no registrants yet.
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links", side_effect=lambda session, year: []
+    ) as fetch_links_1:
+        n1 = scrape.run_backfill(db, years=[2026])
+    assert n1 == 0
+    assert fetch_links_1.call_count == 1
+
+    # Second run: a registrant has since filed for the same year.
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, year: [summary_url],
+    ) as fetch_links_2, patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: _meta(url, [disc_url]),
+    ), patch(
+        "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+    ), patch(
+        "scrape.write_registrant", return_value=None
+    ), patch(
+        "scrape.write_filings", return_value=0
+    ):
+        n2 = scrape.run_backfill(db, years=[2026])
+
+    # The year was re-listed (not skipped) and the new disclosure was processed.
+    assert fetch_links_2.call_count == 1
+    assert n2 == 1
+
+
+def test_backfill_skips_already_processed_disclosures_but_not_the_year():
+    """Per-URL dedup still works: a disclosure already marked processed is
+    not reprocessed, even though the year itself is always re-listed."""
+    db = FakeFirestore()
+    summary_url = "https://x/summary/a"
+    disc_url = "https://x/disc/1"
+    meta_by_url = {summary_url: _meta(summary_url, [disc_url])}
+
+    def run():
+        with patch("scrape.make_session", return_value=None), patch(
+            "scrape.fetch_summary_links",
+            side_effect=lambda session, year: [summary_url],
+        ), patch(
+            "scrape.fetch_disclosure_meta",
+            side_effect=lambda session, url: meta_by_url[url],
+        ), patch(
+            "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+        ), patch(
+            "scrape.write_registrant", return_value=None
+        ), patch(
+            "scrape.write_filings", return_value=0
+        ):
+            return scrape.run_backfill(db, years=[2024])
+
+    n1 = run()
+    n2 = run()
+
+    assert n1 == 1  # first run: one new disclosure
+    assert n2 == 0  # second run: already processed, correctly skipped
+
+
+def test_backfill_dry_run_never_touches_firestore():
+    db = FakeFirestore()
+    summary_url = "https://x/summary/a"
+    disc_url = "https://x/disc/1"
+    meta_by_url = {summary_url: _meta(summary_url, [disc_url])}
+
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, year: [summary_url],
+    ), patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: meta_by_url[url],
+    ), patch(
+        "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+    ):
+        n = scrape.run_backfill(None, years=[2024], dry_run=True)
+
+    assert n == 1
+    assert db.store == {}
+
+
+# ── run_weekly: subcollection cursor (Bug 1 fix) sanity checks ───────────────
+
+
+def test_weekly_skips_already_processed_disclosure():
+    db = FakeFirestore()
+    summary_url = "https://x/summary/a"
+    disc_url = "https://x/disc/1"
+    meta_by_url = {summary_url: _meta(summary_url, [disc_url])}
+
+    def run(year):
+        with patch("scrape.make_session", return_value=None), patch(
+            "scrape.fetch_summary_links",
+            side_effect=lambda session, y: [summary_url],
+        ), patch(
+            "scrape.fetch_disclosure_meta",
+            side_effect=lambda session, url: meta_by_url[url],
+        ), patch(
+            "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+        ), patch(
+            "scrape.write_registrant", return_value=None
+        ), patch(
+            "scrape.write_filings", return_value=0
+        ):
+            return scrape.run_weekly(db, years=[year])
+
+    # Use a prior (non-current) year so caching also gets exercised below.
+    n1 = run(2020)
+    n2 = run(2020)
+
+    assert n1 == 1
+    assert n2 == 0
+
+
+def test_weekly_caches_prior_year_but_not_current_year():
+    db = FakeFirestore()
+    current_year = scrape.datetime.now(tz=scrape.timezone.utc).year
+    prior_year = current_year - 1
+    summary_url = "https://x/summary/a"
+
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, y: [summary_url],
+    ), patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: _meta(url, []),
+    ) as fetch_meta:
+        # Prior year twice: second call should hit the cache, not refetch.
+        scrape.run_weekly(db, years=[prior_year])
+        scrape.run_weekly(db, years=[prior_year])
+        assert fetch_meta.call_count == 1
+
+        # Current year twice: must always refetch live.
+        fetch_meta.reset_mock()
+        scrape.run_weekly(db, years=[current_year])
+        scrape.run_weekly(db, years=[current_year])
+        assert fetch_meta.call_count == 2
+
+
+def test_weekly_cursor_doc_never_exceeds_a_few_small_fields():
+    """Regression guard for the original 1MB-doc bug: the parent scraper doc
+    itself must stay tiny — all real state lives in subcollection docs."""
+    db = FakeFirestore()
+    summary_url = "https://x/summary/a"
+    disc_url = "https://x/disc/1"
+    meta_by_url = {summary_url: _meta(summary_url, [disc_url])}
+
+    with patch("scrape.make_session", return_value=None), patch(
+        "scrape.fetch_summary_links",
+        side_effect=lambda session, y: [summary_url],
+    ), patch(
+        "scrape.fetch_disclosure_meta",
+        side_effect=lambda session, url: meta_by_url[url],
+    ), patch(
+        "scrape.fetch_disclosure_detail", return_value=DisclosureDetail()
+    ), patch(
+        "scrape.write_registrant", return_value=None
+    ), patch(
+        "scrape.write_filings", return_value=0
+    ):
+        scrape.run_weekly(db, years=[2020])
+
+    parent = db.store.get(scrape.SCRAPER_DOC)
+    assert parent is None or "processedDiscUrls" not in parent
+    assert parent is None or "summaryDiscCache" not in parent
+
+    # The actual state must be in per-URL subcollection docs.
+    subcollection_paths = [
+        p for p in db.store if p.startswith(scrape.SCRAPER_DOC + "/")
+    ]
+    assert len(subcollection_paths) >= 2  # one processedUrls doc, one summaryCache doc
diff --git a/lobbying-scraper/writer.py b/lobbying-scraper/writer.py
index cfcfbca38..d9a9fe57b 100644
--- a/lobbying-scraper/writer.py
+++ b/lobbying-scraper/writer.py
@@ -6,9 +6,11 @@
 
 from __future__ import annotations
 
-import json
+import time
+import urllib.parse
 from datetime import datetime, timezone
 
+from google.api_core.exceptions import GoogleAPICallError
 from google.cloud import firestore
 from normalize import normalize_entity_name
 from portal import (
@@ -24,11 +26,80 @@
 REGISTRANTS_COLLECTION = "lobbyingRegistrants"
 FILINGS_COLLECTION = "lobbyingFilings"
 SCRAPER_DOC = "scrapers/lobbying"
+PROCESSED_URLS_COLLECTION = "processedUrls"
+SUMMARY_CACHE_COLLECTION = "summaryCache"
 BACKFILL_DOC = "scrapers/lobbyingBackfill"
 BACKFILL_URLS_COLLECTION = "processedUrls"
 STATS_COLLECTION = "lobbyingMeta"
 STATS_DOC_ID = "stats"
 
+# Sentinel clientName used for pre-2013 legacy filings where compensation is
+# reported as a single total rather than broken down per client. Must match
+# LEGACY_TOTAL_CLIENT in functions/src/lobbying/types.ts.
+LEGACY_TOTAL_CLIENT = "_total_salary_"
+
+
+def _is_legacy_total_client(name: str | None, name_norm: str | None) -> bool:
+    if not name_norm or name_norm == LEGACY_TOTAL_CLIENT:
+        return True
+    if name == LEGACY_TOTAL_CLIENT:
+        return True
+    lc = (name or "").lower()
+    return "total salaries" in lc or "total salary" in lc
+
+
+def _doc_id_for_norm(name_norm: str) -> str:
+    """Firestore doc ID for a normalized name — matches JS encodeURIComponent()
+    exactly (same unreserved character set: alnum, - _ . ! ~ * ' ( )), so the
+    frontend can look up a single summary doc directly via
+    encodeURIComponent(clientNameNorm) without scanning the whole
+    subcollection.
+    """
+    return urllib.parse.quote(name_norm, safe="!*'()")
+
+# compute_stats() streams the full filings/registrants collections, which at
+# MAPLE's current scale (300K+ docs) can exceed Firestore's server-side query
+# timeout. Batching with an explicit cursor keeps each individual RPC small
+# and fast; retry=None disables the client library's built-in stream-retry
+# (which has a version-skew bug that crashes instead of retrying), and the
+# manual retry loop below just re-issues a fresh, small query on failure
+# instead of trying to resume a broken stream.
+_BATCH_SIZE = 50000
+_MAX_RETRIES = 3
+
+
+def _iter_collection(db: firestore.Client, collection_name: str):
+    """Yield every document in a collection via small, cursor-paginated reads."""
+    coll_ref = db.collection(collection_name)
+    last_doc = None
+
+    while True:
+        query = coll_ref.order_by("__name__").limit(_BATCH_SIZE)
+        if last_doc is not None:
+            query = query.start_after(last_doc)
+
+        for attempt in range(_MAX_RETRIES):
+            try:
+                batch = list(query.stream(retry=None))
+                break
+            except GoogleAPICallError as e:
+                if attempt == _MAX_RETRIES - 1:
+                    raise
+                print(
+                    f"  batch read failed ({e}); retrying "
+                    f"({attempt + 1}/{_MAX_RETRIES})…"
+                )
+                time.sleep(2**attempt)
+
+        if not batch:
+            return
+
+        yield from batch
+        last_doc = batch[-1]
+
+        if len(batch) < _BATCH_SIZE:
+            return
+
 
 def _now() -> datetime:
     return datetime.now(tz=timezone.utc)
@@ -60,7 +131,7 @@ def compute_stats(db: firestore.Client) -> None:
     bill_entity_sets: dict[int, dict[str, set]] = {}
     total_filings = 0
 
-    for doc in db.collection(FILINGS_COLLECTION).stream():
+    for doc in _iter_collection(db, FILINGS_COLLECTION):
         d = doc.to_dict()
         year = str(d.get("year", ""))
         gc = d.get("generalCourt")
@@ -114,18 +185,91 @@ def compute_stats(db: firestore.Client) -> None:
     spend_by_year: dict[str, float] = {}
     total_registrants = 0
 
-    for doc in db.collection(REGISTRANTS_COLLECTION).stream():
+    # Per-client and per-firm rollups, computed here (over the full,
+    # paginated registrants scan) instead of client-side in the frontend,
+    # which previously fetched only the first 2,000 of 25,000+ registrant
+    # docs (Firestore query limit) — silently showing an incomplete client
+    # and firm list. See pages/lobbying/clients/index.tsx and
+    # pages/lobbying/firms/index.tsx.
+    client_summaries: dict[str, dict] = {}
+    firm_summaries: dict[str, dict] = {}
+
+    for doc in _iter_collection(db, REGISTRANTS_COLLECTION):
         d = doc.to_dict()
-        year = str(d.get("year", ""))
-        for c in d.get("clients", []):
+        year = d.get("year")
+        year_str = str(year) if year is not None else ""
+        entity_name = d.get("entityName")
+        entity_norm = d.get("entityNameNorm")
+        reg_type = d.get("regType")
+        clients = d.get("clients", [])
+
+        if entity_norm:
+            firm = firm_summaries.setdefault(
+                entity_norm,
+                {
+                    "entityName": entity_name or entity_norm,
+                    "entityNameNorm": entity_norm,
+                    "regType": reg_type or "",
+                    "years": set(),
+                    "clientCount": 0,
+                },
+            )
+            if year is not None:
+                firm["years"].add(year)
+            if reg_type:
+                firm["regType"] = reg_type
+            # Matches the frontend's prior groupByFirm() semantics exactly:
+            # sum of raw clients[] array length, unfiltered.
+            firm["clientCount"] += len(clients)
+
+        for c in clients:
             norm = c.get("clientNameNorm")
-            if norm:
-                client_norms.add(norm)
+            name = c.get("clientName")
             comp = c.get("compensation")
-            if comp is not None and year:
-                spend_by_year[year] = spend_by_year.get(year, 0) + comp
+
+            if comp is not None and year_str:
+                spend_by_year[year_str] = spend_by_year.get(year_str, 0) + comp
+
+            if _is_legacy_total_client(name, norm):
+                continue
+
+            client_norms.add(norm)
+
+            cs = client_summaries.setdefault(
+                norm,
+                {
+                    "clientName": name or norm,
+                    "clientNameNorm": norm,
+                    "totalCompensation": None,
+                    "registrantCount": 0,
+                    "firms": {},
+                },
+            )
+            cs["registrantCount"] += 1
+            if comp is not None:
+                cs["totalCompensation"] = (cs["totalCompensation"] or 0) + comp
+
+            if entity_norm:
+                fb = cs["firms"].setdefault(
+                    entity_norm,
+                    {
+                        "entityName": entity_name or entity_norm,
+                        "entityNameNorm": entity_norm,
+                        "compensation": None,
+                    },
+                )
+                if comp is not None:
+                    fb["compensation"] = (fb["compensation"] or 0) + comp
+
         total_registrants += 1
 
+    for cs in client_summaries.values():
+        cs["firms"] = sorted(
+            cs["firms"].values(), key=lambda f: f["entityNameNorm"]
+        )
+    for fs in firm_summaries.values():
+        fs["years"] = sorted(fs["years"], reverse=True)
+
     stats = {
         "totalFilings": total_filings,
         "totalRegistrants": total_registrants,
@@ -143,14 +287,66 @@ def compute_stats(db: firestore.Client) -> None:
         client_filing_counts
     )
     for gc, bills_map in bill_summaries.items():
-        db.collection(STATS_COLLECTION).document(f"billSummaries_{gc}").set(
-            {"data": json.dumps(bills_map)}
+        # One small doc per bill, not one JSON blob per court: a court's blob
+        # eventually exceeds Firestore's 1MB field-size limit as its session
+        # accumulates filings (hit at 1,057KB for court 194 with ~5,600
+        # bills). Per-bill docs have no such ceiling.
+        parent_ref = db.collection(STATS_COLLECTION).document(f"billSummaries_{gc}")
+        parent_ref.set(
+            {"billCount": len(bills_map), "updatedAt": _now().isoformat()}
         )
+        bills_coll = parent_ref.collection("bills")
+        batch = db.batch()
+        count = 0
+        for bill_id, counts in bills_map.items():
+            batch.set(bills_coll.document(bill_id), counts)
+            count += 1
+            if count % 400 == 0:
+                batch.commit()
+                batch = db.batch()
+        if count % 400 != 0:
+            batch.commit()
+
+    # Client and firm summaries: same one-small-doc-per-item subcollection
+    # pattern as billSummaries above (avoids the 1MB per-document/field
+    # limit — at ~5,300 clients and ~4,800 firms this is already close to
+    # that ceiling as a single blob/map).
+    client_parent = db.collection(STATS_COLLECTION).document("clientSummaries")
+    client_parent.set(
+        {"count": len(client_summaries), "updatedAt": _now().isoformat()}
+    )
+    client_coll = client_parent.collection("clients")
+    batch = db.batch()
+    count = 0
+    for norm, cs in client_summaries.items():
+        batch.set(client_coll.document(_doc_id_for_norm(norm)), cs)
+        count += 1
+        if count % 400 == 0:
+            batch.commit()
+            batch = db.batch()
+    if count % 400 != 0:
+        batch.commit()
+
+    firm_parent = db.collection(STATS_COLLECTION).document("firmSummaries")
+    firm_parent.set({"count": len(firm_summaries), "updatedAt": _now().isoformat()})
+    firm_coll = firm_parent.collection("firms")
+    batch = db.batch()
+    count = 0
+    for norm, fs in firm_summaries.items():
+        batch.set(firm_coll.document(_doc_id_for_norm(norm)), fs)
+        count += 1
+        if count % 400 == 0:
+            batch.commit()
+            batch = db.batch()
+    if count % 400 != 0:
+        batch.commit()
+
     print(
         f"  stats written: {total_filings} filings, "
         f"{total_registrants} registrants, {len(client_norms)} clients, "
         f"{len(entity_filing_counts)} entities, {len(client_filing_counts)} client norms, "
-        f"bill summaries for courts {sorted(bill_summaries.keys())}"
+        f"bill summaries for courts {sorted(bill_summaries.keys())}, "
+        f"{len(client_summaries)} client summaries, {len(firm_summaries)} firm summaries"
     )
 
 
diff --git a/pages/lobbying/clients/[clientSlug].tsx b/pages/lobbying/clients/[clientSlug].tsx
index c4fcc6cb9..c2bf62625 100644
--- a/pages/lobbying/clients/[clientSlug].tsx
+++ b/pages/lobbying/clients/[clientSlug].tsx
@@ -6,7 +6,7 @@ import { createPage } from "components/page"
 import { createGetStaticTranslationProps } from "components/translations"
 import {
   useLobbyingFilingsForClient,
-  useLobbyingAllRegistrants
+  useLobbyingClientSummary
 } from "components/db/lobbying"
 import { LobbyingFilingsTable } from "components/lobbying/LobbyingFilingsTable"
 import {
@@ -18,43 +18,9 @@ import { LobbyingAttribution } from "components/lobbying/LobbyingAttribution"
 import { LobbyingSubnav } from "components/lobbying/LobbyingSubnav"
 import { usePagination } from "components/lobbying/usePagination"
 import { LobbyingPaginationBar } from "components/lobbying/LobbyingPaginationBar"
-import type { LobbyingRegistrant } from "functions/src/lobbying/types"
 
 const PAGE_SIZE = 25
 
-function findFirmsForClient(
-  registrants: LobbyingRegistrant[] | undefined,
-  clientNameNorm: string
-): Array<{
-  entityName: string
-  entityNameNorm: string
-  compensation: number | null
-}> {
-  if (!registrants) return []
-  const map = new Map<
-    string,
-    { entityName: string; entityNameNorm: string; compensation: number | null }
-  >()
-  for (const r of registrants) {
-    const match = r.clients.find(c => c.clientNameNorm === clientNameNorm)
-    if (!match) continue
-    if (!map.has(r.entityNameNorm)) {
-      map.set(r.entityNameNorm, {
-        entityName: r.entityName,
-        entityNameNorm: r.entityNameNorm,
-        compensation: null
-      })
-    }
-    const entry = map.get(r.entityNameNorm)!
-    if (match.compensation != null) {
-      entry.compensation = (entry.compensation ?? 0) + match.compensation
-    }
-  }
-  return [...map.values()].sort((a, b) =>
-    a.entityNameNorm.localeCompare(b.entityNameNorm)
-  )
-}
-
 function ClientDetail() {
   const { t } = useTranslation("lobbying")
   const { query } = useRouter()
@@ -68,12 +34,10 @@ function ClientDetail() {
     status: filStatus,
     error: filError
   } = useLobbyingFilingsForClient(clientNameNorm)
-  const { result: registrants, status: regStatus } = useLobbyingAllRegistrants()
+  const { result: clientSummary, status: summaryStatus } =
+    useLobbyingClientSummary(clientNameNorm)
 
-  const firms = useMemo(
-    () => findFirmsForClient(registrants, clientNameNorm),
-    [registrants, clientNameNorm]
-  )
+  const firms = clientSummary?.firms ?? []
 
   const positionCounts = useMemo(() => {
     const counts = { support: 0, oppose: 0, neutral: 0, none: 0 }
@@ -116,8 +80,8 @@ function ClientDetail() {
   const loading =
     filStatus === "loading" ||
     filStatus === "not-requested" ||
-    regStatus === "loading" ||
-    regStatus === "not-requested"
+    summaryStatus === "loading" ||
+    summaryStatus === "not-requested"
 
   const displayName = filings?.[0]?.clientName ?? clientNameNorm
 
diff --git a/pages/lobbying/clients/index.tsx b/pages/lobbying/clients/index.tsx
index 57e4b8e71..e25bb667e 100644
--- a/pages/lobbying/clients/index.tsx
+++ b/pages/lobbying/clients/index.tsx
@@ -4,17 +4,15 @@ import { Col, Container, Row } from "components/bootstrap"
 import { createPage } from "components/page"
 import { createGetStaticTranslationProps } from "components/translations"
 import {
-  useLobbyingAllRegistrants,
+  useLobbyingClientSummaries,
   useLobbyingClientFilingCounts
 } from "components/db/lobbying"
 import { MAPLE_COLORS } from "components/lobbying/chartTheme"
 import { LobbyingAttribution } from "components/lobbying/LobbyingAttribution"
 import { usePagination } from "components/lobbying/usePagination"
 import { LobbyingPaginationBar } from "components/lobbying/LobbyingPaginationBar"
-import type { LobbyingRegistrant } from "functions/src/lobbying/types"
 import { LobbyingSubnav } from "components/lobbying/LobbyingSubnav"
 
-const LEGACY_TOTAL_CLIENT = "_total_salary_"
 const PAGE_SIZE = 50
 
 type ClientSortKey = "name" | "compensation" | "firms" | "filings"
@@ -65,41 +63,6 @@ type ClientRow = {
   totalFilings: number | undefined
 }
 
-function deriveClients(
-  registrants: LobbyingRegistrant[] | undefined
-): ClientRow[] {
-  if (!registrants) return []
-  const map = new Map<string, ClientRow>()
-  for (const r of registrants) {
-    for (const c of r.clients) {
-      const lc = c.clientName.toLowerCase()
-      if (
-        !c.clientNameNorm ||
-        c.clientNameNorm === LEGACY_TOTAL_CLIENT ||
-        c.clientName === LEGACY_TOTAL_CLIENT ||
-        lc.includes("total salaries") ||
-        lc.includes("total salary")
-      )
-        continue
-      if (!map.has(c.clientNameNorm)) {
-        map.set(c.clientNameNorm, {
-          clientName: c.clientName,
-          clientNameNorm: c.clientNameNorm,
-          totalCompensation: null,
-          registrantCount: 0,
-          totalFilings: undefined
-        })
-      }
-      const row = map.get(c.clientNameNorm)!
-      row.registrantCount++
-      if (c.compensation != null) {
-        row.totalCompensation = (row.totalCompensation ?? 0) + c.compensation
-      }
-    }
-  }
-  return [...map.values()]
-}
-
 function LobbyingClientsTable() {
   const { t } = useTranslation("lobbying")
   const [search, setSearch] = useState("")
@@ -115,19 +78,18 @@ function LobbyingClientsTable() {
     }
   }
 
-  const { result: registrants, status, error } = useLobbyingAllRegistrants()
+  const { result: summaries, status, error } = useLobbyingClientSummaries()
   const { result: filCounts } = useLobbyingClientFilingCounts()
-  const clients = useMemo(() => deriveClients(registrants), [registrants])
-  const clientsWithCounts = useMemo(
-    () =>
-      filCounts
-        ? clients.map(c => ({
-            ...c,
-            totalFilings: filCounts[c.clientNameNorm]
-          }))
-        : clients,
-    [clients, filCounts]
-  )
+  const clientsWithCounts = useMemo<ClientRow[]>(() => {
+    if (!summaries) return []
+    return summaries.map(c => ({
+      clientName: c.clientName,
+      clientNameNorm: c.clientNameNorm,
+      totalCompensation: c.totalCompensation,
+      registrantCount: c.registrantCount,
+      totalFilings: filCounts?.[c.clientNameNorm]
+    }))
+  }, [summaries, filCounts])
 
   const filtered = useMemo(
     () =>
diff --git a/pages/lobbying/firms/index.tsx b/pages/lobbying/firms/index.tsx
index 02c9dc5e3..f0571a0ee 100644
--- a/pages/lobbying/firms/index.tsx
+++ b/pages/lobbying/firms/index.tsx
@@ -4,7 +4,7 @@ import { Col, Container, Row } from "components/bootstrap"
 import { createPage } from "components/page"
 import { createGetStaticTranslationProps } from "components/translations"
 import {
-  useLobbyingAllRegistrants,
+  useLobbyingFirmSummaries,
   useLobbyingEntityFilingCounts
 } from "components/db/lobbying"
 import { MAPLE_COLORS } from "components/lobbying/chartTheme"
@@ -12,7 +12,6 @@ import { LobbyingAttribution } from "components/lobbying/LobbyingAttribution"
 import { usePagination } from "components/lobbying/usePagination"
 import { LobbyingPaginationBar } from "components/lobbying/LobbyingPaginationBar"
 import { LobbyingSubnav } from "components/lobbying/LobbyingSubnav"
-import type { LobbyingRegistrant } from "functions/src/lobbying/types"
 
 const PAGE_SIZE = 50
 
@@ -59,36 +58,12 @@ function SortTh({
 type FirmRow = {
   entityName: string
   entityNameNorm: string
-  registrantId: string
   regType: string
   years: number[]
   clientCount: number
   totalFilings: number | undefined
 }
 
-function groupByFirm(registrants: LobbyingRegistrant[] | undefined): FirmRow[] {
-  if (!registrants) return []
-  const map = new Map<string, FirmRow>()
-  for (const r of registrants) {
-    if (!map.has(r.entityNameNorm)) {
-      map.set(r.entityNameNorm, {
-        entityName: r.entityName,
-        entityNameNorm: r.entityNameNorm,
-        registrantId: r.registrantId,
-        regType: r.regType,
-        years: [],
-        clientCount: 0,
-        totalFilings: undefined
-      })
-    }
-    const row = map.get(r.entityNameNorm)!
-    if (!row.years.includes(r.year)) row.years.push(r.year)
-    row.clientCount += r.clients.length
-  }
-  for (const row of map.values()) row.years.sort((a, b) => b - a)
-  return [...map.values()]
-}
-
 function LobbyingFirmsTable() {
   const { t } = useTranslation("lobbying")
   const [regTypeFilter, setRegTypeFilter] = useState<
@@ -107,16 +82,19 @@ function LobbyingFirmsTable() {
     }
   }
 
-  const { result: registrants, status, error } = useLobbyingAllRegistrants()
+  const { result: summaries, status, error } = useLobbyingFirmSummaries()
   const { result: filCounts } = useLobbyingEntityFilingCounts()
-  const firms = useMemo(() => groupByFirm(registrants), [registrants])
-  const firmsWithCounts = useMemo(
-    () =>
-      filCounts
-        ? firms.map(f => ({ ...f, totalFilings: filCounts[f.entityNameNorm] }))
-        : firms,
-    [firms, filCounts]
-  )
+  const firmsWithCounts = useMemo<FirmRow[]>(() => {
+    if (!summaries) return []
+    return summaries.map(f => ({
+      entityName: f.entityName,
+      entityNameNorm: f.entityNameNorm,
+      regType: f.regType,
+      years: f.years,
+      clientCount: f.clientCount,
+      totalFilings: filCounts?.[f.entityNameNorm]
+    }))
+  }, [summaries, filCounts])
 
   const filtered = useMemo(() => {
     return firmsWithCounts.filter(f => {
diff --git a/public/locales/en/lobbying.json b/public/locales/en/lobbying.json
index ac9d801b3..c34edf1db 100644
--- a/public/locales/en/lobbying.json
+++ b/public/locales/en/lobbying.json
@@ -7,7 +7,8 @@
   },
   "pagination": {
     "prev": "← Prev",
-    "next": "Next →"
+    "next": "Next →",
+    "pageNumber": "Page number"
   },
   "misc": {
     "disclosures": "Disclosures",
@@ -24,13 +25,13 @@
   "stats": {
     "totalBills": "Lobbied bills",
     "totalClients": "Clients",
-    "totalFirms": "Lobbying firms",
+    "totalFirms": "Lobbyists",
     "totalSpend": "Total compensation reported"
   },
   "sections": {
     "bills": "Bills",
     "clients": "Clients",
-    "firms": "Lobbying Firms"
+    "firms": "Lobbyists"
   },
   "filters": {
     "session": "Session",
@@ -49,7 +50,7 @@
   },
   "fields": {
     "clientName": "Client",
-    "firmName": "Lobbying Firm",
+    "firmName": "Lobbyist",
     "amount": "Compensation",
     "year": "Year",
     "filings": "Filings",
@@ -60,12 +61,12 @@
     "activity": "Activity",
     "sessions": "Sessions",
     "type": "Type",
-    "firms": "Firms"
+    "firms": "Lobbyists"
   },
   "explainers": {
     "statBills": "Bills that have been the subject of at least one lobbying disclosure.",
     "statClients": "Organizations that have hired lobbyists to advocate on their behalf.",
-    "statSpend": "Sum of all annual compensation amounts reported by lobbying firms to the Secretary of State.",
+    "statSpend": "Sum of all annual compensation amounts reported by lobbyists to the Secretary of State.",
     "bills": "Bills that appear in lobbying disclosures filed with the MA Secretary of State, across all sessions in this dataset.",
     "clients": "Organizations that retained lobbyists or lobbied on their own behalf, as reported in annual compensation disclosures.",
     "firms": "Registered lobbyists and employers who filed lobbying disclosures with the MA Secretary of State."
@@ -77,14 +78,16 @@
     "overview": "Lobbying Explorer",
     "bills": "Lobbied Bills",
     "clients": "Clients & Sponsors",
-    "firms": "Lobbying Firms",
+    "firms": "Lobbyists",
     "client": "Client Profile",
-    "firm": "Firm Profile"
+    "firm": "Lobbyist Profile"
   },
   "billCard": {
+    "title": "Lobbying Activity",
     "filingCount_one": "{{count}} lobbying filing",
     "filingCount_other": "{{count}} lobbying filings",
-    "viewAll": "View all lobbying activity →",
-    "showingOf": "Showing {{showing}} of {{total}} filings"
+    "viewAll": "Open in Lobbying Explorer →",
+    "showingOf": "Showing {{showing}} of {{total}} filings",
+    "filingsLabel": "Lobbying filings"
   }
 }
diff --git a/scripts/firebase-admin/checkLobbyingFreshness.ts b/scripts/firebase-admin/checkLobbyingFreshness.ts
new file mode 100644
index 000000000..16809de03
--- /dev/null
+++ b/scripts/firebase-admin/checkLobbyingFreshness.ts
@@ -0,0 +1,78 @@
+import { Script } from "./types"
+
+export const script: Script = async ({ db }) => {
+  const filingsSnap = await db
+    .collection("lobbyingFilings")
+    .orderBy("fetchedAt", "desc")
+    .limit(5)
+    .get()
+
+  console.log(`lobbyingFilings: most recently fetched docs`)
+  filingsSnap.docs.forEach(doc => {
+    const d = doc.data()
+    console.log(
+      `  fetchedAt=${d.fetchedAt?.toDate?.()?.toISOString()} year=${
+        d.year
+      } gc=${d.generalCourt} entity=${d.entityName}`
+    )
+  })
+
+  const registrantsSnap = await db
+    .collection("lobbyingRegistrants")
+    .orderBy("fetchedAt", "desc")
+    .limit(5)
+    .get()
+
+  console.log(`\nlobbyingRegistrants: most recently fetched docs`)
+  registrantsSnap.docs.forEach(doc => {
+    const d = doc.data()
+    console.log(
+      `  fetchedAt=${d.fetchedAt?.toDate?.()?.toISOString()} year=${
+        d.year
+      } entity=${d.entityName}`
+    )
+  })
+
+  const scraperDoc = await db.doc("scrapers/lobbying").get()
+  console.log(`\nscrapers/lobbying doc exists: ${scraperDoc.exists}`)
+  const processedUrlsSnap = await db
+    .collection("scrapers/lobbying/processedUrls")
+    .get()
+  const summaryCacheSnap = await db
+    .collection("scrapers/lobbying/summaryCache")
+    .get()
+  console.log(`  processedUrls subcollection: ${processedUrlsSnap.size} URLs`)
+  console.log(
+    `  summaryCache subcollection: ${summaryCacheSnap.size} registrant summaries cached`
+  )
+
+  const backfillDoc = await db.doc("scrapers/lobbyingBackfill").get()
+  console.log(`\nscrapers/lobbyingBackfill doc exists: ${backfillDoc.exists}`)
+  if (backfillDoc.exists) {
+    console.log(`  completedYears: ${backfillDoc.data()?.completedYears}`)
+  }
+
+  const currentYear = new Date().getFullYear()
+  const yRegistrants = await db
+    .collection("lobbyingRegistrants")
+    .where("year", "==", currentYear)
+    .get()
+  const yFilings = await db
+    .collection("lobbyingFilings")
+    .where("year", "==", currentYear)
+    .get()
+  console.log(
+    `\n${currentYear}: ${yRegistrants.size} registrants, ${yFilings.size} filings`
+  )
+
+  // Grand totals come from the last computed stats doc rather than a live
+  // full-collection scan — cheap, and the corpus (300K+ docs) makes an
+  // aggregate scan here wasteful for what's meant to be a quick check.
+  const statsDoc = await db.doc("lobbyingMeta/stats").get()
+  const stats = statsDoc.data()
+  console.log(
+    `\nlobbyingMeta/stats (as of last compute): ${
+      stats?.totalFilings ?? "?"
+    } filings, ${stats?.totalRegistrants ?? "?"} registrants`
+  )
+}
diff --git a/scripts/firebase-admin/seedLobbyingStats.ts b/scripts/firebase-admin/seedLobbyingStats.ts
index 42f959c72..f0de03bfb 100644
--- a/scripts/firebase-admin/seedLobbyingStats.ts
+++ b/scripts/firebase-admin/seedLobbyingStats.ts
@@ -26,6 +26,43 @@ const REGISTRANTS_COLLECTION = "lobbyingRegistrants"
 const STATS_COLLECTION = "lobbyingMeta"
 const STATS_DOC_ID = "stats"
 
+// Sentinel clientName used for pre-2013 legacy filings where compensation is
+// reported as a single total rather than broken down per client. Must match
+// LEGACY_TOTAL_CLIENT in functions/src/lobbying/types.ts.
+const LEGACY_TOTAL_CLIENT = "_total_salary_"
+
+function isLegacyTotalClient(
+  name: string | undefined,
+  nameNorm: string | undefined
+): boolean {
+  if (!nameNorm || nameNorm === LEGACY_TOTAL_CLIENT) return true
+  if (name === LEGACY_TOTAL_CLIENT) return true
+  const lc = (name ?? "").toLowerCase()
+  return lc.includes("total salaries") || lc.includes("total salary")
+}
+
+type FirmBreakdownEntry = {
+  entityName: string
+  entityNameNorm: string
+  compensation: number | null
+}
+
+type ClientSummary = {
+  clientName: string
+  clientNameNorm: string
+  totalCompensation: number | null
+  registrantCount: number
+  firms: FirmBreakdownEntry[]
+}
+
+type FirmSummary = {
+  entityName: string
+  entityNameNorm: string
+  regType: string
+  years: number[]
+  clientCount: number
+}
+
 export const script: Script = async ({ db }) => {
   console.log("Reading lobbyingFilings…")
   const filingsSnap = await db.collection(FILINGS_COLLECTION).get()
@@ -104,18 +141,103 @@ export const script: Script = async ({ db }) => {
   // Aggregate spend and unique clients from registrant docs.
   // Registrant clients[].compensation is the annual total paid per client
   // relationship — more accurate than the per-bill amount on filings.
+  //
+  // Also builds per-client and per-firm rollups here (over the full
+  // registrants collection) instead of leaving the frontend to derive them
+  // client-side, which previously only fetched the first 2,000 of 25,000+
+  // registrant docs (Firestore query limit) — silently showing an
+  // incomplete client/firm list. See pages/lobbying/clients/index.tsx and
+  // pages/lobbying/firms/index.tsx.
   const clientNorms = new Set<string>()
   const spendByYear: Record<string, number> = {}
+  const clientSummaries: Record<
+    string,
+    ClientSummary & { firmsMap: Record<string, FirmBreakdownEntry> }
+  > = {}
+  const firmSummaries: Record<string, FirmSummary> = {}
+
   for (const doc of registrantsSnap.docs) {
     const d = doc.data()
-    const y = String(d.year)
-    for (const c of d.clients ?? []) {
-      if (c.clientNameNorm) clientNorms.add(c.clientNameNorm)
-      if (c.compensation != null) {
-        spendByYear[y] = (spendByYear[y] ?? 0) + c.compensation
+    const year: number | undefined = d.year
+    const y = String(year)
+    const entityName: string | undefined = d.entityName
+    const entityNorm: string | undefined = d.entityNameNorm
+    const regType: string | undefined = d.regType
+    const clients = d.clients ?? []
+
+    if (entityNorm) {
+      if (!firmSummaries[entityNorm]) {
+        firmSummaries[entityNorm] = {
+          entityName: entityName ?? entityNorm,
+          entityNameNorm: entityNorm,
+          regType: regType ?? "",
+          years: [],
+          clientCount: 0
+        }
       }
+      const firm = firmSummaries[entityNorm]
+      if (year != null && !firm.years.includes(year)) firm.years.push(year)
+      if (regType) firm.regType = regType
+      // Matches the frontend's prior groupByFirm() semantics exactly: sum
+      // of raw clients[] array length, unfiltered.
+      firm.clientCount += clients.length
     }
+
+    for (const c of clients) {
+      const norm: string | undefined = c.clientNameNorm
+      const name: string | undefined = c.clientName
+      const comp: number | null | undefined = c.compensation
+
+      if (comp != null) {
+        spendByYear[y] = (spendByYear[y] ?? 0) + comp
+      }
+
+      if (isLegacyTotalClient(name, norm)) continue
+      if (!norm) continue
+
+      clientNorms.add(norm)
+
+      if (!clientSummaries[norm]) {
+        clientSummaries[norm] = {
+          clientName: name ?? norm,
+          clientNameNorm: norm,
+          totalCompensation: null,
+          registrantCount: 0,
+          firms: [],
+          firmsMap: {}
+        }
+      }
+      const cs = clientSummaries[norm]
+      cs.registrantCount++
+      if (comp != null) {
+        cs.totalCompensation = (cs.totalCompensation ?? 0) + comp
+      }
+
+      if (entityNorm) {
+        if (!cs.firmsMap[entityNorm]) {
+          cs.firmsMap[entityNorm] = {
+            entityName: entityName ?? entityNorm,
+            entityNameNorm: entityNorm,
+            compensation: null
+          }
+        }
+        const fb = cs.firmsMap[entityNorm]
+        if (comp != null) {
+          fb.compensation = (fb.compensation ?? 0) + comp
+        }
+      }
+    }
+  }
+
+  for (const cs of Object.values(clientSummaries)) {
+    cs.firms = Object.values(cs.firmsMap).sort((a, b) =>
+      a.entityNameNorm.localeCompare(b.entityNameNorm)
+    )
   }
+  for (const fs of Object.values(firmSummaries)) {
+    fs.years.sort((a, b) => b - a)
+  }
+
   const totalClients = clientNorms.size
 
   const stats = {
@@ -151,13 +273,71 @@ export const script: Script = async ({ db }) => {
     .set(clientFilingCounts)
 
   for (const [court, billsMap] of Object.entries(billSummaries)) {
-    await db
+    // One small doc per bill, not one JSON blob per court: a court's blob
+    // eventually exceeds Firestore's 1MB field-size limit as its session
+    // accumulates filings (hit at 1,057KB for court 194 with ~5,600 bills).
+    // Per-bill docs have no such ceiling.
+    const parentRef = db
       .collection(STATS_COLLECTION)
       .doc(`billSummaries_${court}`)
-      .set({ data: JSON.stringify(billsMap) })
+    const entries = Object.entries(billsMap)
+    await parentRef.set({
+      billCount: entries.length,
+      updatedAt: new Date().toISOString()
+    })
+    const billsColl = parentRef.collection("bills")
+    for (let i = 0; i < entries.length; i += 400) {
+      const batch = db.batch()
+      for (const [billId, counts] of entries.slice(i, i + 400)) {
+        batch.set(billsColl.doc(billId), counts)
+      }
+      await batch.commit()
+    }
+  }
+
+  // Client and firm summaries: same one-small-doc-per-item subcollection
+  // pattern as billSummaries above (avoids the 1MB per-document/field limit
+  // — at ~5,300 clients and ~4,800 firms this is already close to that
+  // ceiling as a single blob/map). Doc IDs are encodeURIComponent(norm), so
+  // the frontend can look up one client/firm directly without fetching the
+  // whole subcollection.
+  const clientParentRef = db.collection(STATS_COLLECTION).doc("clientSummaries")
+  const clientEntries = Object.entries(clientSummaries)
+  await clientParentRef.set({
+    count: clientEntries.length,
+    updatedAt: new Date().toISOString()
+  })
+  const clientsColl = clientParentRef.collection("clients")
+  for (let i = 0; i < clientEntries.length; i += 400) {
+    const batch = db.batch()
+    for (const [norm, { firmsMap: _firmsMap, ...cs }] of clientEntries.slice(
+      i,
+      i + 400
+    )) {
+      batch.set(clientsColl.doc(encodeURIComponent(norm)), cs)
+    }
+    await batch.commit()
+  }
+
+  const firmParentRef = db.collection(STATS_COLLECTION).doc("firmSummaries")
+  const firmEntries = Object.entries(firmSummaries)
+  await firmParentRef.set({
+    count: firmEntries.length,
+    updatedAt: new Date().toISOString()
+  })
+  const firmsColl = firmParentRef.collection("firms")
+  for (let i = 0; i < firmEntries.length; i += 400) {
+    const batch = db.batch()
+    for (const [norm, fs] of firmEntries.slice(i, i + 400)) {
+      batch.set(firmsColl.doc(encodeURIComponent(norm)), fs)
+    }
+    await batch.commit()
   }
 
   console.log(`Written to ${STATS_COLLECTION}/${STATS_DOC_ID}`)
+  console.log(
+    `  clientSummaries: ${clientEntries.length}, firmSummaries: ${firmEntries.length}`
+  )
   console.log(
     `  entityFilingCounts: ${Object.keys(entityFilingCounts).length} entities`
   )