fix(sources): never pay Indian Kanoon again for a judgment whose case is waiting to publish

pipeline/__main__.py

@@ -197,6 +197,17 @@ async def _go() -> tuple[list[RawDocument], list[dict[str, Any]]]:
     return asyncio.run(_go())
 
 
+def _already_handled(ledger: Ledger) -> Callable[[str], bool]:
+    """Tell a paid source (Indian Kanoon) which documents not to download again.
+
+    That's every document an earlier run settled, and also one whose case is still waiting
+    to publish: its record is carried over from data-staging, so downloading it again buys
+    nothing. While publishing was blocked (2026-10-12..14) we paid for the same 38 judgments
+    three days running. A document whose extraction failed is still retried.
+    """
+    return lambda url: ledger.is_pending(url) or not ledger.should_process(url)
+
+
 def _fetch_article_bodies(docs: list[RawDocument]) -> list[RawDocument]:  # pragma: no cover
     async def _go() -> list[RawDocument]:
         async with PoliteClient() as client:
@@ -1137,9 +1148,8 @@ def run(
         if docs is not None:
             raw_docs, court_prefilter = docs, []
         else:
-            settled = ledger
             raw_docs, court_prefilter = _fetch_documents(
-                run_date, court_only=is_refresh, seen=lambda url: not settled.should_process(url)
+                run_date, court_only=is_refresh, seen=_already_handled(ledger)
             )
         report.source_alerts = source_alerts(court_prefilter)
         for alert in report.source_alerts:

pipeline/tests/test_orchestrator.py

@@ -3060,3 +3060,19 @@ def test_refresh_keeps_cases_still_waiting_on_data_staging(
     )
     cnrs = {r.get("cnr") for r in json.loads((main / "2026" / "TG.json").read_text())}
     assert {"C-EXISTING", "C-BRAND-NEW"} <= cnrs
+
+
+def test_paid_source_skips_settled_and_waiting_documents_but_retries_failures() -> None:
+    """Indian Kanoon must not download a judgment again once it is settled, or while its case
+    is waiting to publish (carried over from data-staging); a failed one is still retried."""
+    from pipeline.ledger import Ledger
+
+    ledger = Ledger()
+    ledger.record("https://ex.invalid/settled", "published", "2026-10-12")
+    ledger.record("https://ex.invalid/waiting", "staged_pending", "2026-10-12")
+    ledger.record("https://ex.invalid/failed-once", "failed", "2026-10-12")
+    seen = orchestrator._already_handled(ledger)
+    assert seen("https://ex.invalid/settled") is True
+    assert seen("https://ex.invalid/waiting") is True
+    assert seen("https://ex.invalid/failed-once") is False
+    assert seen("https://ex.invalid/new") is False