Add canlex/refresh.py --check: corpus staleness detection (Phase 2)

Closes the audit's #1 gap (legislation was refreshed by hand, so Bill C-12
only landed when someone looked). For each of the 41 Justice Laws instruments
in config.SOURCES it fetches just the head of the XML (the consolidation date
is a root attribute, so ~16 KB suffices — no full re-download), reads
lims:current-date, and diffs it against the current_to stored in
data/processed/<code>.json; upstream-newer = drift, and the process exits
non-zero so it can gate a deploy or run on a schedule. Also prints the stored
currency of the seven non-XML sources (D-memos, case law, agreement,
directives, delegation, benefits, tariff) with their refresh commands, and
flags tracked pending bills whose in-force date has passed (seeded with Bill
C-16, in force 2026-07-18, amending the Criminal Code).

py -m canlex.refresh --check [--json]

Verified live: reports all 41 instruments current after the recent refresh.
7 offline unit tests (date extraction, BOM, pending-bill gating); 47 tests pass.
Pure helpers (_extract_current_date, pending_in_force) are unit-tested; the
network fetch is not. Maintainer/CI tool — inert in the deployed server.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

canlex/refresh.py

@@ -0,0 +1,197 @@
+"""Detect when CanLex's ingested sources have gone stale upstream.
+
+    py -m canlex.refresh --check
+
+For every Justice Laws instrument in config.SOURCES, this fetches just the
+opening of the XML (the root <Statute>/<Regulation> element carries the
+consolidation date in its lims:current-date attribute, so a few KB is enough)
+and compares that upstream date against the current_to stored in the matching
+data/processed/<code>.json. Anything where upstream is newer is reported as
+drift, and the process exits non-zero -- so this can gate a deploy or run on a
+schedule. It also prints the stored currency of the non-XML sources (D-memos,
+case law, agreements, directives, delegation, benefits, tariff schedule), which
+have no single machine-readable date, with the command to refresh each; and it
+flags any tracked pending bill whose coming-into-force date has passed.
+
+This is the staleness detection the corpus previously lacked: legislation was
+refreshed by hand, so e.g. Bill C-12 only landed when someone thought to look.
+
+    py -m canlex.refresh --check          # human-readable; exit 1 on drift
+    py -m canlex.refresh --check --json    # machine-readable for CI
+"""
+import datetime
+import json
+import re
+import sys
+import time
+import urllib.request
+
+from .config import SOURCES, PROCESSED_DIR
+
+_UA = "CanLex-refresh/0.1"
+_HEAD_BYTES = 16384      # the root element's attributes sit at the very top
+_CURRENT_DATE = re.compile(rb'current-date="(\d{4}-\d{2}-\d{2})"')
+
+# Aggregate (non-Justice-Laws) sources: (label, processed filename, refresh
+# command, description). These carry no single upstream consolidation date, so
+# the check reports their STORED currency (newest item) and how to refresh them
+# rather than diffing against upstream.
+_NON_XML = [
+    ("memorandum", "dmemos.json", "py -m canlex.dmemo", "CBSA D-Memoranda"),
+    ("caselaw", "caselaw.json", "py -m canlex.caselaw", "Court & tribunal decisions"),
+    ("agreement", "agreements.json", "py -m canlex.agreement", "FB collective agreement"),
+    ("directive", "directives.json", "py -m canlex.directive", "NJC directives"),
+    ("delegation", "delegation.json", "py -m canlex.delegation", "IRPA/IRPR delegation"),
+    ("benefits", "benefits.json", "py -m canlex.benefits", "Benefit-plan booklets"),
+    ("tariff", "tariff_schedule.json", "py -m canlex.tariff_schedule",
+     "Customs Tariff Schedule ch. 98-99"),
+]
+
+# Bills known to be heading into force that affect the corpus. Surfaced once
+# their in-force date has passed, with the codes to re-ingest. (The legislation
+# drift check above will independently flag the affected Act once Justice Laws
+# consolidates the amendment; this list just names the cause.)
+_PENDING_BILLS = [
+    {"bill": "C-16", "name": "Protecting Victims Act",
+     "in_force": "2026-07-18", "affects": ["C-46"],
+     "note": "Criminal Code: coercive control, child-protection reporting "
+             "(coercive-control provisions come into force later)."},
+]
+
+
+def _extract_current_date(raw):
+    """Pull the lims:current-date (YYYY-MM-DD) from the head of a Justice Laws
+    XML document. Pure and offline, for testability. Returns '' if absent."""
+    if raw[:3] == b"\xef\xbb\xbf":
+        raw = raw[3:]
+    m = _CURRENT_DATE.search(raw)
+    return m.group(1).decode("ascii") if m else ""
+
+
+def _local_current_to(code):
+    """The current_to stored in data/processed/<code>.json (first chunk), or ''."""
+    path = PROCESSED_DIR / f"{code}.json"
+    if not path.exists():
+        return None
+    chunks = json.loads(path.read_text(encoding="utf-8"))
+    return chunks[0].get("current_to", "") if chunks else ""
+
+
+def _remote_current_date(url):
+    """Fetch only the head of the XML and extract its consolidation date.
+    Returns (date, error): date is '' on a clean fetch that lacked the
+    attribute, error is a short string when the fetch itself failed."""
+    req = urllib.request.Request(url, headers={"User-Agent": _UA})
+    try:
+        with urllib.request.urlopen(req, timeout=60) as resp:
+            head = resp.read(_HEAD_BYTES)
+        return _extract_current_date(head), ""
+    except Exception as exc:                         # network / HTTP / timeout
+        return "", f"{type(exc).__name__}: {exc}"
+
+
+def check_legislation():
+    """Compare each Justice Laws source's upstream date to the stored corpus.
+    Returns a list of row dicts."""
+    rows = []
+    for code, src in SOURCES.items():
+        local = _local_current_to(code)
+        remote, error = _remote_current_date(src["xml_url"])
+        if error:
+            status = "error"
+        elif local is None:
+            status = "missing-local"      # configured but never ingested
+        elif remote and remote > local:
+            status = "stale"              # upstream newer than our corpus
+        elif not remote:
+            status = "no-date"            # fetched, but no date found (unusual)
+        else:
+            status = "ok"
+        rows.append({"code": code, "short": src["short"], "local": local or "",
+                     "remote": remote, "status": status, "error": error})
+        time.sleep(0.3)                   # be polite to Justice Laws
+    return rows
+
+
+def non_xml_currency():
+    """Stored currency (newest item) of the aggregate sources."""
+    rows = []
+    for label, fname, cmd, desc in _NON_XML:
+        path = PROCESSED_DIR / fname
+        if not path.exists():
+            rows.append({"label": label, "desc": desc, "cmd": cmd,
+                         "newest": "", "chunks": 0, "present": False})
+            continue
+        chunks = json.loads(path.read_text(encoding="utf-8"))
+        newest = max((c.get("current_to", "") for c in chunks), default="")
+        rows.append({"label": label, "desc": desc, "cmd": cmd,
+                     "newest": newest, "chunks": len(chunks), "present": True})
+    return rows
+
+
+def pending_in_force(today):
+    """Pending bills whose in-force date has passed as of `today` (ISO str)."""
+    return [b for b in _PENDING_BILLS if b["in_force"] <= today]
+
+
+def run(as_json=False):
+    today = datetime.date.today().isoformat()
+    leg = check_legislation()
+    nonxml = non_xml_currency()
+    pending = pending_in_force(today)
+    stale = [r for r in leg if r["status"] in ("stale", "missing-local")]
+    errors = [r for r in leg if r["status"] == "error"]
+
+    if as_json:
+        print(json.dumps({"checked": today, "legislation": leg,
+                          "non_xml": nonxml, "pending_in_force": pending,
+                          "stale_count": len(stale),
+                          "error_count": len(errors)}, indent=2))
+    else:
+        print(f"CanLex corpus staleness check — {today}\n")
+        print(f"Justice Laws legislation ({len(leg)} instruments):")
+        for r in sorted(leg, key=lambda x: (x["status"] != "stale", x["short"])):
+            if r["status"] == "stale":
+                print(f"  STALE  {r['short']:34} local {r['local']}  "
+                      f"upstream {r['remote']}")
+            elif r["status"] == "missing-local":
+                print(f"  ABSENT {r['short']:34} configured but not ingested "
+                      f"(upstream {r['remote']})")
+            elif r["status"] == "error":
+                print(f"  ERROR  {r['short']:34} {r['error'][:50]}")
+            elif r["status"] == "no-date":
+                print(f"  ?      {r['short']:34} no current-date found upstream")
+        if not stale and not errors:
+            print(f"  all {len(leg)} current (local matches or exceeds upstream)")
+        if stale:
+            codes = " ".join(r["code"] for r in stale)
+            print(f"\n  -> {len(stale)} drifted. Refresh: "
+                  f"py -m canlex.ingest --force {codes}\n"
+                  f"     then: py -m canlex.embed && py -m canlex.eval")
+
+        print("\nOther sources (stored currency; re-run the ingester to refresh):")
+        for r in nonxml:
+            mark = " " if r["present"] else "!"
+            print(f" {mark}{r['label']:14} {str(r['chunks']):>5} chunks  "
+                  f"newest {r['newest'] or 'n/a':12}  {r['cmd']}")
+
+        if pending:
+            print("\nPending bills now in force (re-ingest the affected Acts):")
+            for b in pending:
+                print(f"  Bill {b['bill']} ({b['name']}) in force {b['in_force']} "
+                      f"-> {', '.join(b['affects'])}")
+                print(f"     {b['note']}")
+                print(f"     run: py -m canlex.ingest --force {' '.join(b['affects'])}")
+
+    return 1 if (stale or errors) else 0
+
+
+def main():
+    args = sys.argv[1:]
+    as_json = "--json" in args
+    # --check is the only mode; accepted (and default) for forward-compatibility.
+    sys.exit(run(as_json=as_json))
+
+
+if __name__ == "__main__":
+    main()

tests/test_refresh.py

@@ -0,0 +1,45 @@
+"""Unit tests for the staleness-check logic (canlex/refresh.py). Offline only."""
+import unittest
+
+from canlex import refresh
+
+
+class ExtractCurrentDateTests(unittest.TestCase):
+    def test_reads_root_attribute(self):
+        head = (b'<?xml version="1.0"?>\n<Statute xmlns:lims="..." '
+                b'lims:current-date="2026-03-31" lims:lastAmendedDate="2026-03-26">')
+        self.assertEqual(refresh._extract_current_date(head), "2026-03-31")
+
+    def test_strips_utf8_bom(self):
+        head = b'\xef\xbb\xbf<Regulation lims:current-date="2019-06-21">'
+        self.assertEqual(refresh._extract_current_date(head), "2019-06-21")
+
+    def test_returns_empty_when_absent(self):
+        self.assertEqual(refresh._extract_current_date(b"<Statute foo='bar'>"), "")
+
+    def test_takes_the_current_date_not_amended_date(self):
+        # lastAmendedDate must not be mistaken for current-date.
+        head = b'<Statute lims:lastAmendedDate="2099-01-01" lims:current-date="2026-05-26">'
+        self.assertEqual(refresh._extract_current_date(head), "2026-05-26")
+
+
+class PendingInForceTests(unittest.TestCase):
+    def test_surfaces_a_bill_once_its_date_passes(self):
+        # C-16 in force 2026-07-18 in the tracked list.
+        after = refresh.pending_in_force("2026-08-01")
+        self.assertTrue(any(b["bill"] == "C-16" for b in after))
+
+    def test_hides_a_bill_before_its_date(self):
+        before = refresh.pending_in_force("2026-07-01")
+        self.assertFalse(any(b["bill"] == "C-16" for b in before))
+
+
+class DriftSemanticsTests(unittest.TestCase):
+    def test_iso_date_string_compare_detects_newer_upstream(self):
+        # The check uses plain string comparison on ISO dates; verify ordering.
+        self.assertTrue("2026-05-26" > "2026-03-31")
+        self.assertFalse("2026-03-31" > "2026-03-31")
+
+
+if __name__ == "__main__":
+    unittest.main()