fix(backend): balanced clustering, retry failed transcriptions, pytest

- Clustering: switch from average-linkage agglomerative to k-means on
normalized embeddings. On the real 740-reel library the old method put
96% of reels into two clusters (404/306) and left singletons; k-means
gives 12 groups of 33-106.
- Worker: a failed transcription now raises so the reel is retried with
backoff; only the final attempt settles for a caption-only extraction.
Previously any Whisper error silently produced a caption-only "done" row.
- Tests: verify_pipeline.py's hand-rolled runner becomes pytest
(backend/test_pipeline.py, monkeypatch fixtures). Drops two
hasattr-only tests; adds clustering balance and transcription retry tests.
Dev dependencies live in requirements-dev.txt.
- Remove batch_import.py, superseded by POST /extract/batch.

Claude-Session: https://claude.ai/code/session_0166W7S7XvHg5P2uyJtP4GiC

batch_import.py

@@ -1,146 +0,0 @@
-"""
-Batch import saved Instagram reels into the Transcriber pipeline.
-
-Reads an Instagram "Download Your Information" export (saved_posts.json),
-filters to /reel/ URLs, and POSTs each to the running backend's /extract/url.
-
-Resumable: progress is tracked in batch_progress.json so re-running skips
-already-done URLs. The backend also caches by URL, so re-processing is cheap
-either way.
-
-Usage (backend must be running, default http://localhost:8000):
-    python batch_import.py path/to/saved_posts.json
-    python batch_import.py saved_posts.json --delay 3 --caption-fallback
-    python batch_import.py saved_posts.json --limit 10          # smoke test
-    python batch_import.py saved_posts.json --retry-failed      # only retry prior failures
-"""
-import argparse
-import json
-import os
-import sys
-import time
-
-import requests
-
-PROGRESS_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "batch_progress.json")
-
-
-from saved_parser import parse_saved_posts
-
-
-def parse_saved(json_path):
-    """Return list of {url, caption, title} for every /reel/ URL in the export."""
-    with open(json_path, encoding="utf-8") as f:
-        data = json.load(f)
-    return parse_saved_posts(data)
-
-
-def load_progress():
-    if os.path.exists(PROGRESS_PATH):
-        with open(PROGRESS_PATH, encoding="utf-8") as f:
-            return json.load(f)
-    return {}
-
-
-def save_progress(progress):
-    with open(PROGRESS_PATH, "w", encoding="utf-8") as f:
-        json.dump(progress, f, indent=2)
-
-
-def post_with_retry(url, payload, retries=3, backoff=5):
-    """POST with retry on 429/5xx. Returns (ok, status_code, detail)."""
-    for attempt in range(retries):
-        try:
-            r = requests.post(url, json=payload, timeout=300)
-            if r.status_code == 200:
-                return True, 200, ""
-            # silent-hook / no audio = 400, permanent for this reel — don't retry
-            if r.status_code == 400:
-                return False, 400, r.json().get("detail", r.text[:200])
-            if r.status_code == 429 or r.status_code >= 500:
-                wait = backoff * (attempt + 1)
-                print(f"    {r.status_code}, retry in {wait}s ({attempt + 1}/{retries})")
-                time.sleep(wait)
-                continue
-            return False, r.status_code, r.text[:200]
-        except requests.RequestException as e:
-            wait = backoff * (attempt + 1)
-            print(f"    network error {e}, retry in {wait}s ({attempt + 1}/{retries})")
-            time.sleep(wait)
-    return False, 0, "exhausted retries"
-
-
-def main():
-    ap = argparse.ArgumentParser()
-    ap.add_argument("json_path", help="Path to saved_posts.json from the IG export")
-    ap.add_argument("--api", default="http://localhost:8000", help="Backend base URL")
-    ap.add_argument("--delay", type=float, default=2.0, help="Seconds between reels")
-    ap.add_argument("--limit", type=int, default=0, help="Process at most N new reels (0 = all)")
-    ap.add_argument("--caption-fallback", action="store_true",
-                    help="On download failure, extract from export caption (DB row loses its URL)")
-    ap.add_argument("--retry-failed", action="store_true",
-                    help="Only re-attempt URLs marked failed in progress file")
-    args = ap.parse_args()
-
-    items = parse_saved(args.json_path)
-    progress = load_progress()
-    print(f"Found {len(items)} reels in export. Progress file has {len(progress)} records.\n")
-
-    queue = []
-    for it in items:
-        status = progress.get(it["url"], {}).get("status")
-        if args.retry_failed:
-            if status == "failed":
-                queue.append(it)
-        elif status not in ("ok", "caption"):
-            queue.append(it)
-
-    if args.limit:
-        queue = queue[: args.limit]
-    print(f"Queue: {len(queue)} reels to process.\n")
-
-    done = ok = capt = fail = 0
-    for it in queue:
-        done += 1
-        url = it["url"]
-        print(f"[{done}/{len(queue)}] {url}")
-        success, code, detail = post_with_retry(f"{args.api}/extract/url", {"url": url})
-
-        if success:
-            progress[url] = {"status": "ok"}
-            ok += 1
-            print("    ok")
-        elif code == 400:
-            # silent hook / no spoken content — permanent
-            progress[url] = {"status": "failed", "code": 400, "detail": detail}
-            fail += 1
-            print(f"    skip (400): {detail}")
-        elif args.caption_fallback and it["caption"]:
-            s2, c2, d2 = post_with_retry(
-                f"{args.api}/extract/text",
-                {"transcript": "", "caption": it["caption"],
-                 "title": it["title"] or url},
-            )
-            if s2:
-                progress[url] = {"status": "caption", "note": "url not stored in DB"}
-                capt += 1
-                print("    ok (caption fallback)")
-            else:
-                progress[url] = {"status": "failed", "code": code, "detail": detail}
-                fail += 1
-                print(f"    failed: {detail}")
-        else:
-            progress[url] = {"status": "failed", "code": code, "detail": detail}
-            fail += 1
-            print(f"    failed ({code}): {detail}")
-
-        save_progress(progress)
-        time.sleep(args.delay)
-
-    print(f"\nDone. ok={ok} caption={capt} failed={fail}")
-    if fail:
-        print("Re-run with --retry-failed to retry failures (download throttling is often transient).")
-
-
-if __name__ == "__main__":
-    sys.exit(main())

main.py

@@ -257,12 +257,16 @@ def _target_cluster_count(item_count: int) -> int:
 
 
 def _semantic_cluster_labels(embeddings: List) -> List[int]:
-    """Group embeddings with cosine-distance agglomerative clustering."""
+    """Group embeddings with k-means on unit-length vectors (i.e. by cosine similarity).
+
+    Average-linkage agglomerative clustering was used before; on a 740-reel
+    library it put 96% of the reels into two clusters and left singletons."""
     if len(embeddings) <= 1:
         return [0] * len(embeddings)
 
     try:
-        from sklearn.cluster import AgglomerativeClustering
+        from sklearn.cluster import KMeans
+        from sklearn.preprocessing import normalize
     except ImportError as e:
         raise HTTPException(
             status_code=500,
@@ -270,12 +274,9 @@ def _semantic_cluster_labels(embeddings: List) -> List[int]:
                    "Install backend/requirements.txt.",
         ) from e
 
-    clusterer = AgglomerativeClustering(
-        n_clusters=_target_cluster_count(len(embeddings)),
-        metric="cosine",
-        linkage="average",
-    )
-    return [int(label) for label in clusterer.fit_predict(embeddings)]
+    # Fixed seed: recomputing an unchanged library yields the same grouping.
+    clusterer = KMeans(n_clusters=_target_cluster_count(len(embeddings)), n_init=10, random_state=0)
+    return [int(label) for label in clusterer.fit_predict(normalize(embeddings))]
 
 
 def _name_semantic_clusters(items: List[dict], labels: List[int]) -> dict:
@@ -664,13 +665,15 @@ def get_recompute_status():
     """Poll endpoint to check recomputation status."""
     return CLUSTER_JOB
 
-def _run_pipeline(url: str) -> tuple[str, str, str, ReelExtraction]:
+def _run_pipeline(url: str, allow_caption_fallback: bool = True) -> tuple[str, str, str, ReelExtraction]:
     """Download → transcribe → silent-hook guard → LLM extract for one reel.
 
     Owns the temp audio file (always cleaned up). Returns
     (title, raw_transcript, post_caption, extracted). Raises HTTPException(400)
-    on silent-hook reels and HTTPException(500) on download/transcription/LLM
-    failures. Callers handle persistence."""
+    on silent-hook reels and HTTPException(500) on download/LLM failures. A
+    failed transcription falls back to the caption alone unless
+    allow_caption_fallback is False, in which case it raises too. Callers
+    handle persistence."""
     mp3_path = None
     try:
         mp3_path, post_caption, title = download_and_extract_audio(url)
@@ -678,6 +681,8 @@ def _run_pipeline(url: str) -> tuple[str, str, str, ReelExtraction]:
         try:
             raw_transcript = transcribe_audio(mp3_path)
         except Exception as e:
+            if not allow_caption_fallback:
+                raise
             logger.warning(f"Transcription failed: {str(e)}. Proceeding using metadata/caption only.")
         guard_silent_hook(raw_transcript, post_caption)
         extracted = extract_structured_json(raw_transcript, post_caption)
@@ -744,7 +749,11 @@ def process_pending_reel(row: dict) -> None:
         db.mark_failed(reel_id, f"Invalid Instagram Reel URL format: {url}")
         return
     try:
-        title, raw_transcript, post_caption, extracted = _run_pipeline(clean_url)
+        # A failed transcription is retried like any transient error; only the
+        # last attempt settles for a caption-only extraction.
+        is_final_attempt = int(row.get("attempt_count") or 0) >= MAX_WORKER_ATTEMPTS
+        title, raw_transcript, post_caption, extracted = _run_pipeline(
+            clean_url, allow_caption_fallback=is_final_attempt)
         db.update_reel_result(
             reel_id=reel_id,
             title=title,

requirements-dev.txt

@@ -0,0 +1,2 @@
+-r requirements.txt
+pytest>=8

test_pipeline.py

@@ -0,0 +1,500 @@
+"""Pipeline tests. Groq, yt-dlp and Supabase are mocked: no keys or server needed.
+
+Run from the repo root:  python -m pytest backend
+"""
+import io
+import os
+import tempfile
+from datetime import datetime, timezone
+
+os.environ.setdefault("GROQ_API_KEY", "mock_groq_key_for_testing")
+os.environ["ENABLE_WORKER"] = "0"  # never spawn the background worker thread
+
+import httpx
+import pytest
+from fastapi import HTTPException
+from fastapi.testclient import TestClient
+
+import db
+import main
+from main import ReelExtraction
+from saved_parser import normalize_instagram_url, parse_saved_posts
+
+client = TestClient(main.app)
+
+REEL_URL = "https://www.instagram.com/reel/Z/"
+LONG_TRANSCRIPT = " ".join(["word"] * 30)  # clears the 15-word silent-hook guard
+EXTRACTION = ReelExtraction(
+    core_topic="Topic", key_takeaway="Takeaway", action_items=["a"], tools_or_resources=["b"])
+
+
+def fake_insert_reel(url, title, raw_transcript, post_caption, extracted_json):
+    return {"id": "id-1", "url": url, "title": title, "raw_transcript": raw_transcript,
+            "post_caption": post_caption, "extracted_json": extracted_json,
+            "created_at": "2026-06-06T00:00:00Z"}
+
+
+def raiser(exc):
+    def _raise(*args, **kwargs):
+        raise exc
+    return _raise
+
+
+def fake_client(table):
+    """A db.get_client replacement whose every table() is the given fake."""
+    return lambda: type("FakeClient", (), {"table": lambda self, name: table})()
+
+
+class FakeTable:
+    """Stand-in for a Supabase table query: filters with in_() and records inserts."""
+    def __init__(self, rows=None):
+        self.data = rows or []
+        self.inserted = []
+
+    def select(self, *args, **kwargs):
+        return self
+
+    def in_(self, col, vals):
+        return FakeTable([row for row in self.data if row.get(col) in vals])
+
+    def insert(self, rows):
+        self.inserted.extend(rows)
+        return self
+
+    def execute(self):
+        return type("Result", (), {"data": self.data})()
+
+
+@pytest.fixture
+def pipeline(monkeypatch):
+    """Mock download, transcription and extraction so the pipeline runs offline."""
+    monkeypatch.setattr(main, "download_and_extract_audio", lambda url: ("/tmp/does_not_exist.mp3", "cap", "Title"))
+    monkeypatch.setattr(main, "transcribe_audio", lambda path: LONG_TRANSCRIPT)
+    monkeypatch.setattr(main, "extract_structured_json", lambda transcript, caption: EXTRACTION)
+
+
+@pytest.fixture
+def batch_job():
+    """Restore the in-memory batch tracker after the test."""
+    saved = dict(main.BATCH_JOB)
+    yield main.BATCH_JOB
+    main.BATCH_JOB.clear()
+    main.BATCH_JOB.update(saved)
+
+
+# ---------------------------------------------------------------- endpoints
+
+def test_health():
+    response = client.get("/health")
+    assert response.status_code == 200
+    assert response.json()["status"] == "ok"
+
+
+def test_cors_no_credentials():
+    r = client.options("/health", headers={
+        "Origin": "https://reels-transcriber.vercel.app",
+        "Access-Control-Request-Method": "GET",
+    })
+    # With allow_credentials=False, Starlette must not advertise credentials for a wildcard origin.
+    assert r.headers.get("access-control-allow-credentials") != "true", dict(r.headers)
+
+
+def test_extract_text(monkeypatch):
+    monkeypatch.setattr(main, "extract_structured_json", lambda transcript, caption: EXTRACTION)
+    monkeypatch.setattr(main.db, "insert_reel", fake_insert_reel)
+    response = client.post("/extract/text", json={
+        "title": "Test AI Video",
+        "transcript": "Let's build a transcription app with Python and React.",
+        "caption": "#coding #ai #python",
+    })
+    assert response.status_code == 200
+    data = response.json()
+    assert data["title"] == "Test AI Video"
+    assert data["extracted_json"]["core_topic"] == "Topic"
+
+
+def test_extract_url(monkeypatch, pipeline):
+    monkeypatch.setattr(main.db, "get_reel_by_url", lambda url: None)
+    monkeypatch.setattr(main.db, "insert_reel", fake_insert_reel)
+    r = client.post("/extract/url", json={"url": "https://www.instagram.com/reel/REGRESSION1/"})
+    assert r.status_code == 200, r.text
+    assert r.json()["extracted_json"]["core_topic"] == "Topic"
+    assert r.json()["status"] == "done"
+
+
+@pytest.mark.parametrize("status, title, error", [
+    ("pending", None, None),  # title is NULL for rows the iPhone Shortcut inserts over REST
+    ("failed", "Reel (Q)", "Meta blocking request"),
+])
+def test_extract_url_returns_unfinished_row(monkeypatch, status, title, error):
+    """A URL that is already queued or failed comes back as-is instead of a 500."""
+    monkeypatch.setattr(main.db, "get_reel_by_url", lambda url: {
+        "id": "row-q", "url": url, "title": title, "raw_transcript": "", "post_caption": None,
+        "extracted_json": {}, "created_at": "2026-10-10T00:00:00Z", "cluster": "Unclustered",
+        "status": status, "error": error})
+    r = client.post("/extract/url", json={"url": "https://www.instagram.com/reel/QUEUED1/"})
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == status
+    assert r.json()["error"] == error
+
+
+def test_extract_url_rejects_invalid_url():
+    response = client.post("/extract/url", json={"url": "https://google.com"})
+    assert response.status_code == 400
+    assert "Invalid Instagram Reel URL" in response.json()["detail"]
+
+
+def test_url_validation_helpers():
+    assert main.is_valid_instagram_reel("https://www.instagram.com/reel/C7xY9/")
+    assert main.is_valid_instagram_reel("http://instagram.com/reel/abc-123_XYZ/?query=1")
+    assert main.is_valid_instagram_reel("https://instagram.com/reels/abc-123/")
+    assert main.is_valid_instagram_reel("https://instagram.com/p/abc/")
+    assert not main.is_valid_instagram_reel("https://google.com")
+    assert not main.is_valid_instagram_reel("invalid-url")
+    assert not main.is_valid_instagram_reel("")
+
+    assert normalize_instagram_url("https://www.instagram.com/reel/C7xY9/?igsh=123&foo=bar") == "https://www.instagram.com/reel/C7xY9/"
+    assert normalize_instagram_url("https://instagram.com/reels/abc-123?igsi=abc==") == "https://www.instagram.com/reel/abc-123/"
+    assert normalize_instagram_url("https://www.instagram.com/p/ABC123_/?img_index=2") == "https://www.instagram.com/p/ABC123_/"
+
+
+def test_api_auth_token(monkeypatch):
+    monkeypatch.setattr(main, "API_AUTH_TOKEN", "super_secret_test_key")
+    monkeypatch.setattr(main, "extract_structured_json", lambda transcript, caption: EXTRACTION)
+    monkeypatch.setattr(main.db, "insert_reel", fake_insert_reel)
+    payload = {"title": "Test", "transcript": "a", "caption": "b"}
+
+    res = client.post("/extract/text", json=payload)
+    assert res.status_code == 401, res.text
+    assert "Unauthorized" in res.json()["detail"]
+
+    res = client.post("/extract/text", json=payload, headers={"X-API-Key": "wrong"})
+    assert res.status_code == 401, res.text
+
+    res = client.post("/extract/text", json=payload, headers={"X-API-Key": "super_secret_test_key"})
+    assert res.status_code == 200, res.text
+    assert res.json()["title"] == "Test"
+
+
+def test_delete_reel(monkeypatch):
+    calls = []
+    # First call: row exists (True); second: already gone (False -> 404)
+    monkeypatch.setattr(main.db, "delete_reel", lambda rid: (calls.append(rid), len(calls) == 1)[1])
+    r = client.delete("/reels/abc")
+    assert r.status_code == 200, r.text
+    assert r.json()["deleted"] == "abc"
+    assert client.delete("/reels/abc").status_code == 404
+
+
+def test_get_reel_details(monkeypatch):
+    monkeypatch.setattr(main.db, "get_reel_details",
+                        lambda rid: {"raw_transcript": "t", "post_caption": "c"} if rid == "abc" else None)
+    r = client.get("/reels/abc/details")
+    assert r.status_code == 200, r.text
+    assert r.json() == {"raw_transcript": "t", "post_caption": "c"}
+    assert client.get("/reels/missing/details").status_code == 404
+
+
+def test_reels_include_cluster(monkeypatch):
+    monkeypatch.setattr(main.db, "list_reels", lambda limit=20: [{
+        "id": "x", "url": None, "title": "t", "raw_transcript": None,
+        "post_caption": None, "extracted_json": {"core_topic": "c"},
+        "created_at": "2026-06-06T00:00:00Z", "cluster": "Unclustered"}])
+    r = client.get("/reels")
+    assert r.status_code == 200, r.text
+    assert "cluster" in r.json()[0]
+
+
+def test_recompute_clusters(monkeypatch):
+    rows = [
+        {"id": "a", "extracted_json": {"core_topic": "AI email tools"}},
+        {"id": "b", "extracted_json": {"core_topic": "Marathon training"}},
+        {"id": "c", "extracted_json": {}},  # queued/failed reel: nothing to cluster
+    ]
+    applied = []
+    monkeypatch.setattr(main.db, "reels_for_clustering", lambda: rows)
+    monkeypatch.setattr(main.db, "set_cluster",
+                        lambda reel_ids, cluster: applied.extend((rid, cluster) for rid in reel_ids))
+    monkeypatch.setattr(main, "cluster_topics_semantically", lambda items: [
+        {"id": "a", "cluster": "Productivity"}, {"id": "b", "cluster": "Fitness"}])
+
+    r = client.post("/clusters/recompute")
+    assert r.status_code == 200, r.text
+    assert r.json()["status"] == "started"
+
+    job = client.get("/clusters/recompute/status").json()
+    assert job["status"] == "done", job
+    assert job["assigned"] == 2, job
+    assert ("a", "Productivity") in applied
+    assert ("c", None) in applied, "unextracted reel should be reset to Unclustered"
+
+
+def test_cluster_labels_are_balanced():
+    """Well-separated groups must not collapse into one giant cluster plus singletons."""
+    np = pytest.importorskip("numpy")
+    pytest.importorskip("sklearn")
+    rng = np.random.default_rng(0)
+    centers = np.eye(8, 16) * 10  # 8 distinct directions
+    points = [center + rng.normal(scale=0.1, size=16) for center in centers for _ in range(8)]  # 64 points -> 8 clusters
+    labels = main._semantic_cluster_labels(points)
+    sizes = sorted(labels.count(label) for label in set(labels))
+    assert sizes == [8] * 8, sizes
+
+
+# ---------------------------------------------------------------- worker
+
+def test_worker_tick_success(monkeypatch, pipeline):
+    saved = {}
+    monkeypatch.setattr(main.db, "claim_next_pending", lambda: {"id": "row1", "url": REEL_URL})
+    monkeypatch.setattr(main.db, "update_reel_result",
+                        lambda reel_id, title, raw_transcript, post_caption, extracted_json:
+                        saved.update(id=reel_id, title=title))
+    monkeypatch.setattr(main.db, "mark_failed", raiser(AssertionError("should not be marked failed")))
+    assert main.worker_tick() is True
+    assert saved == {"id": "row1", "title": "Title"}
+
+
+def test_worker_tick_failure(monkeypatch):
+    failed = {}
+    monkeypatch.setattr(main.db, "claim_next_pending", lambda: {"id": "row2", "url": "https://www.instagram.com/reel/F/"})
+    monkeypatch.setattr(main, "download_and_extract_audio",
+                        raiser(HTTPException(status_code=400, detail="Meta blocking request")))
+    monkeypatch.setattr(main.db, "mark_failed",
+                        lambda reel_id, error, status="failed": failed.update(id=reel_id, error=str(error), status=status))
+    assert main.worker_tick() is True
+    assert failed["id"] == "row2"
+    assert "Meta blocking" in failed["error"]
+
+
+def test_worker_tick_invalid_url(monkeypatch):
+    failed = {}
+    monkeypatch.setattr(main.db, "claim_next_pending", lambda: {"id": "row_invalid", "url": "https://malicious-ssrf.com"})
+    monkeypatch.setattr(main.db, "mark_failed", lambda reel_id, error: failed.update(id=reel_id, error=str(error)))
+    assert main.worker_tick() is True
+    assert failed["id"] == "row_invalid"
+    assert "Invalid Instagram Reel URL" in failed["error"]
+
+
+def test_worker_tick_empty(monkeypatch):
+    monkeypatch.setattr(main.db, "claim_next_pending", lambda: None)
+    assert main.worker_tick() is False
+
+
+def test_worker_retries_failed_transcription(monkeypatch, pipeline):
+    """A transcription error is retried, not silently replaced by a caption-only extraction."""
+    events = []
+    monkeypatch.setattr(main.db, "claim_next_pending", lambda: {"id": "row1", "url": REEL_URL, "attempt_count": 1})
+    monkeypatch.setattr(main, "transcribe_audio",
+                        raiser(HTTPException(status_code=500, detail="Audio transcription failed: 429")))
+    monkeypatch.setattr(main.db, "schedule_retry", lambda reel_id, error, when: events.append(("retry", reel_id)))
+    monkeypatch.setattr(main.db, "update_reel_result", lambda **kwargs: events.append(("done", kwargs["reel_id"])))
+    monkeypatch.setattr(main.db, "mark_failed", lambda *args, **kwargs: events.append(("failed", args[0])))
+    assert main.worker_tick() is True
+    assert events == [("retry", "row1")]
+
+
+def test_worker_falls_back_to_caption_on_final_attempt(monkeypatch, pipeline):
+    saved = {}
+    monkeypatch.setattr(main.db, "claim_next_pending",
+                        lambda: {"id": "row1", "url": REEL_URL, "attempt_count": main.MAX_WORKER_ATTEMPTS})
+    monkeypatch.setattr(main, "transcribe_audio",
+                        raiser(HTTPException(status_code=500, detail="Audio transcription failed: 413")))
+    monkeypatch.setattr(main.db, "update_reel_result",
+                        lambda **kwargs: saved.update(id=kwargs["reel_id"], transcript=kwargs["raw_transcript"]))
+    assert main.worker_tick() is True
+    assert saved == {"id": "row1", "transcript": ""}
+
+
+def test_run_worker_drain(monkeypatch):
+    """Drain mode stops at the first empty tick and survives a failed stale-claim recovery."""
+    ticks = iter([True, True, False])
+    monkeypatch.setattr(main.db, "recover_stale_processing", raiser(RuntimeError("db offline")))
+    monkeypatch.setattr(main, "worker_tick", lambda: next(ticks))
+    main.run_worker(drain=True)
+    assert next(ticks, None) is None
+
+
+def test_worker_retry_policy(monkeypatch):
+    assert main.is_retryable_worker_error(HTTPException(status_code=429))
+    assert main.is_retryable_worker_error(HTTPException(status_code=503))
+    assert main.is_retryable_worker_error(httpx.ConnectError("offline"))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=400))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=403))
+    assert not main.is_retryable_worker_error(HTTPException(status_code=415))
+    assert 27 <= main.retry_delay_seconds(1, .9) <= 33
+    assert 108 <= main.retry_delay_seconds(2, .9) <= 132
+
+    scheduled = []
+    monkeypatch.setattr(main.db, "schedule_retry", lambda reel_id, error, when: scheduled.append(reel_id))
+    assert main._schedule_worker_retry({"id": "a", "attempt_count": 1}, "temporary")
+    assert main._schedule_worker_retry({"id": "b", "attempt_count": 2}, "temporary")
+    assert not main._schedule_worker_retry({"id": "c", "attempt_count": 3}, "temporary")
+    assert scheduled == ["a", "b"]
+
+
+def test_download_cleanup_on_failure(monkeypatch):
+    reel_id = "test_cleanup_fail_123"
+    leftovers = [os.path.join(tempfile.gettempdir(), f"video_{reel_id}.{ext}") for ext in ("mp4", "part")]
+    for path in leftovers:
+        with open(path, "w") as f:
+            f.write("partial download")
+
+    class FailingYDL:
+        def __init__(self, *args, **kwargs):
+            pass
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *exc):
+            pass
+
+        def extract_info(self, *args, **kwargs):
+            raise Exception("Mock download error")
+
+    monkeypatch.setattr(main.yt_dlp, "YoutubeDL", FailingYDL)
+    try:
+        with pytest.raises(HTTPException) as excinfo:
+            main.download_and_extract_audio(f"https://www.instagram.com/reel/{reel_id}/")
+        assert excinfo.value.status_code == 500
+        assert "Mock download error" in excinfo.value.detail
+        assert not any(os.path.exists(path) for path in leftovers)
+    finally:
+        for path in leftovers:
+            if os.path.exists(path):
+                os.remove(path)
+
+
+# ---------------------------------------------------------------- bulk import
+
+def test_parse_saved_posts():
+    sample = [
+        {"label_values": [
+            {"label": "URL", "value": "https://www.instagram.com/reel/AAA/"},
+            {"label": "Caption", "value": "hello"},
+            {"label": "Title", "value": "T1"}]},
+        {"label_values": [
+            {"label": "URL", "value": "https://www.instagram.com/p/BBB/"},
+            {"label": "Caption", "value": "photo"}]},
+    ]
+    out = parse_saved_posts(sample)
+    assert len(out) == 1, out  # the /p/ photo post is dropped
+    assert out[0] == {"url": "https://www.instagram.com/reel/AAA/", "caption": "hello", "title": "T1"}
+
+
+def test_batch_status_shape():
+    body = client.get("/extract/batch/status").json()
+    for key in ("status", "total", "done", "ok", "failed", "current", "errors"):
+        assert key in body, f"missing {key} in {body}"
+
+
+def test_batch_guard_409(monkeypatch, batch_job):
+    batch_job["status"] = "running"
+    monkeypatch.setattr(main, "update_batch_job_status", lambda: None)
+    files = {"file": ("saved_posts.json", io.BytesIO(b"[]"), "application/json")}
+    assert client.post("/extract/batch", files=files).status_code == 409
+
+
+def test_extract_batch(monkeypatch, batch_job):
+    table = FakeTable([{"url": "https://www.instagram.com/reel/BBB/"}])  # BBB already saved
+    monkeypatch.setattr(main.db, "get_client", fake_client(table))
+    upload = (
+        '[{"label_values": [{"label": "URL", "value": "https://www.instagram.com/reel/AAA/"}]},'
+        ' {"label_values": [{"label": "URL", "value": "https://www.instagram.com/reel/BBB/"}]}]'
+    )
+    files = {"file": ("saved_posts.json", io.BytesIO(upload.encode()), "application/json")}
+
+    r = client.post("/extract/batch", files=files)
+    assert r.status_code == 200, r.text
+    assert r.json() == {"started": True, "total": 2, "enqueued": 1, "existing": 1}
+    assert len(table.inserted) == 1
+    assert table.inserted[0]["url"] == "https://www.instagram.com/reel/AAA/"
+    assert table.inserted[0]["status"] == "pending"
+    assert table.inserted[0]["source"] == "batch"
+
+    status = client.get("/extract/batch/status").json()
+    assert (status["status"], status["total"], status["done"]) == ("running", 2, 1)
+
+    table.data.append({"url": "https://www.instagram.com/reel/AAA/", "status": "done"})  # worker finishes AAA
+    status = client.get("/extract/batch/status").json()
+    assert (status["status"], status["done"], status["ok"]) == ("done", 2, 2)
+
+
+@pytest.mark.parametrize("status", ["cookies_expired", "unsupported_format", "mystery"])
+def test_batch_terminal_statuses(monkeypatch, batch_job, status):
+    table = FakeTable()
+    table.in_ = lambda col, vals: FakeTable([{"url": "u", "status": status, "error": "boom"}])
+    monkeypatch.setattr(main.db, "get_client", fake_client(table))
+    batch_job.update({"status": "running", "urls": ["u"], "finished_at": None})
+
+    main.update_batch_job_status()
+    assert batch_job["status"] == "done"
+    assert batch_job["failed"] == 1
+    if status == "mystery":
+        assert "Unknown queue status: mystery" in batch_job["errors"][0]["detail"]
+
+
+# ---------------------------------------------------------------- data layer
+
+def test_row_to_record_handles_bad_json():
+    assert db.row_to_record({"id": "1", "extracted_json": "{not valid json"})["extracted_json"] == {}
+    assert db.row_to_record({"id": "2", "extracted_json": '{"core_topic": "x"}'})["extracted_json"] == {"core_topic": "x"}
+
+
+def test_chunked_helper():
+    assert list(db._chunked([1, 2, 3, 4, 5], 2)) == [[1, 2], [3, 4], [5]]
+    assert list(db._chunked([], 3)) == []
+    assert list(db._chunked([1, 2], 5)) == [[1, 2]]
+
+
+def test_db_queue_metadata_contract(monkeypatch):
+    class Query:
+        def __init__(self):
+            self.calls = []
+            self.updated = None
+
+        def select(self, *args):
+            self.calls.append(("select", args))
+            return self
+
+        def update(self, payload):
+            self.updated = payload
+            return self
+
+        def eq(self, *args):
+            self.calls.append(("eq", args))
+            return self
+
+        def or_(self, value):
+            self.calls.append(("or", value))
+            return self
+
+        def order(self, *args, **kwargs):
+            return self
+
+        def limit(self, value):
+            return self
+
+        def lt(self, *args):
+            self.calls.append(("lt", args))
+            return self
+
+        def execute(self):
+            rows = [{"id": "r1", "attempt_count": 1}] if self.updated is None else [{"id": "r1"}]
+            return type("Result", (), {"data": rows})()
+
+    query = Query()
+    monkeypatch.setattr(db, "get_client", fake_client(query))
+
+    claimed = db.claim_next_pending()
+    assert claimed["attempt_count"] == 2
+    assert any(kind == "or" and "next_attempt_at.is.null" in value and "next_attempt_at.lte." in value
+               for kind, value in query.calls)
+    assert query.updated["status"] == "processing"
+    assert query.updated["processing_started_at"]
+    assert query.updated["next_attempt_at"] is None
+
+    query.calls.clear()
+    assert db.recover_stale_processing(datetime.now(timezone.utc)) == 1
+    assert ("eq", ("status", "processing")) in query.calls
+    assert any(kind == "lt" and args[0] == "processing_started_at" for kind, args in query.calls)

verify_pipeline.py

@@ -1,745 +0,0 @@
-import os
-import sys
-
-# Ensure backend directory is in path
-sys.path.append(os.path.dirname(os.path.abspath(__file__)))
-
-# Set mock env variables for testing if they are not defined
-if not os.environ.get("GROQ_API_KEY"):
-    os.environ["GROQ_API_KEY"] = "mock_groq_key_for_testing"
-
-# Never spawn the background worker thread during tests
-os.environ["ENABLE_WORKER"] = "0"
-
-try:
-    from fastapi.testclient import TestClient
-except ImportError:
-    print("Installing 'httpx' for TestClient support...")
-    import subprocess
-    subprocess.check_call([sys.executable, "-m", "pip", "install", "httpx"])
-    from fastapi.testclient import TestClient
-
-from main import app, ReelExtraction
-
-client = TestClient(app)
-
-def test_health():
-    print("Testing GET /health...")
-    response = client.get("/health")
-    assert response.status_code == 200
-    assert response.json()["status"] == "ok"
-    print("[OK] Health check passed!")
-
-def test_extract_text_mock():
-    print("Testing POST /extract/text with mock data...")
-    # Monkeypatch the extract_structured_json function in main to prevent external calls
-    import main
-    
-    mock_extracted = ReelExtraction(
-        core_topic="AI Video Automation",
-        key_takeaway="You can automate your video transcription easily.",
-        action_items=["Get a URL", "Extract audio", "Run Whisper"],
-        tools_or_resources=["FastAPI", "Groq", "Supabase"]
-    )
-    
-    original_extract = main.extract_structured_json
-    main.extract_structured_json = lambda t, c: mock_extracted
-    
-    # Mock saving to database to prevent real database call
-    original_save = main.db.insert_reel
-    main.db.insert_reel = lambda url, title, raw_transcript, post_caption, extracted_json: {
-        "id": "123e4567-e89b-12d3-a456-426614174000",
-        "url": url,
-        "title": title,
-        "raw_transcript": raw_transcript,
-        "post_caption": post_caption,
-        "extracted_json": extracted_json,
-        "created_at": "2026-06-05T00:00:00Z"
-    }
-    
-    try:
-        response = client.post("/extract/text", json={
-            "title": "Test AI Video",
-            "transcript": "Let's build a transcription app with Python and React.",
-            "caption": "#coding #ai #python"
-        })
-        assert response.status_code == 200
-        data = response.json()
-        assert data["title"] == "Test AI Video"
-        assert data["extracted_json"]["core_topic"] == "AI Video Automation"
-        assert len(data["extracted_json"]["action_items"]) == 3
-        print("[OK] Text extraction pipeline endpoint passed (mocked)!")
-    finally:
-        # Restore functions
-        main.extract_structured_json = original_extract
-        main.db.insert_reel = original_save
-
-def test_recompute_clusters_mock():
-    print("Testing POST /clusters/recompute (db + llm mocked)...")
-    import main
-    fake_rows = [
-        {"id": "a", "extracted_json": {"core_topic": "AI email tools"}},
-        {"id": "b", "extracted_json": {"core_topic": "Marathon training"}},
-        {"id": "c", "extracted_json": {}},  # queued/failed reel: nothing to cluster
-    ]
-    applied = []
-    orig_for = main.db.reels_for_clustering
-    orig_set = main.db.set_cluster
-    orig_llm = main.cluster_topics_semantically
-    main.db.reels_for_clustering = lambda: fake_rows
-    main.db.set_cluster = lambda reel_ids, cluster: applied.extend((rid, cluster) for rid in reel_ids)
-    main.cluster_topics_semantically = lambda items: [
-        {"id": "a", "cluster": "Productivity"}, {"id": "b", "cluster": "Fitness"}]
-    try:
-        r = client.post("/clusters/recompute")
-        assert r.status_code == 200, r.text
-        assert r.json()["status"] == "started", r.text
-        
-        status_resp = client.get("/clusters/recompute/status")
-        assert status_resp.status_code == 200
-        job = status_resp.json()
-        assert job["status"] == "done", job
-        assert job["assigned"] == 2, job
-        assert ("a", "Productivity") in applied
-        assert ("c", None) in applied, "unextracted reel should be reset to Unclustered"
-        print("[OK] recompute clusters passed!")
-    finally:
-        main.db.reels_for_clustering = orig_for
-        main.db.set_cluster = orig_set
-        main.cluster_topics_semantically = orig_llm
-
-def test_row_to_record_handles_bad_json():
-    print("Testing row_to_record tolerates malformed extracted_json...")
-    import db
-    rec = db.row_to_record({"id": "1", "extracted_json": "{not valid json"})
-    assert rec["extracted_json"] == {}, rec
-    rec2 = db.row_to_record({"id": "2", "extracted_json": '{"core_topic": "x"}'})
-    assert rec2["extracted_json"]["core_topic"] == "x", rec2
-    print("[OK] row_to_record bad-json guard passed!")
-
-
-def test_db_module_surface():
-    print("Testing db.py exposes the expected data-layer functions...")
-    import db
-    for fn in ("get_reel_by_url", "insert_reel", "list_reels",
-               "reels_for_clustering", "set_cluster",
-               "row_to_record"):
-        assert hasattr(db, fn), f"db.{fn} missing"
-    print("[OK] db module surface passed!")
-
-
-def test_db_queue_surface():
-    print("Testing db.py exposes queue functions...")
-    import db
-    for fn in ("claim_next_pending", "update_reel_result", "mark_failed"):
-        assert hasattr(db, fn), f"db.{fn} missing"
-    print("[OK] db queue surface passed!")
-
-
-def test_extract_url_regression():
-    print("Testing POST /extract/url via process_reel_url (mocked)...")
-    import main
-    long_transcript = " ".join(["word"] * 30)  # clears the 15-word silent-hook guard
-    mocked = ReelExtraction(
-        core_topic="Topic", key_takeaway="Takeaway",
-        action_items=["a"], tools_or_resources=["b"])
-    orig_dl = main.download_and_extract_audio
-    orig_tr = main.transcribe_audio
-    orig_ex = main.extract_structured_json
-    orig_save = main.db.insert_reel
-    orig_cache = main.db.get_reel_by_url
-    main.db.get_reel_by_url = lambda url: None
-    main.download_and_extract_audio = lambda url: ("/tmp/does_not_exist.mp3", "cap", "Title")
-    main.transcribe_audio = lambda p: long_transcript
-    main.extract_structured_json = lambda t, c: mocked
-    main.db.insert_reel = lambda url, title, raw_transcript, post_caption, extracted_json: {
-        "id": "id-1", "url": url, "title": title,
-        "raw_transcript": raw_transcript, "post_caption": post_caption,
-        "extracted_json": extracted_json, "created_at": "2026-06-06T00:00:00Z"}
-    try:
-        r = client.post("/extract/url", json={"url": "https://www.instagram.com/reel/REGRESSION1/"})
-        assert r.status_code == 200, r.text
-        assert r.json()["extracted_json"]["core_topic"] == "Topic"
-        print("[OK] extract_url regression passed!")
-    finally:
-        main.download_and_extract_audio = orig_dl
-        main.transcribe_audio = orig_tr
-        main.extract_structured_json = orig_ex
-        main.db.insert_reel = orig_save
-        main.db.get_reel_by_url = orig_cache
-
-
-def test_extract_url_returns_unfinished_row():
-    print("Testing POST /extract/url returns an already-queued or failed row instead of 500...")
-    import main
-    orig = main.db.get_reel_by_url
-    try:
-        # title is None for rows the iPhone Shortcut inserts over REST
-        for status, title, error in (("pending", None, None), ("failed", "Reel (Q)", "Meta blocking request")):
-            main.db.get_reel_by_url = lambda url: {
-                "id": "row-q", "url": url, "title": title, "raw_transcript": "", "post_caption": None,
-                "extracted_json": {}, "created_at": "2026-10-10T00:00:00Z", "cluster": "Unclustered",
-                "status": status, "error": error}
-            r = client.post("/extract/url", json={"url": "https://www.instagram.com/reel/QUEUED1/"})
-            assert r.status_code == 200, r.text
-            assert r.json()["status"] == status, r.json()
-            assert r.json()["error"] == error, r.json()
-        print("[OK] extract_url unfinished-row passthrough passed!")
-    finally:
-        main.db.get_reel_by_url = orig
-
-
-def test_worker_tick_success():
-    print("Testing worker_tick processes a pending reel (mocked)...")
-    import main
-    long_transcript = " ".join(["word"] * 30)
-    mocked = ReelExtraction(core_topic="T", key_takeaway="K",
-                            action_items=["a"], tools_or_resources=["b"])
-    captured = {}
-    orig = (main.db.claim_next_pending, main.download_and_extract_audio,
-            main.transcribe_audio, main.extract_structured_json,
-            main.db.update_reel_result, main.db.mark_failed)
-    main.db.claim_next_pending = lambda: {"id": "row1", "url": "https://www.instagram.com/reel/Z/"}
-    main.download_and_extract_audio = lambda url: ("/tmp/nope.mp3", "cap", "Title")
-    main.transcribe_audio = lambda p: long_transcript
-    main.extract_structured_json = lambda t, c: mocked
-    main.db.update_reel_result = lambda reel_id, title, raw_transcript, post_caption, extracted_json, status="done": captured.update(
-        {"id": reel_id, "status": status, "title": title})
-    main.db.mark_failed = lambda reel_id, error: captured.update({"failed": reel_id})
-    try:
-        did = main.worker_tick()
-        assert did is True
-        assert captured["id"] == "row1", captured
-        assert captured["status"] == "done", captured
-        assert "failed" not in captured
-        print("[OK] worker_tick success passed!")
-    finally:
-        (main.db.claim_next_pending, main.download_and_extract_audio,
-         main.transcribe_audio, main.extract_structured_json,
-         main.db.update_reel_result, main.db.mark_failed) = orig
-
-
-def test_worker_tick_failure():
-    print("Testing worker_tick marks a failed download (mocked)...")
-    import main
-    from fastapi import HTTPException
-    captured = {}
-    orig = (main.db.claim_next_pending, main.download_and_extract_audio, main.db.mark_failed)
-    main.db.claim_next_pending = lambda: {"id": "row2", "url": "https://www.instagram.com/reel/F/"}
-    def _boom(url):
-        raise HTTPException(status_code=400, detail="Meta blocking request")
-    main.download_and_extract_audio = _boom
-    
-    main.db.mark_failed = lambda reel_id, error, status="failed": captured.update({"id": reel_id, "error": str(error), "status": status})
-    try:
-        did = main.worker_tick()
-        assert did is True
-        assert captured["id"] == "row2", captured
-        assert "Meta blocking" in captured["error"]
-        print("[OK] worker_tick failure passed!")
-    finally:
-        (main.db.claim_next_pending, main.download_and_extract_audio, main.db.mark_failed) = orig
-
-
-def test_worker_tick_empty():
-    print("Testing worker_tick no-op on empty queue...")
-    import main
-    orig = main.db.claim_next_pending
-    main.db.claim_next_pending = lambda: None
-    try:
-        assert main.worker_tick() is False
-        print("[OK] worker_tick empty passed!")
-    finally:
-        main.db.claim_next_pending = orig
-
-
-def test_run_worker_drain():
-    print("Testing run_worker drains the queue and survives a failed recovery...")
-    import main
-    ticks = iter([True, True, False])
-    orig = (main.db.recover_stale_processing, main.worker_tick)
-    def _boom(stale_before):
-        raise RuntimeError("db offline")
-    main.db.recover_stale_processing = _boom
-    main.worker_tick = lambda: next(ticks)
-    try:
-        main.run_worker(drain=True)
-        assert next(ticks, None) is None, "drain should stop at the first empty tick"
-        print("[OK] run_worker drain passed!")
-    finally:
-        main.db.recover_stale_processing, main.worker_tick = orig
-
-
-def test_delete_reel():
-    print("Testing DELETE /reels/{id} (db mocked)...")
-    import main
-    calls = []
-    orig = main.db.delete_reel
-    # First call: row exists (returns True); second: missing (False -> 404)
-    main.db.delete_reel = lambda rid: (calls.append(rid), len(calls) == 1)[1]
-    try:
-        r = client.delete("/reels/abc")
-        assert r.status_code == 200, r.text
-        assert r.json()["deleted"] == "abc"
-        assert client.delete("/reels/abc").status_code == 404
-        print("[OK] delete reel passed!")
-    finally:
-        main.db.delete_reel = orig
-
-
-def test_get_reel_details():
-    print("Testing GET /reels/{id}/details (db mocked)...")
-    import main
-    orig = main.db.get_reel_details
-    main.db.get_reel_details = lambda rid: {"raw_transcript": "t", "post_caption": "c"} if rid == "abc" else None
-    try:
-        r = client.get("/reels/abc/details")
-        assert r.status_code == 200, r.text
-        assert r.json()["raw_transcript"] == "t"
-        assert r.json()["post_caption"] == "c"
-        
-        assert client.get("/reels/missing/details").status_code == 404
-        print("[OK] get reel details passed!")
-    finally:
-        main.db.get_reel_details = orig
-
-
-def test_reels_include_cluster():
-    print("Testing GET /reels includes cluster (db mocked)...")
-    import main
-    orig = main.db.list_reels
-    main.db.list_reels = lambda limit=20: [{
-        "id": "x", "url": None, "title": "t", "raw_transcript": None,
-        "post_caption": None, "extracted_json": {"core_topic": "c"},
-        "created_at": "2026-06-06T00:00:00Z", "cluster": "Unclustered"}]
-    try:
-        r = client.get("/reels")
-        assert r.status_code == 200, r.text
-        assert "cluster" in r.json()[0]
-        print("[OK] reels include cluster passed!")
-    finally:
-        main.db.list_reels = orig
-
-
-def test_cors_no_credentials():
-    print("Testing CORS does not advertise credentials with wildcard origin...")
-    r = client.options(
-        "/health",
-        headers={
-            "Origin": "https://reels-transcriber.vercel.app",
-            "Access-Control-Request-Method": "GET",
-        },
-    )
-    # With allow_credentials=False, Starlette must NOT emit this header as true.
-    assert r.headers.get("access-control-allow-credentials") != "true", dict(r.headers)
-    print("[OK] CORS no-credentials passed!")
-
-
-def test_chunked_helper():
-    print("Testing _chunked splits sequences into fixed-size lists...")
-    from db import _chunked
-    assert list(_chunked([1, 2, 3, 4, 5], 2)) == [[1, 2], [3, 4], [5]]
-    assert list(_chunked([], 3)) == []
-    assert list(_chunked([1, 2], 5)) == [[1, 2]]
-    print("[OK] _chunked helper passed!")
-
-
-def test_parse_saved_posts():
-    print("Testing parse_saved_posts filters reels...")
-    from saved_parser import parse_saved_posts
-    sample = [
-        {"label_values": [
-            {"label": "URL", "value": "https://www.instagram.com/reel/AAA/"},
-            {"label": "Caption", "value": "hello"},
-            {"label": "Title", "value": "T1"}]},
-        {"label_values": [
-            {"label": "URL", "value": "https://www.instagram.com/p/BBB/"},
-            {"label": "Caption", "value": "photo"}]},
-    ]
-    out = parse_saved_posts(sample)
-    assert len(out) == 1, out
-    assert out[0]["url"].endswith("/reel/AAA/")
-    assert out[0]["caption"] == "hello"
-    assert out[0]["title"] == "T1"
-    print("[OK] parse_saved_posts passed!")
-
-
-def test_batch_status_initial():
-    print("Testing GET /extract/batch/status shape...")
-    r = client.get("/extract/batch/status")
-    assert r.status_code == 200, r.text
-    body = r.json()
-    for key in ("status", "total", "done", "ok", "failed", "current", "errors"):
-        assert key in body, f"missing {key} in {body}"
-    print("[OK] batch status shape passed!")
-
-
-def test_batch_guard_409():
-    print("Testing POST /extract/batch returns 409 while running...")
-    import main, io
-    prev = dict(main.BATCH_JOB)
-    main.BATCH_JOB["status"] = "running"
-    def fake_update():
-        pass
-    orig_update = main.update_batch_job_status
-    main.update_batch_job_status = fake_update
-    try:
-        files = {"file": ("saved_posts.json", io.BytesIO(b"[]"), "application/json")}
-        r = client.post("/extract/batch", files=files)
-        assert r.status_code == 409, r.text
-        print("[OK] batch 409 guard passed!")
-    finally:
-        main.BATCH_JOB.update(prev)
-        main.update_batch_job_status = orig_update
-
-
-
-def test_extract_batch_supabase():
-    print("Testing POST /extract/batch enqueuing (db mocked)...")
-    import main, io
-    # Mock supabase client and database calls
-    class FakeTable:
-        def __init__(self, data=None):
-            self.data = data or []
-            self.inserted = []
-
-        def select(self, *args, **kwargs):
-            return self
-
-        def in_(self, col, vals):
-            # simulate existing URLs: say BBB already exists, AAA is new
-            matching = [d for d in self.data if d.get(col) in vals]
-            return FakeTable(matching)
-
-        def insert(self, rows):
-            if isinstance(rows, list):
-                self.inserted.extend(rows)
-            else:
-                self.inserted.append(rows)
-            return self
-
-        def execute(self):
-            class Res:
-                def __init__(self, data):
-                    self.data = data
-            return Res(self.data)
-
-    fake_db = [{"url": "https://www.instagram.com/reel/BBB/"}]
-    fake_table = FakeTable(fake_db)
-    
-    orig_client = main.db.get_client
-    main.db.get_client = lambda: type('FakeClient', (), {'table': lambda self, name: fake_table})()
-    
-    sample_json = (
-        '['
-        '  {"label_values": [{"label": "URL", "value": "https://www.instagram.com/reel/AAA/"}]},'
-        '  {"label_values": [{"label": "URL", "value": "https://www.instagram.com/reel/BBB/"}]}'
-        ']'
-    )
-    
-    try:
-        files = {"file": ("saved_posts.json", io.BytesIO(sample_json.encode('utf-8')), "application/json")}
-        r = client.post("/extract/batch", files=files)
-        assert r.status_code == 200, r.text
-        res_json = r.json()
-        assert res_json["total"] == 2
-        assert res_json["enqueued"] == 1
-        assert res_json["existing"] == 1
-        
-        # AAA should have been inserted
-        assert len(fake_table.inserted) == 1
-        assert fake_table.inserted[0]["url"] == "https://www.instagram.com/reel/AAA/"
-        assert fake_table.inserted[0]["status"] == "pending"
-        assert fake_table.inserted[0]["source"] == "batch"
-        
-        # Poll status
-        # status should show running, total=2, done=1, ok=1
-        r_status = client.get("/extract/batch/status")
-        assert r_status.status_code == 200
-        status_data = r_status.json()
-        assert status_data["status"] == "running"
-        assert status_data["total"] == 2
-        assert status_data["done"] == 1
-        
-        # Simulate processing AAA -> done
-        fake_table.data.append({"url": "https://www.instagram.com/reel/AAA/", "status": "done"})
-        r_status2 = client.get("/extract/batch/status")
-        assert r_status2.json()["status"] == "done"
-        assert r_status2.json()["done"] == 2
-        assert r_status2.json()["ok"] == 2
-        
-        print("[OK] extract batch supabase mocked test passed!")
-    finally:
-        main.db.get_client = orig_client
-        main.BATCH_JOB["status"] = "idle"
-        main.BATCH_JOB["urls"] = []
-
-
-def test_url_validation_helper():
-    print("Testing is_valid_instagram_reel and normalize_instagram_url helpers...")
-    from main import is_valid_instagram_reel
-    from saved_parser import normalize_instagram_url
-    assert is_valid_instagram_reel("https://www.instagram.com/reel/C7xY9/") is True
-    assert is_valid_instagram_reel("http://instagram.com/reel/abc-123_XYZ/?query=1") is True
-    assert is_valid_instagram_reel("https://instagram.com/reels/abc-123/") is True
-    assert is_valid_instagram_reel("https://instagram.com/p/abc/") is True
-    assert is_valid_instagram_reel("https://google.com") is False
-    assert is_valid_instagram_reel("invalid-url") is False
-    assert is_valid_instagram_reel("") is False
-
-    assert normalize_instagram_url("https://www.instagram.com/reel/C7xY9/?igsh=123&foo=bar") == "https://www.instagram.com/reel/C7xY9/"
-    assert normalize_instagram_url("https://instagram.com/reels/abc-123?igsi=abc==") == "https://www.instagram.com/reel/abc-123/"
-    assert normalize_instagram_url("https://www.instagram.com/p/ABC123_/?img_index=2") == "https://www.instagram.com/p/ABC123_/"
-    print("[OK] URL validation and normalization helpers passed!")
-
-def test_extract_url_validation_error():
-    print("Testing POST /extract/url blocks invalid URLs...")
-    response = client.post("/extract/url", json={"url": "https://google.com"})
-    assert response.status_code == 400
-    assert "Invalid Instagram Reel URL" in response.json()["detail"]
-    print("[OK] extract_url validation error passed!")
-
-def test_worker_tick_invalid_url():
-    print("Testing worker_tick marks invalid URLs as failed...")
-    import main
-    captured = {}
-    orig = (main.db.claim_next_pending, main.db.mark_failed)
-    main.db.claim_next_pending = lambda: {"id": "row_invalid", "url": "https://malicious-ssrf.com"}
-    main.db.mark_failed = lambda reel_id, error: captured.update({"id": reel_id, "error": str(error)})
-    try:
-        did = main.worker_tick()
-        assert did is True
-        assert captured["id"] == "row_invalid"
-        assert "Invalid Instagram Reel URL" in captured["error"]
-        print("[OK] worker_tick invalid URL failure passed!")
-    finally:
-        main.db.claim_next_pending, main.db.mark_failed = orig
-
-
-def test_download_cleanup_on_failure():
-    print("Testing download_and_extract_audio cleans up files on failure...")
-    import tempfile
-    import os
-    import main
-    
-    # Create fake files in temp dir
-    temp_dir = tempfile.gettempdir()
-    reel_id = "test_cleanup_fail_123"
-    url = f"https://www.instagram.com/reel/{reel_id}/"
-    
-    fake_mp4 = os.path.join(temp_dir, f"video_{reel_id}.mp4")
-    fake_part = os.path.join(temp_dir, f"video_{reel_id}.part")
-    
-    # Write empty files
-    with open(fake_mp4, "w") as f:
-        f.write("mp4")
-    with open(fake_part, "w") as f:
-        f.write("part")
-        
-    assert os.path.exists(fake_mp4) is True
-    assert os.path.exists(fake_part) is True
-    
-    # Mock YoutubeDL to throw an Exception
-    class FakeYDLThrow:
-        def __init__(self, *args, **kwargs):
-            pass
-        def __enter__(self):
-            return self
-        def __exit__(self, exc_type, exc_val, exc_tb):
-            pass
-        def extract_info(self, *args, **kwargs):
-            raise Exception("Mock download error")
-            
-    orig_ydl = main.yt_dlp.YoutubeDL
-    main.yt_dlp.YoutubeDL = FakeYDLThrow
-    
-    try:
-        from fastapi import HTTPException
-        try:
-            main.download_and_extract_audio(url)
-            assert False, "Should have raised HTTPException"
-        except HTTPException as he:
-            assert he.status_code == 500
-            assert "Mock download error" in he.detail
-            
-        # Check that the files were deleted
-        assert os.path.exists(fake_mp4) is False, f"fake_mp4 still exists at {fake_mp4}"
-        assert os.path.exists(fake_part) is False, f"fake_part still exists at {fake_part}"
-        print("[OK] download_and_extract_audio failure cleanup passed!")
-    finally:
-        main.yt_dlp.YoutubeDL = orig_ydl
-        # Clean up files in case they were not deleted to avoid side effects
-        if os.path.exists(fake_mp4):
-            os.remove(fake_mp4)
-        if os.path.exists(fake_part):
-            os.remove(fake_part)
-
-
-def test_api_auth_token_check():
-    print("Testing API Access Gate authentication check (API_AUTH_TOKEN set)...")
-    import main
-    orig_token = main.API_AUTH_TOKEN
-    main.API_AUTH_TOKEN = "super_secret_test_key"
-    try:
-        # Request without header -> 401
-        res = client.post("/extract/text", json={"title": "Test", "transcript": "a", "caption": "b"})
-        assert res.status_code == 401, res.text
-        assert "Unauthorized" in res.json()["detail"]
-
-        # Request with incorrect header -> 401
-        res = client.post("/extract/text", json={"title": "Test", "transcript": "a", "caption": "b"}, headers={"X-API-Key": "wrong"})
-        assert res.status_code == 401, res.text
-
-        # Request with correct header -> mock output to pass
-        mock_extracted = ReelExtraction(
-            core_topic="Topic", key_takeaway="Takeaway", action_items=["a"], tools_or_resources=["b"]
-        )
-        orig_extract = main.extract_structured_json
-        main.extract_structured_json = lambda t, c: mock_extracted
-        orig_save = main.db.insert_reel
-        main.db.insert_reel = lambda url, title, raw_transcript, post_caption, extracted_json: {
-            "id": "1", "title": title, "extracted_json": extracted_json, "created_at": "2026-06-06T00:00:00Z"
-        }
-        try:
-            res = client.post("/extract/text", json={"title": "Test", "transcript": "a", "caption": "b"}, headers={"X-API-Key": "super_secret_test_key"})
-            assert res.status_code == 200, res.text
-            assert res.json()["title"] == "Test"
-        finally:
-            main.extract_structured_json = orig_extract
-            main.db.insert_reel = orig_save
-
-        print("[OK] API Access Gate authentication check passed!")
-    finally:
-        main.API_AUTH_TOKEN = orig_token
-
-
-
-def test_batch_terminal_statuses():
-    print("Testing all terminal and unknown batch statuses...")
-    import main
-    original_client = main.db.get_client
-    previous = dict(main.BATCH_JOB)
-
-    class Query:
-        def select(self, *args): return self
-        def in_(self, column, values): return self
-        def execute(self):
-            return type("Result", (), {"data": [{"url": "u", "status": self.status, "error": "boom"}]})()
-
-    try:
-        for status in ("cookies_expired", "unsupported_format", "mystery"):
-            query = Query()
-            query.status = status
-            main.db.get_client = lambda q=query: type("Client", (), {"table": lambda self, name: q})()
-            main.BATCH_JOB.update({"status": "running", "urls": ["u"], "finished_at": None})
-            main.update_batch_job_status()
-            assert main.BATCH_JOB["status"] == "done"
-            assert main.BATCH_JOB["failed"] == 1
-            if status == "mystery":
-                assert "Unknown queue status: mystery" in main.BATCH_JOB["errors"][0]["detail"]
-    finally:
-        main.db.get_client = original_client
-        main.BATCH_JOB.clear()
-        main.BATCH_JOB.update(previous)
-    print("[OK] terminal batch statuses passed!")
-
-
-def test_worker_retry_policy():
-    print("Testing bounded worker retry policy...")
-    import main, httpx
-    from fastapi import HTTPException
-    assert main.is_retryable_worker_error(HTTPException(status_code=429))
-    assert main.is_retryable_worker_error(HTTPException(status_code=503))
-    assert main.is_retryable_worker_error(httpx.ConnectError("offline"))
-    assert not main.is_retryable_worker_error(HTTPException(status_code=400))
-    assert not main.is_retryable_worker_error(HTTPException(status_code=403))
-    assert not main.is_retryable_worker_error(HTTPException(status_code=415))
-    assert 27 <= main.retry_delay_seconds(1, .9) <= 33
-    assert 108 <= main.retry_delay_seconds(2, .9) <= 132
-    calls = []
-    original = main.db.schedule_retry
-    main.db.schedule_retry = lambda reel_id, error, when: calls.append((reel_id, error, when))
-    try:
-        assert main._schedule_worker_retry({"id": "a", "attempt_count": 1}, "temporary")
-        assert main._schedule_worker_retry({"id": "b", "attempt_count": 2}, "temporary")
-        assert not main._schedule_worker_retry({"id": "c", "attempt_count": 3}, "temporary")
-        assert [call[0] for call in calls] == ["a", "b"]
-    finally:
-        main.db.schedule_retry = original
-    print("[OK] bounded retry policy passed!")
-
-
-def test_db_queue_metadata_contract():
-    print("Testing queue metadata data-layer contract...")
-    import db
-    from datetime import datetime, timezone
-
-    class Result:
-        def __init__(self, data): self.data = data
-    class Query:
-        def __init__(self):
-            self.calls = []
-            self.selected_rows = [{"id": "r1", "attempt_count": 1}]
-            self.updated = None
-        def select(self, *args): self.calls.append(("select", args)); return self
-        def update(self, payload): self.updated = payload; self.calls.append(("update", payload)); return self
-        def eq(self, *args): self.calls.append(("eq", args)); return self
-        def or_(self, value): self.calls.append(("or", value)); return self
-        def order(self, *args, **kwargs): return self
-        def limit(self, value): return self
-        def lt(self, *args): self.calls.append(("lt", args)); return self
-        def execute(self):
-            return Result(self.selected_rows if self.updated is None else [{"id": "r1"}])
-
-    query = Query()
-    original = db.get_client
-    db.get_client = lambda: type("Client", (), {"table": lambda self, name: query})()
-    try:
-        claimed = db.claim_next_pending()
-        assert claimed["attempt_count"] == 2
-        assert any(call[0] == "or" and "next_attempt_at.is.null" in call[1] and "next_attempt_at.lte." in call[1] for call in query.calls)
-        assert query.updated["status"] == "processing"
-        assert query.updated["processing_started_at"]
-        assert query.updated["next_attempt_at"] is None
-        query.calls.clear()
-        assert db.recover_stale_processing(datetime.now(timezone.utc)) == 1
-        assert ("eq", ("status", "processing")) in query.calls
-        assert any(call[0] == "lt" and call[1][0] == "processing_started_at" for call in query.calls)
-    finally:
-        db.get_client = original
-    print("[OK] queue metadata contract passed!")
-
-
-if __name__ == "__main__":
-    print("--- Starting Transcriber Pipeline Test ---")
-    test_health()
-    test_extract_text_mock()
-    test_recompute_clusters_mock()
-    test_row_to_record_handles_bad_json()
-    test_db_module_surface()
-    test_db_queue_surface()
-    test_worker_tick_success()
-    test_worker_tick_failure()
-    test_worker_tick_empty()
-    test_run_worker_drain()
-    test_extract_url_regression()
-    test_extract_url_returns_unfinished_row()
-    test_delete_reel()
-    test_get_reel_details()
-    test_reels_include_cluster()
-    test_cors_no_credentials()
-    test_chunked_helper()
-    test_parse_saved_posts()
-    test_batch_status_initial()
-    test_batch_guard_409()
-    test_extract_batch_supabase()
-    test_url_validation_helper()
-    test_extract_url_validation_error()
-    test_worker_tick_invalid_url()
-    test_download_cleanup_on_failure()
-    test_api_auth_token_check()
-    test_batch_terminal_statuses()
-    test_worker_retry_policy()
-    test_db_queue_metadata_contract()
-    print("--- All tests completed successfully! ---")
-    sys.exit(0)