docmontage: corpus builder hardening from P1 + P2 audit observations

All of these changes came out of running the P1 and P2 documentary-montage
audits end-to-end and watching specific things break. Grouping them into
one commit because they share a theme: making the corpus builder and its
stock source adapters robust enough that a real brief can produce a
real corpus without special-casing.

corpus_builder.py + new clip_cache.py + test_clip_cache.py
  Phase 1 of the shared-corpus architecture. Adds
  ~/.openmontage/clips_cache/ — a process-safe, LRU-evicted cache of
  downloaded clip files keyed by clip_id. Before each candidate download,
  corpus_builder asks the cache whether the bytes already exist on disk
  from a previous run; on a hit it hard-links (or copies on cross-drive)
  the blob into the caller's corpus dir and skips the network fetch. On
  a miss it downloads as usual and ingests the fresh file. Motivation:
  re-running the P1 audit after every tool fix was re-downloading gigs
  of archive.org footage that had already been fetched in the prior run.
  Cache faults never block the pipeline — they degrade gracefully to
  normal downloads. The cache bubbles counters into the corpus_builder
  return payload so the production report can show hit/miss/bytes-saved.
  Default 20 GB cap, overridable via OPENMONTAGE_CACHE_MAX_GB.
  Full test coverage: try_link, ingest, stats, LRU eviction, manifest
  persistence, lock behavior. 23 tests, tmp_path-scoped.

stock_sources/archive_org.py
  1. Three-strategy query cascade (phrase_prox_10 → distinctive_and →
     distinctive_or). Motivation: natural-language documentary queries
     against archive.org Solr were zeroing out — "1950s family watching
     television" returned 0 hits because Solr's default multi-term AND
     over-intersects. Walks strict to loose and returns the first
     non-empty strategy. Stop words, source hints ("prelinger",
     "archive", "footage"), and year tokens ("1950s") are excluded from
     the distinctive-token picks since they don't correlate with
     Prelinger title tokens.
  2. 150 MB per-rendition size cap. archive.org routinely hosts
     multi-hundred-megabyte h.264 masters and one 2 GB Prelinger item
     poisons corpus build wall-time and disk. Within a format bucket
     we now pick the largest rendition under the cap; if nothing fits
     we fall through to the next format rather than dropping the item.
  3. 180 s default max-duration ceiling when the caller hasn't set one
     — archive.org is the only source that routinely hosts feature-
     length material and a naive fan-out pulls them into corpora that
     only ever want a few seconds per clip.

stock_sources/wikimedia.py
  Parallel 3-strategy cascade (full → top2_or → single_best). Motivation:
  Commons CirrusSearch also defaults to AND across multi-word queries;
  our first P2 diagnostic pass returned 0 video results for 10/10
  queries. Same stop-word / source-hint / year-token stripping as
  archive_org so the two adapters stay symmetric.

test_stock_source_adapters.py
  Rewrote the wikimedia query-builder tests against the new cascade
  API. Added coverage for multi-word fallback + source-hint/year
  stripping.

video_compose.py
  Two small fixes for the Remotion renderer on Windows:
  1. Resolve output_path to absolute before invoking the CLI so the
     binary can write wherever the caller asked regardless of cwd.
  2. Pass cwd=composer_dir to run_command so npx can find the local
     Remotion binary under node_modules/.bin. Without this, Windows
     npx returns "could not determine executable to run" because it
     resolves .bin relative to the process cwd rather than the
     script's parent.

.gitignore

@@ -24,6 +24,13 @@ tests/qa/output/
 # Project workspaces (generated assets, renders — all regenerable)
 projects/
 
+# Corpus directories (downloaded stock clips, thumbnails, embeddings —
+# regenerable from source APIs via corpus_builder). Explicit rule even
+# though most corpora live under projects/ since agents sometimes
+# build scratch corpora at the repo root.
+corpus/
+**/corpus/
+
 # User music library (personal royalty-free tracks — not part of repo)
 music_library/
 

tests/tools/test_clip_cache.py

@@ -0,0 +1,437 @@
+"""Tests for the shared clip bytes cache (tools/video/clip_cache.py).
+
+These tests do not touch the network and do not require filelock —
+they exercise the cache's public surface (try_link, ingest, stats)
+plus the LRU eviction and manifest-persistence paths. Cache dirs are
+scoped to pytest's ``tmp_path`` so nothing escapes the test session.
+"""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+
+import pytest
+
+from tools.video.clip_cache import (
+    CacheEntry,
+    ClipCache,
+    _link_or_copy,
+    default_cache_dir,
+    default_max_total_bytes,
+    get_default_cache,
+    reset_default_cache,
+)
+
+
+# ----------------------------------------------------------------------
+# Helpers
+# ----------------------------------------------------------------------
+
+
+def _fake_clip(path: Path, size_bytes: int) -> Path:
+    """Write a fixed-size dummy file and return the path."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    # Use a repeating pattern so hex dumps of the file are readable
+    # but the size is exactly what we asked for.
+    with open(path, "wb") as f:
+        f.write(b"x" * size_bytes)
+    return path
+
+
+def _default_metadata(clip_id: str = "test_001") -> dict:
+    return {
+        "source": "test_source",
+        "source_id": clip_id.split("_", 1)[-1],
+        "source_url": f"https://example.test/{clip_id}",
+        "license": "CC0",
+        "creator": "test_rig",
+        "source_tags": "smoke test metadata",
+    }
+
+
+# ----------------------------------------------------------------------
+# Config resolution
+# ----------------------------------------------------------------------
+
+
+def test_default_cache_dir_uses_env_override(monkeypatch, tmp_path):
+    monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "overridden"))
+    assert default_cache_dir() == tmp_path / "overridden"
+
+
+def test_default_cache_dir_falls_back_to_home(monkeypatch):
+    monkeypatch.delenv("OPENMONTAGE_CACHE_DIR", raising=False)
+    result = default_cache_dir()
+    assert result == Path.home() / ".openmontage" / "clips_cache"
+
+
+def test_default_max_total_bytes_respects_env_override(monkeypatch):
+    monkeypatch.setenv("OPENMONTAGE_CACHE_MAX_GB", "5")
+    assert default_max_total_bytes() == 5 * 1024 * 1024 * 1024
+
+
+def test_default_max_total_bytes_ignores_garbage_override(monkeypatch):
+    monkeypatch.setenv("OPENMONTAGE_CACHE_MAX_GB", "not-a-number")
+    assert default_max_total_bytes() == 20 * 1024 * 1024 * 1024
+
+
+# ----------------------------------------------------------------------
+# CacheEntry dataclass round-trip
+# ----------------------------------------------------------------------
+
+
+def test_cache_entry_round_trip_through_dict():
+    entry = CacheEntry(
+        clip_id="test_001",
+        file_name="test_001.mp4",
+        size_bytes=42,
+        added_at=1000.0,
+        last_access_at=2000.0,
+        source="test_source",
+        source_id="001",
+        source_url="https://example.test/001",
+        license="CC0",
+        creator="rig",
+        source_tags="smoke test",
+    )
+    d = entry.to_dict()
+    restored = CacheEntry.from_dict(d)
+    assert restored == entry
+
+
+def test_cache_entry_from_dict_tolerates_missing_fields():
+    minimal = {"clip_id": "x", "file_name": "x.mp4"}
+    entry = CacheEntry.from_dict(minimal)
+    assert entry.clip_id == "x"
+    assert entry.file_name == "x.mp4"
+    assert entry.size_bytes == 0
+    assert entry.source == ""
+    assert entry.license == ""
+
+
+# ----------------------------------------------------------------------
+# try_link / ingest core flow
+# ----------------------------------------------------------------------
+
+
+def test_miss_on_empty_cache(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    dest = tmp_path / "project" / "clips" / "pexels_1.mp4"
+    assert cache.try_link("pexels_1", dest) is False
+    assert not dest.exists()
+    assert cache.misses == 1
+    assert cache.hits == 0
+
+
+def test_ingest_then_hit_round_trip(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    project = tmp_path / "project1"
+    src = _fake_clip(project / "clips" / "pexels_1.mp4", 5000)
+
+    ok = cache.ingest("pexels_1", src, _default_metadata("pexels_1"))
+    assert ok is True
+
+    # A second project should get a cache hit linking the SAME blob in.
+    project2 = tmp_path / "project2"
+    dest2 = project2 / "clips" / "pexels_1.mp4"
+    assert cache.try_link("pexels_1", dest2) is True
+    assert dest2.exists()
+    assert dest2.stat().st_size == 5000
+    assert cache.hits == 1
+    assert cache.misses == 0
+
+
+def test_cache_hit_is_a_hard_link_not_a_copy_when_possible(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    project = tmp_path / "project1"
+    src = _fake_clip(project / "clips" / "pexels_1.mp4", 2048)
+    cache.ingest("pexels_1", src, _default_metadata("pexels_1"))
+
+    project2 = tmp_path / "project2"
+    dest2 = project2 / "clips" / "pexels_1.mp4"
+    cache.try_link("pexels_1", dest2)
+
+    # Same filesystem → hard link → same inode.
+    cache_blob = cache.cache_dir / "pexels_1.mp4"
+    assert cache_blob.exists()
+    if os.name != "nt" or (
+        cache_blob.stat().st_dev == dest2.stat().st_dev
+    ):
+        # Hard links have identical inode numbers on the same FS.
+        assert cache_blob.stat().st_ino == dest2.stat().st_ino
+        assert cache_blob.stat().st_nlink >= 2
+
+
+def test_ingest_rejects_missing_source(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    missing = tmp_path / "does_not_exist.mp4"
+    assert cache.ingest("pexels_missing", missing, {}) is False
+
+
+def test_ingest_rejects_too_small_file(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    tiny = _fake_clip(tmp_path / "tiny.mp4", 100)  # < 1024-byte threshold
+    assert cache.ingest("pexels_tiny", tiny, {}) is False
+
+
+def test_ingest_same_clip_twice_bumps_last_access(tmp_path):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    src = _fake_clip(tmp_path / "a.mp4", 2000)
+    cache.ingest("pexels_dup", src, _default_metadata("pexels_dup"))
+
+    # Read the initial last_access_at
+    entries = cache._read_manifest()
+    first_access = entries["pexels_dup"].last_access_at
+
+    # Wait a tick, then ingest again with a "fresh" source that
+    # simulates a rebuild. The existing entry should have its
+    # last_access bumped and the blob on disk kept intact.
+    import time
+    time.sleep(0.01)
+    src2 = _fake_clip(tmp_path / "b.mp4", 2000)
+    ok = cache.ingest("pexels_dup", src2, _default_metadata("pexels_dup"))
+    assert ok is True
+
+    entries2 = cache._read_manifest()
+    assert entries2["pexels_dup"].last_access_at > first_access
+    # And the cache still only has one entry (the rebuild replaced itself).
+    assert len(entries2) == 1
+
+
+def test_manifest_drift_falls_back_to_miss(tmp_path):
+    """If the manifest has a row but the blob file is gone, report
+    a miss and prune the stale entry."""
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    src = _fake_clip(tmp_path / "real.mp4", 2000)
+    cache.ingest("pexels_drift", src, _default_metadata("pexels_drift"))
+
+    # Manually delete the blob to simulate filesystem drift.
+    blob_path = cache.cache_dir / "pexels_drift.mp4"
+    assert blob_path.exists()
+    blob_path.unlink()
+
+    # try_link should miss AND prune the stale row.
+    dest = tmp_path / "project" / "clips" / "pexels_drift.mp4"
+    assert cache.try_link("pexels_drift", dest) is False
+    entries = cache._read_manifest()
+    assert "pexels_drift" not in entries
+
+
+# ----------------------------------------------------------------------
+# LRU eviction
+# ----------------------------------------------------------------------
+
+
+def test_lru_eviction_evicts_oldest_first(tmp_path):
+    # Cap at 30 KB. Ingest three 10 KB clips, then a fourth — the
+    # oldest should be evicted to make room.
+    cache = ClipCache(
+        cache_dir=tmp_path / "cache",
+        max_total_bytes=30 * 1024,
+    )
+
+    import time
+    for i in range(1, 4):
+        src = _fake_clip(tmp_path / f"clip_{i}.mp4", 10 * 1024)
+        cache.ingest(f"test_{i}", src, _default_metadata(f"test_{i}"))
+        time.sleep(0.01)  # ensure distinct last_access_at
+
+    # Bump access on clip_2 so clip_1 is the LRU victim.
+    dest = tmp_path / "project" / "clip_2.mp4"
+    cache.try_link("test_2", dest)
+
+    # Add a fourth 10 KB clip — should evict clip_1 (the LRU).
+    src4 = _fake_clip(tmp_path / "clip_4.mp4", 10 * 1024)
+    cache.ingest("test_4", src4, _default_metadata("test_4"))
+
+    entries = cache._read_manifest()
+    assert "test_1" not in entries, "LRU victim should have been evicted"
+    assert "test_2" in entries
+    assert "test_3" in entries
+    assert "test_4" in entries
+    assert cache.evictions_count == 1
+    assert cache.bytes_evicted == 10 * 1024
+    # Blob file for the evicted clip should be gone.
+    assert not (cache.cache_dir / "test_1.mp4").exists()
+
+
+def test_eviction_multiple_victims_when_single_clip_is_large(tmp_path):
+    cache = ClipCache(
+        cache_dir=tmp_path / "cache",
+        max_total_bytes=50 * 1024,
+    )
+
+    import time
+    # Five 10 KB clips fit exactly at the cap.
+    for i in range(1, 6):
+        src = _fake_clip(tmp_path / f"small_{i}.mp4", 10 * 1024)
+        cache.ingest(f"small_{i}", src, _default_metadata(f"small_{i}"))
+        time.sleep(0.005)
+
+    # A 30 KB incoming clip forces eviction of the 3 oldest.
+    big = _fake_clip(tmp_path / "big.mp4", 30 * 1024)
+    cache.ingest("big_1", big, _default_metadata("big_1"))
+
+    entries = cache._read_manifest()
+    # big_1 + small_4 + small_5 survive → 3 entries.
+    assert "big_1" in entries
+    assert "small_4" in entries
+    assert "small_5" in entries
+    assert "small_1" not in entries
+    assert "small_2" not in entries
+    assert "small_3" not in entries
+    assert cache.evictions_count == 3
+
+
+# ----------------------------------------------------------------------
+# Manifest persistence across ClipCache instances
+# ----------------------------------------------------------------------
+
+
+def test_manifest_survives_process_boundary(tmp_path):
+    """A second ClipCache instance pointing at the same dir should see
+    entries written by the first instance. This simulates the common
+    multi-process scenario where two corpus_builder runs share the
+    cache via filesystem."""
+    cache_dir = tmp_path / "cache"
+    src = _fake_clip(tmp_path / "clip.mp4", 4000)
+
+    cache_a = ClipCache(cache_dir=cache_dir)
+    assert cache_a.ingest("test_persist", src, _default_metadata("test_persist")) is True
+
+    cache_b = ClipCache(cache_dir=cache_dir)
+    dest = tmp_path / "project" / "clip.mp4"
+    assert cache_b.try_link("test_persist", dest) is True
+    assert dest.stat().st_size == 4000
+
+
+def test_manifest_tolerates_corrupt_lines(tmp_path):
+    cache_dir = tmp_path / "cache"
+    cache = ClipCache(cache_dir=cache_dir)
+
+    # Write a manifest with one good row and two malformed rows.
+    good = CacheEntry(
+        clip_id="good_1",
+        file_name="good_1.mp4",
+        size_bytes=1234,
+        added_at=1000.0,
+        last_access_at=1000.0,
+    )
+    _fake_clip(cache_dir / "good_1.mp4", 1234)
+    manifest_path = cache_dir / ClipCache.MANIFEST_NAME
+    with open(manifest_path, "w", encoding="utf-8") as f:
+        f.write("this is not json\n")
+        f.write(json.dumps(good.to_dict()) + "\n")
+        f.write('{"missing_required_fields": true}\n')
+
+    entries = cache._read_manifest()
+    assert list(entries.keys()) == ["good_1"]
+
+
+# ----------------------------------------------------------------------
+# Cross-drive / link-fail fallback
+# ----------------------------------------------------------------------
+
+
+def test_link_or_copy_falls_back_to_copy_when_link_fails(tmp_path, monkeypatch):
+    src = _fake_clip(tmp_path / "src.mp4", 3000)
+    dst = tmp_path / "other" / "dst.mp4"
+    dst.parent.mkdir(exist_ok=True)
+
+    def broken_link(a, b):
+        raise OSError("simulated cross-device link failure")
+
+    monkeypatch.setattr(os, "link", broken_link)
+    ok = _link_or_copy(src, dst)
+    assert ok is True
+    assert dst.exists()
+    assert dst.read_bytes() == src.read_bytes()
+
+
+def test_link_or_copy_returns_false_when_both_fail(tmp_path, monkeypatch):
+    src = _fake_clip(tmp_path / "src.mp4", 3000)
+    dst = tmp_path / "other" / "dst.mp4"
+    dst.parent.mkdir(exist_ok=True)
+
+    def broken_link(a, b):
+        raise OSError("no link")
+
+    def broken_copy(a, b):
+        raise OSError("no copy")
+
+    monkeypatch.setattr(os, "link", broken_link)
+
+    import shutil
+    monkeypatch.setattr(shutil, "copy2", broken_copy)
+
+    ok = _link_or_copy(src, dst)
+    assert ok is False
+    assert not dst.exists()
+
+
+def test_try_link_falls_back_to_copy_and_still_bumps_access(tmp_path, monkeypatch):
+    cache = ClipCache(cache_dir=tmp_path / "cache")
+    src = _fake_clip(tmp_path / "src.mp4", 3000)
+    cache.ingest("pexels_x", src, _default_metadata("pexels_x"))
+
+    def broken_link(a, b):
+        raise OSError("cross-drive")
+
+    monkeypatch.setattr(os, "link", broken_link)
+
+    dest = tmp_path / "project" / "clips" / "pexels_x.mp4"
+    ok = cache.try_link("pexels_x", dest)
+    assert ok is True
+    assert dest.exists()
+    assert cache.hits == 1
+
+
+# ----------------------------------------------------------------------
+# Stats
+# ----------------------------------------------------------------------
+
+
+def test_stats_reports_cache_state_and_session_counters(tmp_path):
+    cache = ClipCache(
+        cache_dir=tmp_path / "cache",
+        max_total_bytes=10 * 1024 * 1024,
+    )
+    _fake_clip(tmp_path / "a.mp4", 2000)
+    cache.ingest("test_stats_a", tmp_path / "a.mp4", _default_metadata("test_stats_a"))
+
+    dest = tmp_path / "project" / "a.mp4"
+    cache.try_link("test_stats_a", dest)
+    cache.try_link("missing_clip", tmp_path / "project" / "missing.mp4")
+
+    s = cache.stats()
+    assert s["entry_count"] == 1
+    assert s["total_bytes"] == 2000
+    assert s["hits_this_session"] == 1
+    assert s["misses_this_session"] == 1
+    assert s["evictions_this_session"] == 0
+    assert s["max_total_bytes"] == 10 * 1024 * 1024
+    assert "cache_dir" in s
+    assert s["filelock_backend"] in ("filelock", "o_excl_fallback")
+
+
+# ----------------------------------------------------------------------
+# get_default_cache singleton
+# ----------------------------------------------------------------------
+
+
+def test_get_default_cache_honors_env_var(monkeypatch, tmp_path):
+    reset_default_cache()
+    monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "envcache"))
+    cache = get_default_cache()
+    assert Path(cache.cache_dir) == tmp_path / "envcache"
+    reset_default_cache()
+
+
+def test_get_default_cache_returns_same_instance(monkeypatch, tmp_path):
+    reset_default_cache()
+    monkeypatch.setenv("OPENMONTAGE_CACHE_DIR", str(tmp_path / "singleton"))
+    a = get_default_cache()
+    b = get_default_cache()
+    assert a is b
+    reset_default_cache()

tests/tools/test_stock_source_adapters.py

@@ -1,6 +1,10 @@
 from tools.video.stock_sources import all_sources
 from tools.video.stock_sources.unsplash import _build_download_url, _orientation_for_unsplash
-from tools.video.stock_sources.wikimedia import _build_search_query, _kind_from_mime, _meta_value
+from tools.video.stock_sources.wikimedia import (
+    _build_search_queries,
+    _kind_from_mime,
+    _meta_value,
+)
 
 
 def test_stock_source_autodiscovery_includes_new_sources():
@@ -10,9 +14,48 @@ def test_stock_source_autodiscovery_includes_new_sources():
 
 
 def test_wikimedia_search_query_respects_kind():
-    assert _build_search_query("rain city", "video").startswith("filetype:video")
-    assert _build_search_query("rain city", "image").startswith("filetype:image")
-    assert _build_search_query("rain city", "any") == "rain city"
+    # The cascade's first ("full") query should always carry the
+    # filetype filter for video/image kinds. "any" drops the prefix.
+    video_cascade = _build_search_queries("rain city", "video")
+    assert video_cascade[0][0] == "full"
+    assert video_cascade[0][1].startswith("filetype:video")
+
+    image_cascade = _build_search_queries("rain city", "image")
+    assert image_cascade[0][0] == "full"
+    assert image_cascade[0][1].startswith("filetype:image")
+
+    any_cascade = _build_search_queries("rain city", "any")
+    assert any_cascade[0][0] == "full"
+    assert any_cascade[0][1] == "rain city"
+
+
+def test_wikimedia_cascade_falls_back_on_multi_word():
+    # Multi-word query should produce a 3-stage cascade: full, top2_or,
+    # single_best. Tokens are picked by length, so "television" beats
+    # "family" and "watching".
+    cascade = _build_search_queries(
+        "1950s family watching television", "video"
+    )
+    labels = [label for label, _ in cascade]
+    assert labels == ["full", "top2_or", "single_best"]
+    assert cascade[1][1] == "filetype:video television watching"
+    assert cascade[2][1] == "filetype:video television"
+
+
+def test_wikimedia_cascade_strips_source_hints_and_years():
+    # "prelinger" is a source hint (redundant on Commons) and "1955" is
+    # a year — both are excluded from distinctive-token picks.
+    cascade = _build_search_queries(
+        "Prelinger 1955 housewife kitchen", "video"
+    )
+    # Full query keeps the source hint + year (first attempt is strict).
+    assert cascade[0][1] == "filetype:video Prelinger 1955 housewife kitchen"
+    # Distinctive picks do NOT include prelinger or 1955.
+    joined = " ".join(sq for _, sq in cascade[1:])
+    assert "housewife" in joined
+    assert "kitchen" in joined
+    assert "prelinger" not in joined.lower()
+    assert "1955" not in joined
 
 
 def test_wikimedia_kind_and_metadata_helpers():

tools/video/clip_cache.py

@@ -0,0 +1,572 @@
+"""Shared clip bytes cache for the corpus builder.
+
+Phase 1 of the shared-corpus architecture: a process-safe, LRU-evicted
+cache of downloaded clip files at ``~/.openmontage/clips_cache/``.
+When the corpus builder decides to fetch a candidate, it first asks
+this cache whether the bytes are already on disk from a previous
+project run. If yes, the cache hard-links (or copies on cross-drive)
+the existing blob into the caller's corpus directory and skips the
+network fetch entirely. If no, the caller downloads as usual and then
+``ingest()``s the fresh bytes so the next project benefits.
+
+Design decisions
+----------------
+
+- **Per-project corpus index stays authoritative.** This cache holds
+  *bytes only*. The ``index.jsonl`` / embeddings of a project corpus
+  are scoped to that project's brief, so retrieval quality is not
+  polluted by clips from unrelated topics. The cache only eliminates
+  redundant downloads and disk copies of the underlying mp4/jpg files.
+
+- **Manifest is JSONL, rewritten atomically.** One entry per line with
+  provenance (source, license, creator) and LRU metadata (added_at,
+  last_access_at). Mutations rewrite the whole file via
+  ``os.replace`` so readers always see a consistent snapshot and
+  crash-mid-write only loses the in-flight mutation, not the whole
+  manifest.
+
+- **File locking on every mutation.** Concurrent corpus_builder runs
+  against the same cache serialize on ``cache_manifest.lock``. Uses
+  ``filelock`` when available (it is, per pip), else a naive exclusive
+  create-file fallback with a polling retry. 60s timeout.
+
+- **Hard links first, copies as fallback.** On the same filesystem,
+  hard-linking the cache blob into the caller's dest dir is free on
+  disk and instant on wall-time. Cross-drive (Windows C:→D:) falls
+  back to ``shutil.copy2`` automatically and logs nothing — the
+  caller doesn't need to care either way.
+
+- **LRU eviction at cap.** Default cap is 20 GB, overridable via
+  ``OPENMONTAGE_CACHE_MAX_GB``. When ``ingest()`` would push total
+  bytes above the cap, the cache evicts least-recently-accessed
+  entries until there's room. Evictions unlink the blob file and drop
+  the manifest row. In-flight entries (currently being ingested) are
+  protected by the lock.
+
+- **Transparent to the agent.** No skill-level changes. The agent
+  keeps calling ``corpus_builder.execute(...)`` the way it always has.
+  The cache tells its story via counters in the ``stats()`` payload
+  that corpus_builder bubbles up into its return value.
+
+Non-goals (intentional for Phase 1)
+-----------------------------------
+- **No embedding cache.** CLIP vectors are still computed per-project.
+  That's Phase 2 and is where the real wall-time wins compound.
+- **No query-result cache.** Source-API calls still run every time.
+  That's Phase 3 and has the highest staleness risk.
+- **No cross-machine sync.** Cache lives on one filesystem; there is
+  no S3 / Dropbox / rsync story. A future phase can add that.
+"""
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import tempfile
+import time
+from contextlib import contextmanager
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from typing import Any, Iterator, Optional
+
+try:
+    import filelock  # type: ignore
+    _HAVE_FILELOCK = True
+except ImportError:
+    _HAVE_FILELOCK = False
+
+
+# Default 20 GB cap. Overridable via OPENMONTAGE_CACHE_MAX_GB.
+_DEFAULT_MAX_TOTAL_BYTES = 20 * 1024 * 1024 * 1024
+
+# Reject ingesting a source file under this size — almost always a
+# failed/empty download that the caller didn't catch.
+_MIN_USABLE_BYTES = 1024
+
+
+# ----------------------------------------------------------------------
+# Config resolution
+# ----------------------------------------------------------------------
+
+
+def default_cache_dir() -> Path:
+    """Resolve the cache directory.
+
+    Honors ``OPENMONTAGE_CACHE_DIR`` if set, else falls back to
+    ``~/.openmontage/clips_cache``. Does not create the directory —
+    that happens in ``ClipCache.__init__`` on first use.
+    """
+    override = os.environ.get("OPENMONTAGE_CACHE_DIR")
+    if override:
+        return Path(override).expanduser()
+    return Path.home() / ".openmontage" / "clips_cache"
+
+
+def default_max_total_bytes() -> int:
+    """Resolve the max-cache-size budget.
+
+    Honors ``OPENMONTAGE_CACHE_MAX_GB`` (float or int) if set, else
+    returns the default 20 GB. Invalid overrides silently fall back to
+    the default rather than crashing — the cache shouldn't bring down
+    a production run over a bad env var.
+    """
+    override = os.environ.get("OPENMONTAGE_CACHE_MAX_GB")
+    if override:
+        try:
+            return int(float(override) * 1024 * 1024 * 1024)
+        except ValueError:
+            pass
+    return _DEFAULT_MAX_TOTAL_BYTES
+
+
+# ----------------------------------------------------------------------
+# Dataclass for one manifest row
+# ----------------------------------------------------------------------
+
+
+@dataclass
+class CacheEntry:
+    """One row in the cache manifest.
+
+    Every field except ``clip_id``/``file_name``/``size_bytes`` is
+    provenance metadata that the agent may want to display or attribute
+    downstream. Stored flat (no nesting) so JSONL lines stay short.
+    """
+
+    clip_id: str
+    file_name: str          # relative to cache_dir, e.g. "pexels_10039002.mp4"
+    size_bytes: int
+    added_at: float
+    last_access_at: float
+    source: str = ""
+    source_id: str = ""
+    source_url: str = ""
+    license: str = ""
+    creator: str = ""
+    source_tags: str = ""
+
+    def to_dict(self) -> dict[str, Any]:
+        return asdict(self)
+
+    @classmethod
+    def from_dict(cls, d: dict[str, Any]) -> "CacheEntry":
+        # Tolerate extra fields from future schema evolution — only
+        # read the ones we know about. Missing fields default.
+        return cls(
+            clip_id=str(d["clip_id"]),
+            file_name=str(d["file_name"]),
+            size_bytes=int(d.get("size_bytes", 0) or 0),
+            added_at=float(d.get("added_at", 0.0) or 0.0),
+            last_access_at=float(
+                d.get("last_access_at", d.get("added_at", 0.0)) or 0.0
+            ),
+            source=str(d.get("source", "") or ""),
+            source_id=str(d.get("source_id", "") or ""),
+            source_url=str(d.get("source_url", "") or ""),
+            license=str(d.get("license", "") or ""),
+            creator=str(d.get("creator", "") or ""),
+            source_tags=str(d.get("source_tags", "") or ""),
+        )
+
+
+# ----------------------------------------------------------------------
+# The cache itself
+# ----------------------------------------------------------------------
+
+
+class ClipCache:
+    """Process-safe, LRU-evicted cache of downloaded clip files.
+
+    Not a singleton — but see ``get_default_cache()`` for the common
+    singleton-at-default-path pattern the corpus builder uses.
+    """
+
+    MANIFEST_NAME = "cache_manifest.jsonl"
+    LOCK_NAME = "cache_manifest.lock"
+
+    def __init__(
+        self,
+        cache_dir: Optional[Path] = None,
+        max_total_bytes: Optional[int] = None,
+    ):
+        self.cache_dir = Path(cache_dir) if cache_dir else default_cache_dir()
+        self.max_total_bytes = (
+            int(max_total_bytes)
+            if max_total_bytes is not None
+            else default_max_total_bytes()
+        )
+        self.cache_dir.mkdir(parents=True, exist_ok=True)
+        self.manifest_path = self.cache_dir / self.MANIFEST_NAME
+        self.lock_path = self.cache_dir / self.LOCK_NAME
+
+        # Per-instance runtime counters. Reset every time a new
+        # ClipCache object is built. For persistent totals, sum
+        # across runs in the calling layer.
+        self.hits = 0
+        self.misses = 0
+        self.evictions_count = 0
+        self.bytes_evicted = 0
+
+    # ------------------------------------------------------------------
+    # Locking
+    # ------------------------------------------------------------------
+
+    @contextmanager
+    def _locked(self, timeout: float = 60.0) -> Iterator[None]:
+        """Acquire an exclusive lock for the duration of the block.
+
+        Prefers ``filelock.FileLock`` (proper cross-platform, timeout
+        support, reentrant). Falls back to a naive O_EXCL create-file
+        lock with polling retry so the cache still works if filelock
+        is somehow unavailable. The fallback is not reentrant — don't
+        nest ``_locked()`` blocks.
+        """
+        if _HAVE_FILELOCK:
+            lock = filelock.FileLock(str(self.lock_path), timeout=timeout)
+            with lock:
+                yield
+            return
+
+        # Fallback: O_EXCL create-file lock.
+        deadline = time.time() + timeout
+        acquired = False
+        while time.time() < deadline:
+            try:
+                fd = os.open(
+                    str(self.lock_path),
+                    os.O_CREAT | os.O_EXCL | os.O_WRONLY,
+                )
+                os.close(fd)
+                acquired = True
+                break
+            except FileExistsError:
+                time.sleep(0.05)
+        if not acquired:
+            raise TimeoutError(
+                f"ClipCache: could not acquire lock at {self.lock_path} "
+                f"after {timeout}s"
+            )
+        try:
+            yield
+        finally:
+            try:
+                os.unlink(self.lock_path)
+            except OSError:
+                pass
+
+    # ------------------------------------------------------------------
+    # Manifest I/O (caller holds the lock)
+    # ------------------------------------------------------------------
+
+    def _read_manifest(self) -> dict[str, CacheEntry]:
+        """Read the manifest file into a dict keyed by clip_id.
+
+        Malformed lines are skipped silently — a single bad row must
+        not poison the whole manifest. Missing file returns an empty
+        dict (first-run case).
+        """
+        entries: dict[str, CacheEntry] = {}
+        if not self.manifest_path.exists():
+            return entries
+        try:
+            with open(self.manifest_path, "r", encoding="utf-8") as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    try:
+                        d = json.loads(line)
+                        entry = CacheEntry.from_dict(d)
+                        entries[entry.clip_id] = entry
+                    except (json.JSONDecodeError, KeyError, TypeError, ValueError):
+                        continue
+        except OSError:
+            # Manifest disappeared between exists() and open() —
+            # treat as empty. The lock prevents this under normal
+            # operation but filesystems can surprise us.
+            return {}
+        return entries
+
+    def _write_manifest(self, entries: dict[str, CacheEntry]) -> None:
+        """Rewrite the manifest file atomically.
+
+        Writes to a sibling tmpfile and uses ``os.replace`` which is
+        atomic on both POSIX and Windows. A crash between write and
+        replace leaves the old manifest intact.
+        """
+        tmp_fd, tmp_name = tempfile.mkstemp(
+            prefix="cache_manifest.", suffix=".tmp", dir=str(self.cache_dir)
+        )
+        try:
+            with os.fdopen(tmp_fd, "w", encoding="utf-8") as f:
+                for entry in entries.values():
+                    f.write(json.dumps(entry.to_dict(), ensure_ascii=False) + "\n")
+            os.replace(tmp_name, self.manifest_path)
+        except Exception:
+            # Best-effort cleanup of the tmpfile if replace failed
+            try:
+                if os.path.exists(tmp_name):
+                    os.unlink(tmp_name)
+            except OSError:
+                pass
+            raise
+
+    # ------------------------------------------------------------------
+    # Public API — try_link, ingest, stats
+    # ------------------------------------------------------------------
+
+    def try_link(self, clip_id: str, dest: Path) -> bool:
+        """Hard-link (or copy) a cached clip into ``dest`` if present.
+
+        Returns ``True`` on cache hit, ``False`` on miss. On hit, the
+        entry's ``last_access_at`` is bumped so LRU eviction keeps
+        recently-used clips around.
+
+        On manifest/filesystem drift (entry exists but blob file is
+        gone), the stale entry is pruned and the call reports a miss,
+        so the caller falls back to downloading fresh.
+        """
+        dest = Path(dest)
+        with self._locked():
+            entries = self._read_manifest()
+            entry = entries.get(clip_id)
+            if entry is None:
+                self.misses += 1
+                return False
+
+            blob_path = self.cache_dir / entry.file_name
+            if not blob_path.exists():
+                # Drift — prune and miss.
+                del entries[clip_id]
+                self._write_manifest(entries)
+                self.misses += 1
+                return False
+
+            dest.parent.mkdir(parents=True, exist_ok=True)
+            # Remove any existing file at dest first so the hard link
+            # can be created cleanly. Harmless if dest didn't exist.
+            if dest.exists() or dest.is_symlink():
+                try:
+                    dest.unlink()
+                except OSError:
+                    pass
+
+            if not _link_or_copy(blob_path, dest):
+                # Link and copy both failed — treat as miss so the
+                # caller redownloads. Leave the cache entry alone;
+                # the blob is still valid, we just can't reach dest.
+                self.misses += 1
+                return False
+
+            entry.last_access_at = time.time()
+            entries[clip_id] = entry
+            self._write_manifest(entries)
+            self.hits += 1
+            return True
+
+    def ingest(
+        self,
+        clip_id: str,
+        source_path: Path,
+        metadata: Optional[dict[str, Any]] = None,
+    ) -> bool:
+        """Copy/link a freshly downloaded clip file into the cache.
+
+        ``source_path`` is the file as it already sits in the caller's
+        project directory after a successful download. We do NOT move
+        or mutate it — the caller keeps the file for its own pipeline
+        and the cache holds a second reference via hard link (or copy
+        on cross-drive).
+
+        Returns ``True`` if the clip was added (or was already present
+        and had its access time bumped), ``False`` if ingest failed
+        (missing source, empty file, lock timeout, or link/copy fail).
+        """
+        source_path = Path(source_path)
+        if not source_path.exists():
+            return False
+        try:
+            size_bytes = source_path.stat().st_size
+        except OSError:
+            return False
+        if size_bytes < _MIN_USABLE_BYTES:
+            return False
+
+        metadata = dict(metadata or {})
+
+        with self._locked():
+            entries = self._read_manifest()
+
+            # Already cached → just bump last_access and return.
+            if clip_id in entries and (
+                self.cache_dir / entries[clip_id].file_name
+            ).exists():
+                entries[clip_id].last_access_at = time.time()
+                self._write_manifest(entries)
+                return True
+
+            # Make room.
+            self._evict_to_fit_locked(entries, size_bytes)
+
+            # Name the blob ``{clip_id}{ext}``. Stable and collision-free
+            # as long as clip_ids are unique (they are — {source}_{source_id}).
+            ext = source_path.suffix or ""
+            blob_name = f"{clip_id}{ext}"
+            blob_path = self.cache_dir / blob_name
+
+            # Clean any stale blob at the same path (drift or interrupted
+            # ingest from a previous run).
+            if blob_path.exists():
+                try:
+                    blob_path.unlink()
+                except OSError:
+                    return False
+
+            if not _link_or_copy(source_path, blob_path):
+                return False
+
+            now = time.time()
+            entries[clip_id] = CacheEntry(
+                clip_id=clip_id,
+                file_name=blob_name,
+                size_bytes=size_bytes,
+                added_at=now,
+                last_access_at=now,
+                source=str(metadata.get("source", "") or ""),
+                source_id=str(metadata.get("source_id", "") or ""),
+                source_url=str(metadata.get("source_url", "") or ""),
+                license=str(metadata.get("license", "") or ""),
+                creator=str(metadata.get("creator", "") or ""),
+                source_tags=str(metadata.get("source_tags", "") or ""),
+            )
+            self._write_manifest(entries)
+            return True
+
+    def stats(self) -> dict[str, Any]:
+        """Return a snapshot of cache state plus session counters.
+
+        The persistent fields (``entry_count``, ``total_bytes``) reflect
+        what's on disk right now. The ``*_this_session`` fields reflect
+        only what this ``ClipCache`` instance has observed — they
+        reset when the process exits.
+        """
+        with self._locked():
+            entries = self._read_manifest()
+        total_bytes = sum(e.size_bytes for e in entries.values())
+        return {
+            "cache_dir": str(self.cache_dir),
+            "entry_count": len(entries),
+            "total_bytes": total_bytes,
+            "total_mb": round(total_bytes / (1024 * 1024), 1),
+            "max_total_bytes": self.max_total_bytes,
+            "max_total_gb": round(self.max_total_bytes / (1024 ** 3), 2),
+            "usage_fraction": (
+                round(total_bytes / self.max_total_bytes, 3)
+                if self.max_total_bytes > 0 else 0.0
+            ),
+            "hits_this_session": self.hits,
+            "misses_this_session": self.misses,
+            "evictions_this_session": self.evictions_count,
+            "bytes_evicted_this_session": self.bytes_evicted,
+            "filelock_backend": "filelock" if _HAVE_FILELOCK else "o_excl_fallback",
+        }
+
+    # ------------------------------------------------------------------
+    # LRU eviction (caller holds the lock)
+    # ------------------------------------------------------------------
+
+    def _evict_to_fit_locked(
+        self, entries: dict[str, CacheEntry], needed_bytes: int
+    ) -> None:
+        """Evict least-recently-accessed entries until ``needed_bytes`` fits.
+
+        Mutates ``entries`` in place. Silently skips victims whose
+        blob file has already vanished (drift) so eviction is
+        best-effort and does not block the ingest path.
+        """
+        if needed_bytes <= 0:
+            return
+        current_bytes = sum(e.size_bytes for e in entries.values())
+        if current_bytes + needed_bytes <= self.max_total_bytes:
+            return
+
+        # Oldest first.
+        sorted_victims = sorted(
+            entries.values(), key=lambda e: e.last_access_at
+        )
+        for victim in sorted_victims:
+            if current_bytes + needed_bytes <= self.max_total_bytes:
+                break
+            blob_path = self.cache_dir / victim.file_name
+            unlinked = False
+            try:
+                if blob_path.exists():
+                    blob_path.unlink()
+                unlinked = True
+            except OSError:
+                # Could not delete the blob (in-use on Windows, for
+                # instance). Leave the entry in place and try the next.
+                continue
+            if not unlinked:
+                continue
+            current_bytes -= victim.size_bytes
+            del entries[victim.clip_id]
+            self.evictions_count += 1
+            self.bytes_evicted += victim.size_bytes
+
+
+# ----------------------------------------------------------------------
+# Module-level helpers
+# ----------------------------------------------------------------------
+
+
+def _link_or_copy(src: Path, dst: Path) -> bool:
+    """Hard-link ``src`` to ``dst``; on failure, fall back to ``shutil.copy2``.
+
+    Hard linking is instant and uses zero extra disk on the same
+    filesystem. Cross-drive (Windows C:→D:) and cross-filesystem
+    situations raise ``OSError`` on ``os.link`` and we transparently
+    copy the bytes instead. Returns ``True`` on success, ``False`` if
+    both link and copy failed.
+    """
+    src = Path(src)
+    dst = Path(dst)
+    try:
+        os.link(str(src), str(dst))
+        return True
+    except (OSError, NotImplementedError):
+        pass
+    try:
+        shutil.copy2(str(src), str(dst))
+        return True
+    except (OSError, shutil.SameFileError):
+        return False
+
+
+# ----------------------------------------------------------------------
+# Default-singleton accessor
+# ----------------------------------------------------------------------
+
+
+_DEFAULT_CACHE: Optional[ClipCache] = None
+
+
+def get_default_cache() -> ClipCache:
+    """Return a process-level default ``ClipCache`` at the default path.
+
+    Lazily constructed on first call. Tests that want a pristine cache
+    should instantiate ``ClipCache(cache_dir=tmp_path)`` directly
+    rather than going through this accessor.
+    """
+    global _DEFAULT_CACHE
+    if _DEFAULT_CACHE is None:
+        _DEFAULT_CACHE = ClipCache()
+    return _DEFAULT_CACHE
+
+
+def reset_default_cache() -> None:
+    """Drop the cached singleton so a subsequent ``get_default_cache()``
+    re-reads env vars. Useful for tests that mutate ``OPENMONTAGE_CACHE_DIR``.
+    """
+    global _DEFAULT_CACHE
+    _DEFAULT_CACHE = None

tools/video/corpus_builder.py

@@ -240,6 +240,7 @@ class CorpusBuilder(BaseTool):
         start = time.time()
         try:
             from lib.corpus import Corpus
+            from tools.video.clip_cache import get_default_cache
             from tools.video.stock_sources import (
                 SearchFilters,
                 all_sources,
@@ -301,6 +302,14 @@ class CorpusBuilder(BaseTool):
             corp.load()
             corp.ensure_dirs()
 
+            # Shared clip bytes cache (Phase 1). Hits hard-link blobs
+            # from a previous run's download into this corpus dir so we
+            # don't re-hit the source API or re-download megabytes.
+            # Faults never block the pipeline — a cache miss just means
+            # we download like before.
+            cache = get_default_cache()
+            run_cache_stats = {"hits": 0, "misses": 0, "bytes_saved": 0}
+
             per_source_counts: dict[str, int] = {s.name: 0 for s in sources}
             added_ids: list[str] = []
             errors: list[dict] = []
@@ -354,6 +363,8 @@ class CorpusBuilder(BaseTool):
                                 corp=corp,
                                 query=query,
                                 thumbs_per_video=thumbs_per_video,
+                                cache=cache,
+                                run_cache_stats=run_cache_stats,
                             )
                         except Exception as e:
                             failed += 1
@@ -376,6 +387,11 @@ class CorpusBuilder(BaseTool):
             corp.save()
 
             elapsed = time.time() - start
+            try:
+                cache_snapshot = cache.stats()
+            except Exception as e:
+                cache_snapshot = {"error": f"{type(e).__name__}: {e}"}
+
             return ToolResult(
                 success=True,
                 data={
@@ -391,6 +407,13 @@ class CorpusBuilder(BaseTool):
                     "requested_sources": source_names or [],
                     "resolved_sources": [s.name for s in sources],
                     "source_provider_summary": source_summary(),
+                    # Shared clip bytes cache (Phase 1): per-run
+                    # counters plus a full stats snapshot for the
+                    # agent to display in the production report.
+                    "cache_hits": run_cache_stats["hits"],
+                    "cache_misses": run_cache_stats["misses"],
+                    "cache_bytes_saved": run_cache_stats["bytes_saved"],
+                    "cache_stats": cache_snapshot,
                     "errors": errors[:25],  # cap log noise
                 },
                 cost_usd=0.0,
@@ -414,12 +437,23 @@ class CorpusBuilder(BaseTool):
         corp,
         query: str,
         thumbs_per_video: int,
+        cache,
+        run_cache_stats: dict,
     ):
         """Download → thumb → embed → add one Candidate to the corpus.
 
         Returns the created `ClipRecord` on success, None if the clip
         was rejected (download empty, thumb extraction failed, etc.).
         Raises on unexpected errors (the caller logs them).
+
+        Before downloading, consults the shared clip bytes cache at
+        ``~/.openmontage/clips_cache/``: if the file is already on
+        disk from a previous run (the same clip surfaced for a
+        different project), the cache hard-links it straight into
+        ``local_abs`` and we skip the network fetch entirely. On a
+        miss we download as before and ingest the fresh file so the
+        next run benefits. Cache faults never block the pipeline —
+        they degrade gracefully to normal downloads.
         """
         import cv2
 
@@ -432,18 +466,58 @@ class CorpusBuilder(BaseTool):
         local_rel = Path("clips") / f"{cand.clip_id}{ext}"
         local_abs = corp.corpus_dir / local_rel
 
-        # Download. Any HTTP/IO exception propagates up to the
-        # per-candidate try in execute().
-        src.download(cand, local_abs)
-        if not local_abs.exists() or local_abs.stat().st_size < 1024:
-            # Empty / near-empty file = bad download. Clean up so a
-            # retry doesn't mistake it for success.
+        # Try the shared cache first. A hit links the cached blob
+        # into local_abs (same filesystem → hard link, cross-drive
+        # → copy) and we skip the source fetch entirely.
+        cache_hit = False
+        try:
+            cache_hit = cache.try_link(cand.clip_id, local_abs)
+        except Exception:
+            # Never let a cache fault block the pipeline — fall
+            # through to a fresh download. The cache surfaces faults
+            # via its own stats counters.
+            cache_hit = False
+
+        if cache_hit:
+            run_cache_stats["hits"] += 1
             try:
-                if local_abs.exists():
-                    local_abs.unlink()
+                run_cache_stats["bytes_saved"] += local_abs.stat().st_size
             except OSError:
                 pass
-            return None
+        else:
+            run_cache_stats["misses"] += 1
+            # Download. Any HTTP/IO exception propagates up to the
+            # per-candidate try in execute().
+            src.download(cand, local_abs)
+            if not local_abs.exists() or local_abs.stat().st_size < 1024:
+                # Empty / near-empty file = bad download. Clean up so a
+                # retry doesn't mistake it for success.
+                try:
+                    if local_abs.exists():
+                        local_abs.unlink()
+                except OSError:
+                    pass
+                return None
+
+            # Ingest the fresh file into the shared cache so the
+            # next run can hit it. Swallow ingest failures — the
+            # current run already has the bytes locally, which is
+            # what matters for this build.
+            try:
+                cache.ingest(
+                    cand.clip_id,
+                    local_abs,
+                    metadata={
+                        "source": cand.source,
+                        "source_id": cand.source_id,
+                        "source_url": cand.source_url,
+                        "license": cand.license,
+                        "creator": cand.creator,
+                        "source_tags": cand.source_tags,
+                    },
+                )
+            except Exception:
+                pass
 
         thumb_dir_rel = Path("thumbnails") / cand.clip_id
         thumb_dir_abs = corp.corpus_dir / thumb_dir_rel

tools/video/stock_sources/archive_org.py

@@ -52,7 +52,7 @@ _DEFAULT_COLLECTIONS = ("prelinger", "opensource_movies", "home_movies")
 
 # File formats we accept, in preference order. Archive.org runs every
 # upload through a derivative pipeline so most items have multiple
-# renditions — we want the best mp4.
+# renditions — we want the best mp4 we can afford under the size cap.
 _VIDEO_FORMAT_PRIORITY = (
     "h.264",                  # mp4, usually 480p or 720p
     "MPEG4",                  # older mp4 encoding
@@ -62,6 +62,37 @@ _VIDEO_FORMAT_PRIORITY = (
     "WebM",                   # webm
 )
 
+# Skip any rendition bigger than this at pick time so we never queue
+# a multi-hundred-megabyte download. Archive.org hosts full-length
+# films routinely and even one 2 GB Prelinger master poisons a corpus
+# build on wall-time and disk. Lives in this adapter (not in
+# corpus_builder) because only archive.org's multi-rendition
+# derivative pipeline needs the within-item size shopping this
+# resolves against.
+_MAX_FILE_SIZE_BYTES = 150 * 1024 * 1024  # 150 MB
+
+# Archive.org is the only source that routinely hosts multi-hour
+# material, so we apply a duration ceiling even when the caller did
+# not set `filters.max_duration`. Other adapters can stay loose.
+_DEFAULT_MAX_DURATION_SECONDS = 180.0
+
+# Stop words and very short tokens are dropped from the user query
+# before the query cascade so they don't dilute Solr relevance. Kept
+# small and deliberate — we're only killing words that reliably hurt
+# documentary-style searches.
+_STOP_WORDS = frozenset({
+    "the", "and", "for", "with", "that", "this", "from", "into",
+    "its", "their", "about", "over", "under", "while", "during",
+    "your", "you", "our", "are", "was", "were", "have", "has",
+})
+
+# Tokens that are redundant with the collection filter — including them
+# in the search body just matches item descriptions that happen to say
+# "from the prelinger archives" and not much else. Strip them.
+_SOURCE_HINT_TOKENS = frozenset({
+    "prelinger", "archive", "archives", "stock", "footage",
+})
+
 
 class ArchiveOrgSource:
     """Adapter for public-domain video on archive.org.
@@ -96,6 +127,13 @@ class ArchiveOrgSource:
         adapter returns an empty list for `kind="image"` since
         Archive.org's image collections are a separate ecosystem (see
         `nasa.py` for astronomy imagery instead).
+
+        Query strategy is a 3-stage cascade (`_build_queries`). We walk
+        the cascade in order and return the first strategy whose
+        hydrated candidate list is non-empty. See `_build_queries` for
+        why a cascade is needed — pure OR-join is too noisy, pure
+        phrase-proximity is too strict, no single strategy wins across
+        the documentary query space.
         """
         kind = (filters.kind or "video").lower()
         if kind not in ("video", "any"):
@@ -103,33 +141,43 @@ class ArchiveOrgSource:
 
         import requests  # lazy
 
-        q = self._build_query(query)
-        params = [
-            ("q", q),
-            ("fl[]", "identifier"),
-            ("fl[]", "title"),
-            ("fl[]", "description"),
-            ("fl[]", "creator"),
-            ("fl[]", "date"),
-            ("fl[]", "subject"),
-            ("fl[]", "licenseurl"),
-            ("fl[]", "collection"),
-            ("rows", str(max(1, min(filters.per_page, 50)))),
-            ("page", str(max(1, filters.page))),
-            ("output", "json"),
-        ]
+        for _label, solr_q in self._build_queries(query):
+            params = [
+                ("q", solr_q),
+                ("fl[]", "identifier"),
+                ("fl[]", "title"),
+                ("fl[]", "description"),
+                ("fl[]", "creator"),
+                ("fl[]", "date"),
+                ("fl[]", "subject"),
+                ("fl[]", "licenseurl"),
+                ("fl[]", "collection"),
+                ("rows", str(max(1, min(filters.per_page, 50)))),
+                ("page", str(max(1, filters.page))),
+                ("output", "json"),
+            ]
 
-        r = requests.get(_SEARCH_URL, params=params, timeout=30)
-        r.raise_for_status()
-        data = r.json()
-        docs = (data.get("response") or {}).get("docs", []) or []
+            try:
+                r = requests.get(_SEARCH_URL, params=params, timeout=30)
+                r.raise_for_status()
+                data = r.json()
+            except Exception:
+                # One strategy's network/parse error shouldn't kill the
+                # whole cascade — try the next one.
+                continue
+            docs = (data.get("response") or {}).get("docs", []) or []
+            if not docs:
+                continue
 
-        out: list[Candidate] = []
-        for doc in docs:
-            cand = self._hydrate_candidate(doc, filters)
-            if cand is not None:
-                out.append(cand)
-        return out
+            out: list[Candidate] = []
+            for doc in docs:
+                cand = self._hydrate_candidate(doc, filters)
+                if cand is not None:
+                    out.append(cand)
+            if out:
+                return out
+
+        return []
 
     def download(self, candidate: Candidate, out_path: Path) -> Path:
         """Stream the candidate's file to `out_path`.
@@ -161,20 +209,103 @@ class ArchiveOrgSource:
     # Internals
     # ------------------------------------------------------------------
 
-    def _build_query(self, user_query: str) -> str:
-        """Wrap the user's query with mediatype + collection filters.
+    def _build_queries(self, user_query: str) -> list[tuple[str, str]]:
+        """Build a cascade of Solr queries to try in preference order.
 
-        Archive.org's query language is Solr-style — parentheses and
-        booleans work, and spaces default to AND. We quote the user
-        query so multi-word phrases stay intact.
+        Archive.org's Solr index behaves badly for natural-language
+        documentary queries. The obvious strategies fail in different
+        ways and no single strategy wins across the query space, so we
+        try several and return the first one whose results aren't
+        empty:
+
+        1. **phrase_prox_10** — ``"{phrase}"~10`` proximity match.
+           Finds items whose title/description contains all the query
+           tokens within 10 positions of each other. Best for
+           name-dropped items ("duck and cover drill" ->
+           ``DuckandC1951``). Returns 0 for descriptive queries that
+           don't correspond to a real item title.
+
+        2. **distinctive_and** — top-2 longest non-year tokens joined
+           with AND. Catches items that have the distinctive tokens in
+           the same record without requiring proximity. Finds Prelinger
+           gems like "Frigidaire Imperial Line 1956" when the query is
+           "1955 refrigerator kitchen", or "To New Horizons" (1940)
+           when the query is "suburban optimism suburbia".
+
+        3. **distinctive_or** — top-3 longest tokens OR-joined. Last
+           resort. Noisy — we sacrifice precision for non-empty results
+           and rely on the downstream CLIP retrieval filter to reject
+           junk at clip_search time. Much narrower than OR-joining all
+           tokens because longer words are more discriminative.
+
+        Year tokens ("1950s", "1955") are excluded from the
+        distinctive-token picks because Prelinger titles rarely encode
+        years in text. Including "1950s" in an AND join almost always
+        zeros the result set.
+
+        Source-hint tokens ("prelinger", "archive", "footage") are
+        stripped entirely — they're redundant with the collection
+        filter and match junk descriptions.
+
+        Stop words and short tokens are dropped before everything to
+        keep the proximity phrase meaningful.
         """
         coll = " OR ".join(f"collection:{c}" for c in _DEFAULT_COLLECTIONS)
         user = user_query.strip()
         if not user:
-            return f"mediatype:movies AND ({coll})"
-        # Quote the user query as a phrase AND a loose-term search so
-        # we get both precise matches and relevance-ranked hits.
-        return f'mediatype:movies AND ({coll}) AND ({user})'
+            return [("default", f"mediatype:movies AND ({coll})")]
+
+        tokens = [
+            t for t in re.split(r"\s+", user)
+            if len(t) >= 3
+            and t.lower() not in _STOP_WORDS
+            and t.lower() not in _SOURCE_HINT_TOKENS
+        ]
+        if not tokens:
+            # Nothing meaningful survived filtering — fall back to a
+            # quoted phrase search so we don't ship an empty query.
+            return [(
+                "quoted_fallback",
+                f'mediatype:movies AND ({coll}) AND ("{user}")',
+            )]
+
+        queries: list[tuple[str, str]] = []
+
+        # Strategy 1: phrase proximity ~10. Uses the stripped token
+        # sequence (not the raw query) so stop words don't pollute the
+        # phrase match.
+        clean_phrase = " ".join(tokens)
+        queries.append((
+            "phrase_prox_10",
+            f'mediatype:movies AND ({coll}) AND ("{clean_phrase}"~10)',
+        ))
+
+        # Strategy 2: top-2 longest non-year tokens AND-joined.
+        non_year = [t for t in tokens if not _looks_like_year(t)]
+        if len(non_year) >= 2:
+            distinctive = sorted(non_year, key=lambda t: -len(t))[:2]
+            and_q = " AND ".join(distinctive)
+            queries.append((
+                "distinctive_and",
+                f"mediatype:movies AND ({coll}) AND ({and_q})",
+            ))
+        elif len(non_year) == 1:
+            # Single non-year token — wrap in a simple term query.
+            queries.append((
+                "single_term",
+                f"mediatype:movies AND ({coll}) AND ({non_year[0]})",
+            ))
+
+        # Strategy 3: top-3 longest tokens OR-joined. Last-resort
+        # fallback that accepts noise in exchange for non-empty results.
+        top_tokens = sorted(tokens, key=lambda t: -len(t))[:3]
+        or_q = " OR ".join(top_tokens)
+        queries.append((
+            "distinctive_or",
+            f"mediatype:movies AND ({coll}) AND ({or_q})",
+        ))
+
+        return queries
 
     def _hydrate_candidate(
         self, doc: dict, filters: SearchFilters
@@ -208,12 +339,26 @@ class ArchiveOrgSource:
             return None
 
         duration = _parse_length(picked.get("length"))
-        if filters.min_duration is not None and duration < filters.min_duration:
+
+        # Apply a default max-duration ceiling if the caller didn't set
+        # one. Archive.org is the only source that routinely hosts
+        # multi-hour material, and without a default cap a naive
+        # fan-out pulls feature-length items into a corpus that only
+        # ever wants a few seconds per clip.
+        effective_max_duration = filters.max_duration
+        if effective_max_duration is None:
+            effective_max_duration = _DEFAULT_MAX_DURATION_SECONDS
+
+        if (
+            filters.min_duration is not None
+            and duration
+            and duration < filters.min_duration
+        ):
             return None
-        if filters.max_duration is not None and 0 < duration < filters.max_duration:
-            # 0 means "unknown" — keep those, reject only known-too-long
-            pass
-        if filters.max_duration is not None and duration > filters.max_duration:
+        # duration == 0 means "unknown" — pass through and let the
+        # corpus builder's post-download ffprobe decide. Known-too-long
+        # items are rejected here before the download queue.
+        if duration and duration > effective_max_duration:
             return None
 
         width = _safe_int(picked.get("width"))
@@ -271,12 +416,31 @@ class ArchiveOrgSource:
 # ----------------------------------------------------------------------
 
 
+def _looks_like_year(token: str) -> bool:
+    """True if token is a bare year or year-with-decade-suffix.
+
+    Examples: ``"1950"``, ``"1950s"``, ``"2026"``. Used by
+    ``_build_queries`` to exclude year tokens from the distinctive-
+    token picks, since Prelinger item titles rarely carry the year as
+    a searchable token.
+    """
+    bare = token.rstrip("s")
+    return bare.isdigit() and len(bare) == 4
+
+
 def _pick_video_file(files: list[dict]) -> Optional[dict]:
     """Pick the best playable video file from an Archive.org files list.
 
     Preference order is defined by `_VIDEO_FORMAT_PRIORITY`. Within a
-    format, we prefer the largest file (size as a quality proxy), and
-    reject obvious thumbnails / derivative animations.
+    format we first drop renditions that exceed `_MAX_FILE_SIZE_BYTES`,
+    then pick the largest of the survivors — size is a decent bitrate
+    proxy, so biggest-under-budget = best quality we can afford.
+
+    If no rendition in the preferred format fits the budget we fall
+    through to the next format, so a 2 GB h.264 master gracefully
+    degrades to a 40 MB 512Kb MPEG4 derivative rather than having the
+    whole item dropped from the corpus. Thumbnails and preview GIFs
+    are skipped regardless of format.
     """
     if not files:
         return None
@@ -297,8 +461,18 @@ def _pick_video_file(files: list[dict]) -> Optional[dict]:
         bucket = by_format.get(fmt)
         if not bucket:
             continue
-        bucket.sort(key=lambda f: _safe_int(f.get("size")), reverse=True)
-        return bucket[0]
+        affordable = [
+            f for f in bucket
+            if 0 < _safe_int(f.get("size")) <= _MAX_FILE_SIZE_BYTES
+        ]
+        if not affordable:
+            # Every rendition in this format is either too big or has
+            # no reported size. Fall through to the next format.
+            continue
+        affordable.sort(
+            key=lambda f: _safe_int(f.get("size")), reverse=True
+        )
+        return affordable[0]
     return None
 
 

tools/video/stock_sources/wikimedia.py

@@ -20,6 +20,23 @@ _USER_AGENT = "OpenMontageBot/0.1 (https://github.com/calesthio/OpenMontage)"
 _COMMONS_LICENSE = "Wikimedia Commons (verify per-file license)"
 _HTML_TAG_RE = re.compile(r"<[^>]+>")
 
+# Stop words stripped from multi-term queries before the cascade runs.
+# Commons' CirrusSearch defaults to AND semantics across multi-word
+# queries, so each extra common token shrinks the result set fast.
+_STOP_WORDS = frozenset({
+    "the", "and", "for", "with", "that", "this", "from", "into",
+    "its", "their", "about", "over", "under", "while", "during",
+    "your", "you", "our", "are", "was", "were", "have", "has",
+})
+
+# Tokens that refer to other stock archives — useless on Commons and
+# will poison the cascade if they end up in top2_or because Commons
+# file names don't reference Prelinger or other archives. Keeps the
+# cascade parallel to ``archive_org.py``'s own source-hint stripping.
+_SOURCE_HINT_TOKENS = frozenset({
+    "prelinger", "archive", "archives", "stock", "footage",
+})
+
 
 class WikimediaSource:
     """Adapter for Wikimedia Commons media search."""
@@ -37,39 +54,59 @@ class WikimediaSource:
         return True
 
     def search(self, query: str, filters: SearchFilters) -> list[Candidate]:
+        """Search Commons via CirrusSearch, cascading from precise to broad.
+
+        Commons' search defaults to AND across multi-word queries, so
+        our first diagnostic pass against the P2 query set returned 0
+        video results for 10/10 queries — every query was too specific
+        to intersect Commons' relatively sparse video holdings.
+
+        The cascade (see ``_build_search_queries``) tries strict first,
+        then narrows to 2 distinctive tokens, then to 1 — returning the
+        first non-empty video result set.
+        """
         import requests  # lazy
 
-        params = {
-            "action": "query",
-            "format": "json",
-            "generator": "search",
-            "gsrsearch": _build_search_query(query, filters.kind),
-            "gsrnamespace": 6,
-            "gsrlimit": max(1, min(filters.per_page, 50)),
-            "gsroffset": max(0, (max(filters.page, 1) - 1) * max(1, min(filters.per_page, 50))),
-            "prop": "imageinfo|info",
-            "iiprop": "url|size|mime|extmetadata|mediatype",
-            "iiurlwidth": 640,
-            "inprop": "url",
-        }
+        for _label, search_text in _build_search_queries(query, filters.kind):
+            params = {
+                "action": "query",
+                "format": "json",
+                "generator": "search",
+                "gsrsearch": search_text,
+                "gsrnamespace": 6,
+                "gsrlimit": max(1, min(filters.per_page, 50)),
+                "gsroffset": max(0, (max(filters.page, 1) - 1) * max(1, min(filters.per_page, 50))),
+                "prop": "imageinfo|info",
+                "iiprop": "url|size|mime|extmetadata|mediatype",
+                "iiurlwidth": 640,
+                "inprop": "url",
+            }
 
-        r = requests.get(
-            _API_URL,
-            params=params,
-            headers={"User-Agent": _USER_AGENT},
-            timeout=30,
-        )
-        r.raise_for_status()
-        data = r.json()
-        pages = list(((data.get("query") or {}).get("pages") or {}).values())
-        pages.sort(key=lambda page: int(page.get("index", 0)))
+            try:
+                r = requests.get(
+                    _API_URL,
+                    params=params,
+                    headers={"User-Agent": _USER_AGENT},
+                    timeout=30,
+                )
+                r.raise_for_status()
+                data = r.json()
+            except Exception:
+                continue
+            pages = list(((data.get("query") or {}).get("pages") or {}).values())
+            if not pages:
+                continue
+            pages.sort(key=lambda page: int(page.get("index", 0)))
 
-        out: list[Candidate] = []
-        for page in pages:
-            cand = _page_to_candidate(page, filters)
-            if cand is not None:
-                out.append(cand)
-        return out
+            out: list[Candidate] = []
+            for page in pages:
+                cand = _page_to_candidate(page, filters)
+                if cand is not None:
+                    out.append(cand)
+            if out:
+                return out
+
+        return []
 
     def download(self, candidate: Candidate, out_path: Path) -> Path:
         import requests  # lazy
@@ -94,14 +131,65 @@ class WikimediaSource:
         return out_path
 
 
-def _build_search_query(query: str, kind: str) -> str:
+def _build_search_queries(query: str, kind: str) -> list[tuple[str, str]]:
+    """Return a cascade of search queries to try in preference order.
+
+    Commons' CirrusSearch defaults to AND semantics for multi-word
+    queries, so a 4-word descriptive query like
+    "1950s family watching television" intersects to 0 video hits.
+    We walk from specific to loose:
+
+    1. **full** — ``filetype:video <full query>``. Works when Commons
+       has a file whose name/description contains all the tokens
+       (e.g. "atomic bomb test civil defense" finds
+       "Operation Cue 1955").
+    2. **top2_or** — ``filetype:video <token1> <token2>`` using the
+       two longest non-year tokens. AND-combines at the query level
+       but with only 2 terms, it's loose enough to hit most
+       documentary queries.
+    3. **single_best** — ``filetype:video <longest_token>``.
+       Last-resort single-token search. Noisy but non-empty.
+
+    Year tokens are excluded from the distinctive-token picks — they
+    rarely correlate with file name matches on Commons.
+    """
     user_query = query.strip()
-    kind = (kind or "video").lower()
-    if kind == "video":
-        return f"filetype:video {user_query}".strip()
-    if kind == "image":
-        return f"filetype:image {user_query}".strip()
-    return user_query
+    kind_l = (kind or "video").lower()
+
+    prefix = "filetype:video" if kind_l == "video" else (
+        "filetype:image" if kind_l == "image" else ""
+    )
+
+    def _wrap(text: str) -> str:
+        return f"{prefix} {text}".strip() if prefix else text
+
+    if not user_query:
+        return [("default", _wrap(""))]
+
+    tokens = [
+        t for t in user_query.split()
+        if len(t) >= 3
+        and t.lower() not in _STOP_WORDS
+        and t.lower() not in _SOURCE_HINT_TOKENS
+    ]
+    non_year = [t for t in tokens if not _looks_like_year(t)]
+
+    queries: list[tuple[str, str]] = [("full", _wrap(user_query))]
+
+    if len(non_year) >= 2:
+        top2 = sorted(non_year, key=lambda t: -len(t))[:2]
+        queries.append(("top2_or", _wrap(f"{top2[0]} {top2[1]}")))
+
+    if non_year:
+        best = max(non_year, key=len)
+        queries.append(("single_best", _wrap(best)))
+
+    return queries
+
+
+def _looks_like_year(token: str) -> bool:
+    bare = token.rstrip("s")
+    return bare.isdigit() and len(bare) == 4
 
 
 def _page_to_candidate(page: dict[str, Any], filters: SearchFilters) -> Candidate | None:

tools/video/video_compose.py

@@ -973,6 +973,8 @@ class VideoCompose(BaseTool):
 
         output_path = Path(inputs.get("output_path", "renders/remotion_output.mp4"))
         output_path.parent.mkdir(parents=True, exist_ok=True)
+        # Absolutise so the CLI can resolve the output regardless of cwd.
+        output_path = output_path.resolve()
 
         # Deep-copy props so we don't mutate the original
         props = json.loads(json.dumps(composition_data))
@@ -1037,7 +1039,11 @@ class VideoCompose(BaseTool):
                 pass
 
         try:
-            self.run_command(cmd, timeout=600)
+            # Invoke from inside the composer dir so npx can resolve the
+            # local remotion binary via node_modules/.bin. Without this,
+            # Windows npx cannot locate the CLI and returns "could not
+            # determine executable to run".
+            self.run_command(cmd, timeout=600, cwd=composer_dir)
         except Exception as e:
             return ToolResult(success=False, error=f"Remotion render failed: {e}")
         finally: