@@ -219,3 +219,126 @@ def test_search_returns_empty_when_the_source_has_no_results(
         )
         == []
     )
+
+
+# ---------------------------------------------------------------------
+# NARA — the v2 catalog API.
+#
+# The adapter used to call `/api/v2/search` with `rows`/`offset` and read
+# `results[]`. That path now serves the catalog's HTML app rather than
+# JSON, so every search raised on `.json()`. These tests pin the shape of
+# the working endpoint so a silent regression back to v1 is caught here
+# rather than as an empty corpus.
+# ---------------------------------------------------------------------
+
+_NARA_RECORD = {
+    "naId": 7500,
+    "title": "APOLLO 8 NASA MOON FOOTAGE",
+    "generalNotes": ["Following is a general summary of scenes in the roll."],
+    "shotList": "CU Radar antenna. Overall view of earth.",
+    "subjects": [{"heading": "National Aeronautics and Space Administration."}],
+    "digitalObjects": [
+        {
+            "objectId": "14869152",
+            "objectFilename": "33050.mp4",
+            "objectType": "Audio/Visual File (MP4)",
+            "objectUrl": "https://catalog.archives.gov/media/33050.mp4",
+            "objectFileSize": 12345,
+        },
+        {
+            "objectId": "14869153",
+            "objectFilename": "97-445.jpg",
+            "objectType": "Image (JPG)",
+            "objectUrl": "https://catalog.archives.gov/media/97-445.jpg",
+        },
+    ],
+}
+
+
+class _NARAResponse:
+    """A 200 carrying one record in the v2 envelope."""
+
+    status_code = 200
+
+    def raise_for_status(self):
+        return None
+
+    def json(self):
+        return {"body": {"hits": {"hits": [{"_source": {"record": _NARA_RECORD}}]}}}
+
+
+def _nara_search(monkeypatch, kind, page=1):
+    seen = {}
+
+    def fake_get(url, headers=None, params=None, timeout=None):
+        seen["url"] = url
+        seen["headers"] = headers or {}
+        seen["params"] = params or {}
+        return _NARAResponse()
+
+    _install_fake_transport(monkeypatch, fake_get)
+    results = get_source("nara").search(
+        "apollo", SearchFilters(kind=kind, per_page=5, page=page)
+    )
+    return seen, results
+
+
+def test_nara_requires_a_key():
+    # v2 serves HTML without x-api-key, so "no key" is unavailable, not
+    # merely rate-limited.
+    source = get_source("nara")
+    assert source.is_available() is False
+
+
+def test_nara_calls_the_v2_records_endpoint(monkeypatch):
+    seen, _ = _nara_search(monkeypatch, "video", page=3)
+
+    assert seen["url"].endswith("/api/v2/records/search")
+    assert seen["headers"]["x-api-key"] == "test-nara"
+    # v2 paginates by 1-based `page`; offset/from/start are ignored.
+    assert seen["params"]["page"] == 3
+    assert "offset" not in seen["params"]
+    assert seen["params"]["limit"] == 5
+    assert seen["params"]["availableOnline"] == "true"
+
+
+def test_nara_pushes_the_kind_filter_server_side(monkeypatch):
+    seen, _ = _nara_search(monkeypatch, "video")
+    assert seen["params"]["typeOfMaterials"] == "Moving Images"
+
+    seen, _ = _nara_search(monkeypatch, "image")
+    assert seen["params"]["typeOfMaterials"] == "Photographs and other Graphic Materials"
+
+    # "any" must not constrain the search to one material type.
+    seen, _ = _nara_search(monkeypatch, "any")
+    assert "typeOfMaterials" not in seen["params"]
+
+
+def test_nara_extracts_objects_from_the_v2_envelope(monkeypatch):
+    _, videos = _nara_search(monkeypatch, "video")
+
+    assert len(videos) == 1
+    clip = videos[0]
+    assert clip.download_url == "https://catalog.archives.gov/media/33050.mp4"
+    assert clip.kind == "video"
+    assert clip.source_url == "https://catalog.archives.gov/id/7500"
+    assert clip.source_id == "7500_14869152"
+    # v2 reports no dimensions or duration; the Candidate says so rather
+    # than inventing them.
+    assert (clip.width, clip.height, clip.duration) == (0, 0, 0.0)
+
+    _, images = _nara_search(monkeypatch, "image")
+    assert [i.kind for i in images] == ["image"]
+    assert images[0].download_url.endswith("97-445.jpg")
+
+
+def test_nara_tags_flatten_the_scattered_prose_fields(monkeypatch):
+    # v2 dropped scopeAndContentNote; the description now lives across
+    # generalNotes, shotList and subjects.
+    _, videos = _nara_search(monkeypatch, "video")
+    tags = videos[0].source_tags
+
+    assert "APOLLO 8 NASA MOON FOOTAGE" in tags
+    assert "general summary of scenes" in tags
+    assert "Radar antenna" in tags
+    assert "National Aeronautics" in tags
@@ -5,15 +5,16 @@
 significant film and video holdings — all U.S. federal government work
 and therefore public domain.
 
-No API key required for basic searching. For higher rate limits, email
-Catalog_API@nara.gov to request a key. Rate limit: ~10,000 queries per
-month per API key.
+An API key IS required: without an ``x-api-key`` header the catalog serves
+its HTML single-page app instead of JSON, so every search fails to parse.
+Request one from Catalog_API@nara.gov and set ``NARA_API_KEY``. Rate limit
+is ~10,000 queries per month per key.
 
 Fetch pattern
 -------------
-Two-stage like NASA. The search endpoint returns metadata records. Each
-record may contain digital objects (files) in ``objects``. We follow
-those to find downloadable video files.
+Two-stage like NASA. ``/records/search`` returns metadata records under
+``body.hits.hits[]._source.record``; each record may carry ``digitalObjects``
+(files), which is what we follow to find downloadable media.
 
 What NARA is good for
 ---------------------
@@ -33,9 +34,20 @@
 
 _log = logging.getLogger(__name__)
 
-_SEARCH_URL = "https://catalog.archives.gov/api/v2/search"
+_SEARCH_URL = "https://catalog.archives.gov/api/v2/records/search"
 _LICENSE = "Public domain (U.S. federal government work)"
 
+# The catalog's own facet values (aggregation "typeOfMaterials"), used to
+# push the video/image split server-side instead of over-fetching and
+# discarding. Anything else is left unfiltered.
+_MATERIAL_FILTER = {
+    "video": "Moving Images",
+    "image": "Photographs and other Graphic Materials",
+}
+
+_VIDEO_EXT = ("mp4", "mov", "avi", "wmv", "mkv", "webm", "m4v", "mpg", "mpeg")
+_IMAGE_EXT = ("jpg", "jpeg", "png", "tif", "tiff", "gif", "bmp")
+
 
 class NARASource:
     """U.S. National Archives adapter. Satisfies `StockSource`."""
@@ -45,41 +57,46 @@ class NARASource:
     provider = "nara"
     priority = 35
     install_instructions = (
-        "NARA works without an API key. "
-        "Set NARA_API_KEY in .env for higher rate limits."
+        "NARA requires an API key: without one the catalog returns HTML, not JSON.\n"
+        "Request a key from Catalog_API@nara.gov, then set NARA_API_KEY in .env."
     )
     supports = {"video": True, "image": True}
 
     def is_available(self) -> bool:
-        # NARA is always available (no key required)
-        return True
+        return bool(os.environ.get("NARA_API_KEY"))
 
     def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
         import requests
 
+        api_key = os.environ.get("NARA_API_KEY")
+        if not api_key:
+            raise RuntimeError(
+                "NARA_API_KEY is not set. " + self.install_instructions
+            )
+
         kind = (filters.kind or "video").lower()
 
         params: dict[str, Any] = {
             "q": query,
-            "rows": max(1, min(filters.per_page, 50)),
-            "offset": (max(1, filters.page) - 1) * filters.per_page,
+            "limit": max(1, min(filters.per_page, 50)),
+            # 1-based. The v2 API ignores offset/from/start entirely.
+            #
+            # Ordering is not stable: identical requests come back with
+            # relevance ties shuffled, so consecutive pages can overlap or
+            # skip a record. Nothing to fix here — de-dup downstream on
+            # `source_id` if a caller walks several pages.
+            "page": max(1, filters.page),
+            # Records with no digitised file can never yield a candidate.
+            "availableOnline": "true",
         }
-
-        # Filter by type if possible
-        if kind == "video":
-            params["type"] = "moving-image"
-        elif kind == "image":
-            params["type"] = "still-image"
-
-        headers: dict[str, str] = {}
-        api_key = os.environ.get("NARA_API_KEY")
-        if api_key:
-            headers["x-api-key"] = api_key
+        material = _MATERIAL_FILTER.get(kind)
+        if material:
+            params["typeOfMaterials"] = material
 
         try:
             r = requests.get(
                 _SEARCH_URL,
-                headers=headers,
+                headers={"x-api-key": api_key},
                 params=params,
                 timeout=30,
             )
@@ -91,53 +108,72 @@ def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
             _log.warning("NARA search failed: %s", e)
             raise
 
-        results = data.get("results", []) or []
+        hits = (
+            (data.get("body") or {})
+            .get("hits", {})
+            .get("hits", [])
+        ) or []
+
         out: list[Candidate] = []
+        for hit in hits:
+            record = ((hit.get("_source") or {}).get("record")) or {}
+            out.extend(self._extract_candidates(record, kind, filters))
+        return out
 
-        for item in results:
-            candidates = self._extract_candidates(item, kind, filters)
-            out.extend(candidates)
+    @staticmethod
+    def _describe(record: dict) -> str:
+        """Flatten the record's prose fields into one searchable tag blob.
 
-        return out
+        v2 dropped ``scopeAndContentNote``; what description exists is spread
+        across ``generalNotes`` (list of str), ``shotList`` (str) and
+        ``subjects`` (list of dicts with a ``heading``).
+        """
+        parts: list[str] = [str(record.get("title") or "")]
+
+        notes = record.get("generalNotes")
+        if isinstance(notes, list):
+            parts += [str(n) for n in notes if isinstance(n, str)]
+
+        shot_list = record.get("shotList")
+        if isinstance(shot_list, str):
+            parts.append(shot_list)
+
+        subjects = record.get("subjects")
+        if isinstance(subjects, list):
+            parts += [
+                str(s.get("heading"))
+                for s in subjects
+                if isinstance(s, dict) and s.get("heading")
+            ]
+
+        return " ".join(p for p in parts if p).strip()
 
     def _extract_candidates(
-        self, item: dict, kind: str, filters: SearchFilters
+        self, record: dict, kind: str, filters: SearchFilters
     ) -> list[Candidate]:
         """Extract downloadable candidates from a NARA catalog record."""
-        naid = str(item.get("naId", "") or "")
+        naid = str(record.get("naId", "") or "")
         if not naid:
             return []
 
-        title = item.get("title", "") or ""
-        description = item.get("scopeAndContentNote", "") or ""
-        source_tags = f"{title} {description}".strip()
+        source_tags = self._describe(record)
         source_url = f"https://catalog.archives.gov/id/{naid}"
 
-        # Look for digital objects
-        objects = item.get("objects", []) or []
-        if not objects:
-            # Try alternate field names
-            digital = item.get("digitalObjects", []) or []
-            objects = digital
-
         out: list[Candidate] = []
-        for obj in objects:
-            file_url = obj.get("url") or obj.get("fileUrl") or ""
+        for obj in record.get("digitalObjects") or []:
+            file_url = obj.get("objectUrl") or ""
             if not file_url:
                 continue
 
-            # Determine kind from mime type or file extension
-            mime = (obj.get("mimeType", "") or "").lower()
-            ext = file_url.rsplit(".", 1)[-1].lower() if "." in file_url else ""
+            # v2 has no mime type; it labels files like "Audio/Visual File (MP4)"
+            # or "Image (JPG)". The extension is the reliable signal, with the
+            # label as a fallback for the odd extensionless URL.
+            label = str(obj.get("objectType") or "").lower()
+            name = str(obj.get("objectFilename") or file_url)
+            ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
 
-            is_video = (
-                "video" in mime
-                or ext in ("mp4", "mov", "avi", "wmv", "mkv", "webm")
-            )
-            is_image = (
-                "image" in mime
-                or ext in ("jpg", "jpeg", "png", "tif", "tiff", "gif")
-            )
+            is_video = ext in _VIDEO_EXT or (not ext and "audio/visual" in label)
+            is_image = ext in _IMAGE_EXT or (not ext and "image" in label)
 
             if kind == "video" and not is_video:
                 continue
@@ -146,36 +182,27 @@ def _extract_candidates(
             if not is_video and not is_image:
                 continue
 
-            candidate_kind = "video" if is_video else "image"
-            width = int(obj.get("width") or 0)
-            height = int(obj.get("height") or 0)
-            duration = float(obj.get("duration") or 0)
-
-            # Duration filters (client-side)
-            if candidate_kind == "video":
-                if filters.min_duration and duration and duration < filters.min_duration:
-                    continue
-                if filters.max_duration and duration and duration > filters.max_duration:
-                    continue
-
+            # v2 reports no dimensions or duration, so duration filters can only
+            # be applied downstream once the file is probed. Leaving these at 0
+            # keeps the Candidate honest rather than inventing values.
             out.append(
                 Candidate(
                     source=self.name,
                     source_id=f"{naid}_{obj.get('objectId', len(out))}",
                     source_url=source_url,
                     download_url=file_url,
-                    kind=candidate_kind,
-                    width=width,
-                    height=height,
-                    duration=duration,
+                    kind="video" if is_video else "image",
+                    width=0,
+                    height=0,
+                    duration=0.0,
                     creator="U.S. National Archives",
                     license=_LICENSE,
                     source_tags=source_tags,
-                    thumbnail_url=obj.get("thumbnailUrl", "") or "",
+                    thumbnail_url="",
                     extra={
                         "naId": naid,
-                        "mime": mime,
-                        "fileSize": obj.get("fileSize"),
+                        "objectType": obj.get("objectType"),
+                        "fileSize": obj.get("objectFileSize"),
                     },
                 )
             )