refresh.py: upstream staleness check for the NJC directives

Directives graduate from stored-date reporting to upstream-checked, like the
delegation instruments. The effective date directive.py stores comes from the
NJC index page's span.date, not the individual directive pages (9 of 11 carry
no date; d6's page prose says 2009 against a true 2021), so check_directives()
probes one fresh index fetch, with a page-prose fallback only for the
index-undated Foreign Service Directives. Probe fetches go to a temp dir --
never data/raw. Staleness affects the exit code; fetch errors do not. Five
offline tests.

First live run caught real drift: the FSDs at stored 2026-04-01 vs upstream
2026-06-01 (re-ingested in the caselaw commit).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

canlex/refresh.py

@@ -14,11 +14,17 @@ It also checks the two delegation instruments against upstream (check_delegation
 the CBSA IRPA delegation -- by diffing the dated 'irpa-lipr-YYYY-MM-DD' instruments
 the CBSA legislation index publishes against those delegation.SOURCES ingests, so
 a newly-published amendment is flagged -- and the IRCC IL3 instrument, by comparing
-the version stamp in the live PDF against the stored one. Finally it prints the
-stored currency of the remaining non-XML sources (D-memos, case law, agreements,
-directives, benefits, tariff schedule), which expose no comparable upstream
-signal, with the command to refresh each; and flags any tracked pending bill
-whose coming-into-force date has passed.
+the version stamp in the live PDF against the stored one. The NJC directives get
+the same treatment (check_directives): the effective date each chunk stores in
+current_to comes from the NJC directive index (the span.date beside each entry),
+so one fresh index fetch re-reads it for nearly every directive, with the
+'effective <Month D, YYYY>' prose on the directive's own page as fallback for
+entries the index leaves undated (the Foreign Service Directives) -- exactly the
+two signals ingestion uses. Finally it prints the stored currency of the
+remaining non-XML sources (D-memos, case law, agreements, benefits, tariff
+schedule), which expose no comparable upstream signal, with the command to
+refresh each; and 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.
@@ -35,8 +41,10 @@ import time
 import urllib.request
 from pathlib import Path
 
-from . import delegation
-from ._common import BROWSER_UA
+from bs4 import BeautifulSoup
+
+from . import delegation, directive
+from ._common import BROWSER_UA, MONTHS
 from .config import SOURCES, PROCESSED_DIR
 
 _UA = "CanLex-refresh/0.1"
@@ -46,19 +54,26 @@ _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. (Delegation is checked against upstream
-# separately, by check_delegation, because its sources DO expose a comparable
-# signal -- the CBSA instrument index and the IL3 version string.)
+# rather than diffing against upstream. (Delegation and the NJC directives are
+# checked against upstream separately, by check_delegation and check_directives,
+# because their sources DO expose a comparable signal -- the CBSA instrument
+# index and IL3 version string, and the NJC directive index dates.)
 _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"),
     ("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"),
 ]
 
+# NJC directive dates as published: the index prints prose ('June  1, 1993' --
+# double spaces occur in the raw markup) while chunks may store either that
+# prose or the ISO fallback directive.py derives from page text, so both sides
+# are normalized to ISO before comparing.
+_NJC_PROSE_DATE = re.compile(r"([A-Za-z]+)\s+(\d{1,2}),?\s+(\d{4})")
+_ISO_DATE = re.compile(r"\d{4}-\d{2}-\d{2}")
+
 # CBSA publishes its IRPA delegation as a base instrument plus incremental
 # amendments (no consolidation), each a dated 'irpa-lipr-YYYY-MM-DD-eng.html'
 # linked from the CBSA legislation index. A date on the index that we have not
@@ -219,6 +234,123 @@ def check_delegation():
     return rows
 
 
+def _stored_directives():
+    """Per-directive stored currency from data/processed/directives.json:
+    {code: (title, current_to, source_url)}, one entry per act_code."""
+    path = PROCESSED_DIR / "directives.json"
+    if not path.exists():
+        return {}
+    chunks = json.loads(path.read_text(encoding="utf-8"))
+    out = {}
+    for c in chunks:
+        code = c.get("act_code")
+        if code and code not in out:
+            out[code] = (c.get("act_short", ""), c.get("current_to", ""),
+                         c.get("source_url", ""))
+    return out
+
+
+def _directive_iso(text):
+    """Normalize an NJC directive date ('June  1, 1993' or already ISO) to
+    YYYY-MM-DD. Pure, for testability. Returns '' when not a recognizable date."""
+    text = (text or "").strip()
+    if _ISO_DATE.fullmatch(text):
+        return text
+    m = _NJC_PROSE_DATE.fullmatch(text)
+    if m and m.group(1).lower() in MONTHS:
+        return (f"{int(m.group(3)):04d}-{MONTHS[m.group(1).lower()]:02d}-"
+                f"{int(m.group(2)):02d}")
+    return ""
+
+
+def _njc_index_dates(html):
+    """{code: date text} parsed from the NJC directive index, mirroring
+    directive.directive_links: the first <a> in each top-level li of
+    ul.directive-list is the current directive (nested lis are archived
+    versions), and the sibling span.date its effective date ('' if absent)."""
+    soup = BeautifulSoup(html, "html.parser")
+    ul = soup.find("ul", class_="directive-list")
+    out = {}
+    if ul is None:
+        return out
+    for li in ul.find_all("li", recursive=False):
+        a = li.find("a", href=True)
+        if not a:
+            continue
+        m = directive._CODE.search(a["href"])
+        if not m:
+            continue
+        span = li.find("span", class_="date")
+        out[m.group(1)] = span.get_text(strip=True) if span else ""
+    return out
+
+
+def _njc_page_effective(url, td):
+    """Fresh-fetch a directive's own page and pull its 'effective <Month D,
+    YYYY>' prose (directive._page_effective_date), following the 'Print Full
+    Directive' link when the landing page is only a table of contents (the
+    FSDs). Returns (iso_date, error)."""
+    try:
+        html = directive._fetch(url, td / "probe.html", force=True)
+        soup = BeautifulSoup(html, "html.parser")
+        main = soup.find("main")
+        date = directive._page_effective_date(main) if main else ""
+        if not date:
+            print_url = directive._print_link(html)
+            if print_url:
+                full = directive._fetch(print_url, td / "probe-full.html",
+                                        force=True)
+                fmain = BeautifulSoup(full, "html.parser").find("main")
+                date = directive._page_effective_date(fmain) if fmain else ""
+        return date, ""
+    except Exception as exc:
+        return "", f"{type(exc).__name__}: {exc}"
+
+
+def check_directives():
+    """Compare each ingested NJC directive's effective date against upstream.
+    Returns row dicts.
+
+    The date directive.py stores in current_to comes from the NJC directive
+    index (the span.date beside each entry), so one fresh index fetch covers
+    nearly every directive; entries the index leaves undated (the Foreign
+    Service Directives) fall back to the 'effective <Month D, YYYY>' prose on
+    the directive's own page -- the same two signals ingestion uses. Probe
+    fetches go to a temp dir, never data/raw, so a stale probe can't mask a
+    later real refresh; directive._fetch already sleeps 0.5s after each
+    download, so the probes stay polite without extra pauses here."""
+    stored = _stored_directives()
+    rows = []
+    if not stored:
+        return rows
+    with tempfile.TemporaryDirectory() as td:
+        td = Path(td)
+        try:
+            html = directive._fetch(directive.INDEX_URL, td / "_index.html",
+                                    force=True)
+            index_dates, index_error = _njc_index_dates(html), ""
+        except Exception as exc:
+            index_dates, index_error = {}, f"{type(exc).__name__}: {exc}"
+        for code, (title, stored_date, url) in sorted(stored.items()):
+            upstream, error = index_dates.get(code, ""), index_error
+            if not upstream and not error:
+                # Undated on the index: read the date off the page itself.
+                upstream, error = _njc_page_effective(url, td)
+            local, remote = _directive_iso(stored_date), _directive_iso(upstream)
+            if error:
+                status = "error"
+            elif not remote:
+                status = "no-date"
+            elif remote != local:
+                status = "stale"      # NJC reissues in place: any drift counts
+            else:
+                status = "ok"
+            rows.append({"code": code, "title": title, "stored": stored_date,
+                         "local": local, "remote": remote, "status": status,
+                         "error": error})
+    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]
@@ -228,17 +360,20 @@ def run(as_json=False):
     today = datetime.date.today().isoformat()
     leg = check_legislation()
     deleg = check_delegation()
+    dirs = check_directives()
     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"]
     deleg_stale = [r for r in deleg if r["status"] == "stale"]
+    dirs_stale = [r for r in dirs if r["status"] == "stale"]
 
     if as_json:
         print(json.dumps({"checked": today, "legislation": leg,
-                          "delegation": deleg, "non_xml": nonxml,
-                          "pending_in_force": pending,
-                          "stale_count": len(stale) + len(deleg_stale),
+                          "delegation": deleg, "directives": dirs,
+                          "non_xml": nonxml, "pending_in_force": pending,
+                          "stale_count": (len(stale) + len(deleg_stale)
+                                          + len(dirs_stale)),
                           "error_count": len(errors)}, indent=2))
     else:
         print(f"CanLex corpus staleness check — {today}\n")
@@ -285,6 +420,21 @@ def run(as_json=False):
             print("     -> re-ingest: add any new CBSA dates to delegation.SOURCES, "
                   "then py -m canlex.delegation && py -m canlex.embed")
 
+        print(f"\nNJC directives (upstream check, {len(dirs)} ingested):")
+        for r in dirs:
+            if r["status"] == "stale":
+                print(f"  STALE  {r['title']}: upstream {r['remote']} "
+                      f"!= stored {r['local'] or r['stored'] or '?'}")
+            elif r["status"] == "error":
+                print(f"  ERROR  {r['title']}: {r['error'][:50]}")
+            elif r["status"] == "no-date":
+                print(f"  ?      {r['title']}: no effective date found upstream")
+            else:
+                print(f"  ok     {r['title']}: current ({r['remote']})")
+        if dirs_stale:
+            print("     -> re-ingest: py -m canlex.directive --force && "
+                  "py -m canlex.embed")
+
         print("\nOther sources (stored currency; re-run the ingester to refresh):")
         for r in nonxml:
             mark = " " if r["present"] else "!"
@@ -299,7 +449,7 @@ def run(as_json=False):
                 print(f"     {b['note']}")
                 print(f"     run: py -m canlex.ingest --force {' '.join(b['affects'])}")
 
-    return 1 if (stale or errors or deleg_stale) else 0
+    return 1 if (stale or errors or deleg_stale or dirs_stale) else 0
 
 
 def main():

tests/test_refresh.py

@@ -54,6 +54,36 @@ class CbsaInstrumentRegexTests(unittest.TestCase):
         self.assertEqual(m.group(0), "Fall 2025")
 
 
+class DirectiveCheckTests(unittest.TestCase):
+    def test_directive_iso_normalizes_index_prose(self):
+        # The live index prints double spaces before single-digit days.
+        self.assertEqual(refresh._directive_iso("June  1, 1993"), "1993-06-01")
+        self.assertEqual(refresh._directive_iso("March 28, 2026"), "2026-03-28")
+
+    def test_directive_iso_passes_iso_through(self):
+        # FSD chunks store the ISO fallback directive.py derives from page text.
+        self.assertEqual(refresh._directive_iso("2026-04-01"), "2026-04-01")
+
+    def test_directive_iso_rejects_junk(self):
+        self.assertEqual(refresh._directive_iso("not a date"), "")
+        self.assertEqual(refresh._directive_iso("Smarch 1, 2026"), "")
+        self.assertEqual(refresh._directive_iso(""), "")
+
+    def test_index_dates_take_current_entries_not_archived(self):
+        html = ('<ul class="directive-list">'
+                '<li><a href="/directive/d2/en"><strong>Commuting</strong></a> '
+                '<span class="date">October  1, 2020</span>'
+                '<ul><li><a href="/directive/d2/v22/en"><strong>'
+                'Archived version: June 1, 2010</strong></a></li></ul></li>'
+                '<li><a href="/directive/fsd-dse/en"><strong>FSD</strong></a>'
+                '</li></ul>')
+        self.assertEqual(refresh._njc_index_dates(html),
+                         {"d2": "October  1, 2020", "fsd-dse": ""})
+
+    def test_index_dates_empty_when_list_missing(self):
+        self.assertEqual(refresh._njc_index_dates("<html><body/></html>"), {})
+
+
 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.