Guard every ingester against the corpus-wipe failure mode

Every ingester rebuilds its whole processed file from a fresh scrape and
then wrote unconditionally, so an upstream page that changes shape becomes
data loss. That is not hypothetical: it emptied dmemos.json last commit.

_common gains write_corpus / stored_chunks / preserve_dropped, and all 14
corpus writers now use them:

- write_corpus refuses to replace a corpus with under 80% of its stored
chunks and the process exits non-zero, so a chained `&& py -m
canlex.embed` stops instead of propagating a gutted corpus.
--allow-shrink is the override once a drop is understood.
- Writes are atomic (tmp + os.replace). A run killed mid-write used to
leave truncated JSON, which stored_chunks reads as empty -- disabling
the guard on the next run, exactly when it is needed.
- preserve_dropped re-attaches the last-good chunks of items that failed
this run, restricted to items that actually errored so anything removed
on purpose stays gone. A stored chunk with no identity field is skipped
rather than crashing the rebuild.

ingest.py (the 132 per-Act legislation files, the largest corpus) and
irb_guidelines.py were missing from the first sweep and are covered too.
caselaw.py had no failure tracking at all -- its three loops printed and
continued -- so it now records failures keyed on act_short, the identity
its chunks actually carry.

enf.py is the one deliberate exception to preserve-on-failure: _CHAPTERS is
a blind range(1, 41) probe and a 404 is the module's documented signal that
IRCC retired a chapter, so 404s drop the chapter (is_missing_upstream) while
every other error preserves it. Preserving through 404s would have pinned
withdrawn chapters in the corpus forever. enf and pdi also now treat
"fetched fine, parsed to zero chunks" as a failure -- the exact shape the
D-memo loss took.

No corpus content changes: every data/processed file is byte-identical, and
dmemo rebuilds to the same 1,857 chunks through the new write path. 278
tests (up from 129), all offline.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

RUNBOOK.md

@@ -30,6 +30,31 @@ py -m canlex.eval                       # gate: no metric below the last committ
 vectors for chunks missing from embeddings.npz and quietly degrades them to
 BM25-only (a stderr warning is the only signal).
 
+### The corpus-write guard
+
+Every ingester above rebuilds its whole processed file from a fresh scrape, so
+an upstream page that changes shape turns into data loss at the write. All of
+them now go through `_common.write_corpus`, which **refuses to write a corpus
+below 80% of the stored one** and makes the process exit non-zero:
+
+```
+  REFUSING to write: 0 chunks would replace 1870 already in dmemos.json. ...
+```
+
+That is not a transient error — it means the scrape broke. Diagnose it, don't
+retry. Once the drop is understood and intended (a source genuinely shrank),
+re-run the same module with `--allow-shrink`. Because the runbook chains these
+as `py -m canlex.X && py -m canlex.embed`, a refusal now correctly stops the
+embed instead of propagating a gutted corpus.
+
+Most ingesters also **preserve on failure**: an item that errors keeps its
+last-good chunks (`_common.preserve_dropped`), restricted to items that
+actually failed, so anything removed on purpose stays gone. Deliberate
+exception: `enf.py` treats an HTTP 404 as IRCC retiring a chapter — its
+documented policy — so a 404 drops the chapter while any other error preserves
+it. Writes are atomic (tmp + `os.replace`): a killed run cannot leave truncated
+JSON that would read as an empty corpus and disable the guard next time.
+
 ## Deploys
 
 - **MCP Space** (`Beemer0/CanLex`): commit, then `git push space main`.
@@ -86,11 +111,9 @@ only if it disappears. Two consequences worth knowing:
   that fails to fetch, so a 404 (or a dropped connection) can no longer delete
   still-in-force guidance. The two D8s were never ingestable and are absent.
 
-The ingester also refuses to write a corpus smaller than 80% of the stored one
-(`--allow-shrink` overrides). That guard exists because the first run after the
+This is the incident the write guard above exists for: the first run after the
 rebuild scraped zero memos and wrote an empty dmemos.json straight over 1,870
-good chunks — recovered from git. Any ingester that rebuilds a whole file from
-a scrape has this failure mode; dmemo.py is the only one hardened so far.
+good chunks, recovered from git. Every ingester now carries the guard.
 
 ## Curated-dataset review protocol
 

canlex/_common.py

@@ -8,6 +8,8 @@ UA, a longer timeout, a different split target -- stay at the call sites as
 arguments. caselaw.py keeps its own fetch (retry/throttle for Lexum's CAPTCHA)
 and its own splitter (sentence-boundary windowing, a different algorithm).
 """
+import json
+import os
 import re
 import subprocess
 import time
@@ -94,6 +96,90 @@ def split_lines(text, target=1800, fold_stubs=False):
     return pieces
 
 
+# Every ingester rebuilds its whole processed file from a fresh scrape, so an
+# upstream page that changes shape -- or a run that loses its network halfway --
+# shows up as a collapse in chunk count, and the write turns that into data
+# loss. On 2026-07-27 CBSA's rebuilt D-memo index scraped to zero memos and the
+# write replaced 1,870 good chunks with an empty list; one line of stdout was
+# the only warning. Refuse to write below this share of what is already stored.
+CORPUS_SHRINK_GUARD = 0.8
+
+
+def stored_chunks(path):
+    """The chunks already written to `path`, or [] if absent or unreadable."""
+    if not path.exists():
+        return []
+    try:
+        return json.loads(path.read_text(encoding="utf-8"))
+    except (ValueError, OSError):
+        return []
+
+
+def safe_to_write(new_count, old_count, ratio=CORPUS_SHRINK_GUARD):
+    """False when writing `new_count` chunks would gut an existing corpus.
+
+    Pure, for tests. The write is the irreversible step, so it is the one to
+    gate -- a first-ever run (old_count 0) is always allowed.
+    """
+    return old_count == 0 or new_count >= old_count * ratio
+
+
+def preserve_dropped(chunks, stored, *, key=lambda c: c["section"], only=None):
+    """Re-attach the last-good chunks of items this run failed to produce.
+
+    `key` maps a chunk to the item it belongs to; `only` restricts preservation
+    to a set of item identities (pass the ones that actually errored, so items
+    dropped ON PURPOSE -- cancelled memos, retired manuals -- are not
+    resurrected). Returns (chunks, preserved identities).
+    """
+    def identity(chunk):
+        # A chunk written by an older schema may not carry the identity field
+        # at all. It simply is not preservable -- but crashing here would take
+        # down the rebuild at the exact moment the guard exists to protect it.
+        try:
+            return key(chunk)
+        except (KeyError, IndexError, TypeError):
+            return None
+
+    fresh = {identity(c) for c in chunks}
+    by_item = {}
+    for chunk in stored:
+        item = identity(chunk)
+        if item is not None:
+            by_item.setdefault(item, []).append(chunk)
+    dropped = sorted(item for item in by_item
+                     if item not in fresh and (only is None or item in only))
+    for item in dropped:
+        chunks.extend(by_item[item])
+    return chunks, dropped
+
+
+def write_corpus(path, chunks, *, indent=2, allow_shrink=False,
+                 ratio=CORPUS_SHRINK_GUARD):
+    """Write `chunks` to `path`, unless that would gut the stored corpus.
+
+    Returns True if written, False if refused -- callers exit non-zero on False
+    so a broken scrape fails loudly instead of quietly emptying a file.
+    """
+    stored = stored_chunks(path)
+    if not allow_shrink and not safe_to_write(len(chunks), len(stored), ratio):
+        print(f"  REFUSING to write: {len(chunks)} chunks would replace "
+              f"{len(stored)} already in {path.name}. The upstream source "
+              f"likely changed shape -- check the scrape before writing "
+              f"(re-run with --allow-shrink once the drop is understood).")
+        return False
+    path.parent.mkdir(parents=True, exist_ok=True)
+    # Write-then-rename, not write-in-place: a run killed mid-write (Ctrl-C on
+    # the scheduled refresh, a full disk) would otherwise leave truncated JSON
+    # behind, and stored_chunks reads unparseable as empty -- which disables
+    # the guard above on the very next run, when it is needed most.
+    tmp = path.with_name(path.name + ".tmp")
+    tmp.write_text(json.dumps(chunks, ensure_ascii=False, indent=indent),
+                   encoding="utf-8")
+    os.replace(tmp, path)
+    return True
+
+
 def uniquify_ids(chunks):
     """Rename duplicate chunk ids in place (second occurrence -> id-b, -c...).
 

canlex/agreement.py

@@ -4,13 +4,13 @@ A collective agreement is a binding contract between the Treasury Board and a
 bargaining agent for one occupational group. Chunks are tagged
 doc_type="agreement" so CanLex keeps them distinct from legislation and guidance.
 """
-import json
 import re
 import sys
 
 from bs4 import BeautifulSoup
 
-from ._common import fetch_cached, norm_ws as _norm, split_lines
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      split_lines, stored_chunks, write_corpus)
 from .config import RAW_DIR, PROCESSED_DIR
 
 AGREEMENT_DIR = RAW_DIR / "agreements"
@@ -142,9 +142,23 @@ def parse_agreement(html, code):
     return chunks
 
 
-def main():
-    force = "--force" in sys.argv
-    all_chunks = []
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of agreements this run could not rebuild.
+
+    Ingestion is a full rebuild, so an agreement whose page fails to fetch --
+    or whose canada.ca layout changes so that nothing parses out of it --
+    disappears from the corpus. An agreement that is still in force is better
+    served from the copy we already had than not at all. Restricted to the
+    codes that actually failed, so one removed from AGREEMENTS on purpose is
+    not resurrected. Returns (chunks, preserved codes).
+    """
+    failed = {code for code, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["act_code"],
+                            only=failed)
+
+
+def ingest(force=False, allow_shrink=False):
+    all_chunks, failures = [], []
     for code, meta in AGREEMENTS.items():
         print(f"Ingesting {code} ({meta['short']})...")
         try:
@@ -152,12 +166,30 @@ def main():
             chunks = parse_agreement(html, code)
             print(f"  {len(chunks)} chunks")
             all_chunks.extend(chunks)
+            if not chunks:
+                # A page that parses to nothing is a redesign upstream, not an
+                # empty agreement -- same class of failure as a dead fetch.
+                failures.append((code, "no content parsed"))
         except Exception as exc:
+            failures.append((code, f"{type(exc).__name__}: {exc}"))
             print(f"  FAILED {code}: {type(exc).__name__}: {exc}")
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT_FILE.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=2),
-                        encoding="utf-8")
+
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(OUT_FILE))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} agreement(s): "
+              f"{', '.join(preserved)}")
+    if not write_corpus(OUT_FILE, all_chunks, indent=2,
+                        allow_shrink=allow_shrink):
+        return False
     print(f"\n{len(all_chunks)} chunks from {len(AGREEMENTS)} agreement(s) -> {OUT_FILE.name}")
+    return True
+
+
+def main():
+    ok = ingest(force="--force" in sys.argv,
+                allow_shrink="--allow-shrink" in sys.argv)
+    sys.exit(0 if ok else 1)
 
 
 if __name__ == "__main__":

canlex/amps.py

@@ -14,10 +14,11 @@ short), doc_type='memorandum' (administrative guidance, not law).
 
     python -m canlex.amps
 """
-import json
 import re
+import sys
 
-from ._common import fetch_cached, norm_ws as _norm
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      stored_chunks, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 INDEX_URL = "https://www.cbsa-asfc.gc.ca/trade-commerce/amps/mpd-dmi-eng.html"
@@ -89,7 +90,23 @@ def _fetch(url, dest):
     return data
 
 
-def build(limit=None):
+def preserve_failed(chunks, failed, stored):
+    """Re-attach the last-good chunks of contraventions this run could not fetch.
+
+    Ingestion is a full rebuild, so a page that 404s or times out silently
+    leaves the corpus -- and a penalty still in force is worse missing than
+    slightly stale. `failed` carries the (code, why) pairs build() collects;
+    codes are lower-case on the index but upper-case in a chunk's `section`,
+    so match on the upper-cased form. Restricted to codes that actually
+    errored, so a contravention CBSA has genuinely retired stays gone.
+    Returns (chunks, preserved codes).
+    """
+    codes = {code.upper() for code, _why in failed}
+    return preserve_dropped(chunks, stored, key=lambda c: c["section"].upper(),
+                            only=codes)
+
+
+def build(limit=None, allow_shrink=False):
     html = _fetch(INDEX_URL, RAW / "index.html").decode("utf-8", "replace")
     links = sorted({(m.group(1), m.group(2)) for m in _LINK.finditer(html)},
                    key=lambda t: t[1])
@@ -110,12 +127,28 @@ def build(limit=None):
         except Exception as exc:
             failed.append((code, f"{type(exc).__name__}: {exc}"))
     out = PROCESSED_DIR / "amps.json"
-    out.write_text(json.dumps(chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
-    print(f"amps: {len(chunks)} contraventions -> {out.name}"
-          + (f" ({len(failed)} failed: {failed[:5]})" if failed else ""))
-    return chunks
+    # Printed before the write so a refusal still shows what went wrong.
+    if failed:
+        print(f"amps: {len(failed)} failed: {failed[:5]}")
+    if not limit:
+        chunks, preserved = preserve_failed(chunks, failed, stored_chunks(out))
+        if preserved:
+            print(f"amps: kept the last-good copy of {len(preserved)} "
+                  f"unfetchable contravention(s): {', '.join(preserved)}")
+    # The guard applies to --limit runs too: a smoke test over five
+    # contraventions must not be able to write a five-page corpus over the real
+    # one. indent=2 matches what is already on disk -- changing it would
+    # rewrite every line of the file.
+    if not write_corpus(out, chunks, indent=2, allow_shrink=allow_shrink):
+        return False
+    print(f"amps: {len(chunks)} contraventions -> {out.name}")
+    return True
+
+
+def main():
+    ok = build(allow_shrink="--allow-shrink" in sys.argv)
+    sys.exit(0 if ok else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

canlex/benefits.py

@@ -29,13 +29,14 @@ benefit topic.
     py -m canlex.benefits
 """
 import io
-import json
 import re
+import sys
 
 from bs4 import BeautifulSoup
 from pypdf import PdfReader
 
-from ._common import fetch_cached, norm_ws, split_lines
+from ._common import (fetch_cached, norm_ws, preserve_dropped, split_lines,
+                      stored_chunks, write_corpus)
 
 from .config import PROCESSED_DIR, RAW_DIR
 
@@ -282,8 +283,24 @@ def _fetch_html(url, dest):
     return fetch_cached(url, dest, powershell=True)
 
 
-def build():
-    all_chunks = []
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of plans this run could not rebuild.
+
+    Ingestion is a full rebuild of all four booklets into one file, so one plan
+    failing -- Canada Life moves a PDF, canada.ca times out, a re-typeset
+    booklet no longer segments -- silently deletes that plan's coverage while
+    the other three look healthy. A plan's chunks all carry its code in
+    act_code, so the stored copy can be restored intact. Restricted to plans
+    that actually failed, so a plan removed from SOURCES on purpose (a wound-up
+    plan) is not resurrected. Returns (chunks, preserved plan codes).
+    """
+    failed = {code for code, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["act_code"],
+                            only=failed)
+
+
+def build(allow_shrink=False):
+    all_chunks, failures = [], []
     for src in SOURCES.values():
         print(f"Ingesting {src['code']} {src['label'].lower()} ...")
         try:
@@ -296,15 +313,34 @@ def build():
                 chunks = parse_html_booklet(html, src)
         except Exception as exc:
             print(f"  !! {src['code']}: {type(exc).__name__}: {exc}")
+            failures.append((src["code"], f"{type(exc).__name__}: {exc}"))
+            continue
+        if not chunks:
+            # A booklet that downloads fine but segments to nothing has failed
+            # just as surely as one that 404s -- that is how the CBSA index
+            # loss looked -- so it counts as a failure rather than a plan that
+            # legitimately has no content.
+            print(f"  !! {src['code']}: no sections parsed")
+            failures.append((src["code"], "no sections parsed"))
             continue
         all_chunks.extend(chunks)
         print(f"  {len(chunks)} chunks")
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=1),
-                   encoding="utf-8")
+
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(OUT))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} booklet(s): "
+              f"{', '.join(preserved)}")
+    if not write_corpus(OUT, all_chunks, indent=1, allow_shrink=allow_shrink):
+        return False
     print(f"\n{len(all_chunks)} benefit-plan chunks from {len(SOURCES)} "
           f"booklet(s) -> {OUT}")
+    return True
+
+
+def main():
+    sys.exit(0 if build(allow_shrink="--allow-shrink" in sys.argv) else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

canlex/caselaw.py

@@ -10,15 +10,16 @@ deliberately not a comprehensive scrape.
 
     py -m canlex.caselaw
 """
-import json
 import re
+import sys
 import time
 import urllib.error
 import urllib.request
 
 from bs4 import BeautifulSoup
 
-from ._common import MONTHS as _MONTHS, norm_ws as _norm
+from ._common import (MONTHS as _MONTHS, norm_ws as _norm, preserve_dropped,
+                      stored_chunks, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 # Each court or tribunal's Lexum decisions database: (display name, item-URL
@@ -852,9 +853,21 @@ def _irb_chunks(guide, soup):
     return chunks, cite, len(paras)
 
 
-def build():
-    """Fetch, parse and chunk every curated decision into caselaw.json."""
-    all_chunks = []
+def build(allow_shrink=False):
+    """Fetch, parse and chunk every curated decision into caselaw.json.
+
+    Returns True when the corpus was written, False when the write was refused
+    -- a Lexum CAPTCHA wall or an upstream re-platforming makes every fetch
+    fail at once, and the rebuild-then-write shape would turn that into an
+    emptied caselaw.json.
+    """
+    # Every path that drops a decision records it here, keyed the way the
+    # chunks are (act_short): Lexum throttles, CanLII captures go missing and
+    # court templates shift, so on any given run some decisions fail while the
+    # rest succeed -- and the rebuild-then-write shape would quietly delete
+    # exactly those. `failed` is what preserve_dropped is allowed to restore,
+    # so decisions removed on purpose (deleted from CASES) stay gone.
+    all_chunks, failed = [], set()
     for case in CASES:
         try:
             soup = BeautifulSoup(_fetch(case["court"], case["id"]),
@@ -862,11 +875,13 @@ def build():
         except Exception as exc:
             print(f"  !! {case['short']}: fetch failed -- "
                   f"{type(exc).__name__}: {exc}")
+            failed.add(case["short"])
             continue
         chunks, citation, n_paras = _decision_chunks(case, soup)
         if not chunks:
             print(f"  !! {case['short']} ({case['court']} item "
                   f"{case['id']}): 0 chunks -- check parsing")
+            failed.add(case["short"])
             continue
         all_chunks.extend(chunks)
         print(f"  {case['court']:4s} {case['short']:20s} {n_paras:4d} paras -> "
@@ -878,12 +893,14 @@ def build():
                   f"automated fetches (DataDome). Open https://www.canlii.org/"
                   f"{case['canlii_path']} in a browser and save the page text "
                   f"to that path, then re-run.")
+            failed.add(case["short"])
             continue
         text = cache.read_text(encoding="utf-8-sig")
         chunks, citation, n_paras = _canlii_chunks(case, text)
         if not chunks:
             print(f"  !! {case['short']} ({case['file']}): 0 chunks -- "
                   f"check parsing")
+            failed.add(case["short"])
             continue
         all_chunks.extend(chunks)
         print(f"  cnli {case['short']:20s} {n_paras:4d} paras -> "
@@ -895,21 +912,38 @@ def build():
         except Exception as exc:
             print(f"  !! {guide['file']}: fetch failed -- "
                   f"{type(exc).__name__}: {exc}")
+            failed.add(guide["file"])
             continue
         chunks, citation, n_paras = _irb_chunks(guide, soup)
         if not chunks:
             print(f"  !! {guide['file']}: 0 chunks -- check parsing")
+            failed.add(guide["file"])
             continue
         all_chunks.extend(chunks)
         print(f"  irb  {guide['file']:20s} {n_paras:4d} paras -> "
               f"{len(chunks):3d} chunks   {citation}")
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=1),
-                   encoding="utf-8")
+    # act_short is the per-decision identity in this corpus (section is "" for
+    # every case-law chunk), and it is unique across all three lists.
+    all_chunks, preserved = preserve_dropped(all_chunks, stored_chunks(OUT),
+                                             key=lambda c: c["act_short"],
+                                             only=failed)
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} decision(s) "
+              f"that failed this run: {', '.join(preserved)}")
+    # indent=1 is what this corpus was written with; changing it would rewrite
+    # every line of caselaw.json for no content change.
+    if not write_corpus(OUT, all_chunks, indent=1, allow_shrink=allow_shrink):
+        return False
     print(f"\n{len(all_chunks)} case-law chunks from "
           f"{len(CASES) + len(CANLII_CASES) + len(IRB_GUIDES)} decisions "
           f"-> {OUT}")
+    return True
+
+
+def main():
+    ok = build(allow_shrink="--allow-shrink" in sys.argv)
+    sys.exit(0 if ok else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

canlex/charter.py

@@ -11,14 +11,15 @@ the supremacy clause s. 52, and the 1867 heads-of-power ss. 91-92.
 
     python -m canlex.charter
 """
-import json
 import re
+import sys
 
-from ._common import fetch_cached, norm_ws as _norm
+from ._common import fetch_cached, norm_ws as _norm, write_corpus
 from .config import PROCESSED_DIR, RAW_DIR
 
 URL = "https://laws-lois.justice.gc.ca/eng/const/FullText.html"
 RAW = RAW_DIR / "charter"
+OUT_FILE = PROCESSED_DIR / "charter.json"
 
 # (act filter, sections wanted). The 1982 Act's Charter is ss. 1-34; s. 35 is
 # Part II (Aboriginal rights) and s. 52 the supremacy clause -- both cited
@@ -51,8 +52,14 @@ def _sections(segment_html):
         yield num_m.group(1), _norm(_strip(m.group(1))), _strip(body_html)
 
 
-def build():
-    html = fetch_cached(URL, RAW / "FullText.html").decode("utf-8", "replace")
+def parse(html):
+    """Chunks for every in-scope section of both Acts. Pure, for tests.
+
+    Justice Laws could re-shape this page the way CBSA re-shaped its D-memo
+    index, in which case the MarginalNote/Section markup stops matching and
+    this returns []; the write guard in build() is what keeps that from
+    emptying the corpus.
+    """
     # The page is both Acts in sequence; split at the 1982 Act's title so
     # section numbers (which restart) land in the right instrument.
     cut = html.find("CONSTITUTION ACT, 1982")
@@ -90,12 +97,23 @@ def build():
                 "citation": f"{short}, {label}",
                 "source_url": URL,
             })
-    out = PROCESSED_DIR / "charter.json"
-    out.write_text(json.dumps(chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
-    print(f"charter: {len(chunks)} sections -> {out.name}")
     return chunks
 
 
+def build(allow_shrink=False):
+    html = fetch_cached(URL, RAW / "FullText.html").decode("utf-8", "replace")
+    chunks = parse(html)
+    # indent=2 is what charter.json already carries -- changing it would
+    # rewrite every line of the file for nothing.
+    if not write_corpus(OUT_FILE, chunks, indent=2, allow_shrink=allow_shrink):
+        return False
+    print(f"charter: {len(chunks)} sections -> {OUT_FILE.name}")
+    return True
+
+
+def main():
+    sys.exit(0 if build(allow_shrink="--allow-shrink" in sys.argv) else 1)
+
+
 if __name__ == "__main__":
-    build()
+    main()

canlex/commentary.py

@@ -9,10 +9,12 @@ flag their reasoning as interpretation. The source of truth is
 data/curated/us_dispositions.json, which is reviewed by the user before it
 ships; this module just renders it into corpus chunks.
 
-    py -m canlex.commentary
+    py -m canlex.commentary [--allow-shrink]
 """
 import json
+import sys
 
+from ._common import write_corpus
 from .config import DATA_DIR, PROCESSED_DIR
 
 CURATED = DATA_DIR / "curated" / "us_dispositions.json"
@@ -105,7 +107,7 @@ def _entry_text(e, max_states=None):
     return "\n".join(lines)
 
 
-def build():
+def build(allow_shrink=False):
     data = json.loads(CURATED.read_text(encoding="utf-8"))
     chunks = []
     for e in data.get("methodology", []):
@@ -249,14 +251,22 @@ def build():
                 "source_url": "",
             })
 
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(chunks, ensure_ascii=False, indent=1),
-                   encoding="utf-8")
+    # A full re-render of the curated files, so a truncated or half-edited
+    # us_dispositions.json (or an equivalency file that moved) collapses the
+    # chunk count and the write would replace good analysis with the remnant.
+    # indent=1 is the stored file's format -- changing it rewrites every line.
+    if not write_corpus(OUT, chunks, indent=1, allow_shrink=allow_shrink):
+        return False
     print(f"{len(chunks)} commentary chunks "
           f"({len(data.get('dispositions', []))} dispositions, "
           f"{len(by_state)} state pages, {n_pairings} equivalency pairings, "
           f"{len(data.get('methodology', []))}+ methodology) -> {OUT}")
+    return True
+
+
+def main():
+    sys.exit(0 if build(allow_shrink="--allow-shrink" in sys.argv) else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

canlex/delegation.py

@@ -13,13 +13,14 @@ Sources:
     py -m canlex.delegation
 """
 import io
-import json
 import re
+import sys
 
 from bs4 import BeautifulSoup
 from pypdf import PdfReader
 
-from ._common import fetch_cached, norm_ws as _norm, uniquify_ids
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      stored_chunks, uniquify_ids, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 RAW = RAW_DIR / "delegation"
@@ -441,9 +442,25 @@ def parse_ircc(pdf_bytes, src):
     return chunks
 
 
-def build():
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of instruments this run could not build.
+
+    Ingestion is a full rebuild, so an instrument that fails its fetch -- or
+    that parses to nothing because CBSA changed the page's shape again -- would
+    silently vanish from the corpus. An instrument withdrawn on purpose is
+    removed from SOURCES instead, and so never lands in `failures`, which is
+    why preservation is restricted to the codes that actually failed here.
+    act_code is the per-instrument identity every parser stamps on its chunks.
+    Returns (chunks, preserved act_codes).
+    """
+    failed = {act_code for act_code, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["act_code"],
+                            only=failed)
+
+
+def build(allow_shrink=False):
     """Fetch, parse and chunk every delegation instrument into delegation.json."""
-    all_chunks = []
+    all_chunks, failures = [], []
     for src in SOURCES.values():
         print(f"Ingesting {src['act_short']} ...")
         try:
@@ -461,16 +478,29 @@ def build():
                 chunks = []
         except Exception as exc:
             print(f"  !! {src['act_short']}: {type(exc).__name__}: {exc}")
+            failures.append((src["act_code"], f"{type(exc).__name__}: {exc}"))
             continue
+        if not chunks:
+            failures.append((src["act_code"], "no content parsed"))
         all_chunks.extend(chunks)
         print(f"  {len(chunks)} chunks")
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
     uniquify_ids(all_chunks)
-    OUT.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=1),
-                   encoding="utf-8")
+
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(OUT))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} instrument(s): "
+              f"{', '.join(preserved)}")
+    if not write_corpus(OUT, all_chunks, indent=1, allow_shrink=allow_shrink):
+        return False
     print(f"\n{len(all_chunks)} delegation chunks from {len(SOURCES)} "
           f"instrument(s) -> {OUT}")
+    return True
+
+
+def main():
+    sys.exit(0 if build(allow_shrink="--allow-shrink" in sys.argv) else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

canlex/directive.py

@@ -4,13 +4,13 @@ NJC directives are negotiated by employer and bargaining-agent representatives;
 their provisions form part of collective agreements (and the rate tables in
 their appendices apply too). Chunks are tagged doc_type="directive".
 """
-import json
 import re
 import sys
 
 from bs4 import BeautifulSoup
 
-from ._common import MONTHS, fetch_cached, norm_ws as _norm, split_lines
+from ._common import (MONTHS, fetch_cached, norm_ws as _norm, preserve_dropped,
+                      split_lines, stored_chunks, write_corpus)
 from .config import RAW_DIR, PROCESSED_DIR
 
 INDEX_URL = "https://www.njc-cnm.gc.ca/directive/en"
@@ -160,10 +160,23 @@ def _print_link(html):
     return None
 
 
-def main():
-    force = "--force" in sys.argv
-    limit = next((int(a.split("=", 1)[1]) for a in sys.argv[1:]
-                  if a.startswith("--limit=")), None)
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of directives this run could not parse.
+
+    Ingestion is a full rebuild, so a directive whose page 404s or is restyled
+    vanishes from the corpus entirely -- and a directive that is still in force
+    is better served from the copy we already had than not at all. Keyed on
+    act_code (the /directive/<code>/ slug), which survives the title rewordings
+    the NJC index goes through; restricted to the directives that actually
+    failed, so any the NJC has genuinely retired stay gone.
+    Returns (chunks, preserved directive codes).
+    """
+    failed = {code for code, _title, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["act_code"],
+                            only=failed)
+
+
+def ingest(force=False, limit=None, allow_shrink=False):
     directives = directive_links(force=force)
     if limit:
         directives = directives[:limit]
@@ -187,16 +200,35 @@ def main():
                 all_chunks.extend(chunks)
                 print(f"  {title}: {len(chunks)} chunks")
             else:
-                failures.append((title, "no content parsed"))
+                failures.append((code, title, "no content parsed"))
         except Exception as exc:
-            failures.append((title, f"{type(exc).__name__}: {exc}"))
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT_FILE.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=2),
-                        encoding="utf-8")
+            failures.append((code, title, f"{type(exc).__name__}: {exc}"))
+    if not limit:
+        all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                                stored_chunks(OUT_FILE))
+        if preserved:
+            print(f"  kept the last-good copy of {len(preserved)} unparseable "
+                  f"directive(s): {', '.join(preserved)}")
+    # Before the write, so a refusal does not bury why the run came up short.
+    for _code, title, why in failures:
+        print(f"  FAILED {title}: {why}")
+    # The guard applies to --limit runs too: a smoke test over two directives
+    # must not be able to write a two-directive corpus over the real one.
+    if not write_corpus(OUT_FILE, all_chunks, indent=2,
+                        allow_shrink=allow_shrink):
+        return False
     print(f"\n{len(all_chunks)} chunks from {len(directives) - len(failures)} "
           f"directives -> {OUT_FILE.name}")
-    for title, why in failures:
-        print(f"  FAILED {title}: {why}")
+    return True
+
+
+def main():
+    force = "--force" in sys.argv
+    limit = next((int(a.split("=", 1)[1]) for a in sys.argv[1:]
+                  if a.startswith("--limit=")), None)
+    ok = ingest(force=force, limit=limit,
+                allow_shrink="--allow-shrink" in sys.argv)
+    sys.exit(0 if ok else 1)
 
 
 if __name__ == "__main__":

canlex/dmemo.py

@@ -13,7 +13,9 @@ from urllib.parse import urljoin
 from bs4 import BeautifulSoup
 from pypdf import PdfReader
 
-from ._common import POLITE_UA, fetch_cached, norm_ws as _norm, split_lines
+from ._common import (POLITE_UA, fetch_cached, norm_ws as _norm,
+                      preserve_dropped, split_lines, stored_chunks,
+                      write_corpus)
 from .config import RAW_DIR, PROCESSED_DIR
 
 INDEX_URL = "https://www.cbsa-asfc.gc.ca/publications/dm-md/d1-d23-eng.html"
@@ -35,12 +37,6 @@ _MEMO_HREF = re.compile(r"/dm-md/d\d+/d[\d-]+-eng\.html")
 _URL_NUMBER = re.compile(r"/(d\d+-[\d-]+)-eng\.html")
 _DS_HREF = re.compile(r'href="([^"]+)"')
 
-# Refuse to replace the corpus with less than this share of what it already
-# holds. When the index broke, the link-scrape returned zero memos and the
-# ingester happily wrote an empty dmemos.json over 1,870 good chunks; a single
-# stdout line was the only warning.
-_SHRINK_GUARD = 0.8
-
 
 def _fetch(url, dest, force=False):
     return fetch_cached(url, dest, ua=POLITE_UA, force=force)
@@ -245,24 +241,6 @@ def parse_pdf_memo(html, url):
     return chunks
 
 
-def safe_to_write(new_count, old_count, ratio=_SHRINK_GUARD):
-    """False when writing `new_count` chunks would gut an existing corpus.
-
-    Pure, for tests. An upstream shape change shows up as a collapse in chunk
-    count, and the write is the irreversible step -- so it is the one to gate.
-    """
-    return old_count == 0 or new_count >= old_count * ratio
-
-
-def _stored_chunks():
-    if not OUT_FILE.exists():
-        return []
-    try:
-        return json.loads(OUT_FILE.read_text(encoding="utf-8"))
-    except (ValueError, OSError):
-        return []
-
-
 def preserve_failed(chunks, failures, stored):
     """Re-attach the last-good chunks of memos that could not be fetched.
 
@@ -270,23 +248,17 @@ def preserve_failed(chunks, failures, stored):
     the corpus. CBSA's rebuilt index (2026-07) lists several memos as Active
     whose pages 404, and a transient network error looks the same -- in both
     cases dropping guidance that is still in force is worse than serving the
-    copy we already had. Cancelled memos never reach here: they are filtered
-    out before fetching. Returns (chunks, preserved memo numbers).
+    copy we already had. Restricted to memos that actually failed, so the
+    cancelled ones (filtered out before fetching) are not resurrected.
+    Returns (chunks, preserved memo numbers).
     """
-    have = {c["section"].upper() for c in chunks}
-    by_memo = {}
-    for chunk in stored:
-        by_memo.setdefault(chunk["section"].upper(), []).append(chunk)
-    preserved = []
+    failed = set()
     for url, _why in failures:
         match = _URL_NUMBER.search(url)
-        number = match.group(1).upper() if match else ""
-        if number and number not in have and number in by_memo:
-            have.add(number)
-            preserved.append(number)
-    for number in sorted(preserved):
-        chunks.extend(by_memo[number])
-    return chunks, sorted(preserved)
+        if match:
+            failed.add(match.group(1).upper())
+    return preserve_dropped(chunks, stored,
+                            key=lambda c: c["section"].upper(), only=failed)
 
 
 def ingest(force=False, limit=None, allow_shrink=False):
@@ -312,23 +284,16 @@ def ingest(force=False, limit=None, allow_shrink=False):
         for url, why in failures[:15]:
             print(f"    - {url.rsplit('/', 1)[-1]}: {why}")
 
-    stored = _stored_chunks()
     if not limit:
-        all_chunks, preserved = preserve_failed(all_chunks, failures, stored)
+        all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                                stored_chunks(OUT_FILE))
         if preserved:
             print(f"  kept the last-good copy of {len(preserved)} unfetchable "
                   f"memo(s): {', '.join(preserved)}")
     # The guard applies to --limit runs too: a smoke test over five memos must
     # not be able to write a five-memo corpus over the real one.
-    if not allow_shrink and not safe_to_write(len(all_chunks), len(stored)):
-        print(f"  REFUSING to write: {len(all_chunks)} chunks would replace "
-              f"{len(stored)} already in {OUT_FILE.name}. The upstream index or "
-              f"memo pages likely changed shape -- check the scrape before "
-              f"writing (re-run with --allow-shrink once the drop is understood).")
+    if not write_corpus(OUT_FILE, all_chunks, allow_shrink=allow_shrink):
         return False
-
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT_FILE.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=2), encoding="utf-8")
     print(f"  {len(all_chunks)} section-chunks from {len(urls) - len(failures)} memos "
           f"-> {OUT_FILE.name}")
     return True

canlex/enf.py

@@ -13,12 +13,14 @@ retires chapters, so anything that 404s is skipped with a note. doc_type is
 'memorandum' -- operational guidance, persuasive not binding, exactly like
 the D-memoranda.
 
-    python -m canlex.enf
+    python -m canlex.enf [--allow-shrink]
 """
-import json
 import re
+import sys
+import urllib.error
 
-from ._common import fetch_cached, norm_ws as _norm, split_lines, uniquify_ids
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      split_lines, stored_chunks, uniquify_ids, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 BASE = ("https://www.canada.ca/content/dam/ircc/migration/ircc/english/"
@@ -169,8 +171,50 @@ def _norm_block(lines):
     return "\n".join(p for p in (_norm(l) for l in lines) if p)
 
 
-def build(chapters=_CHAPTERS):
-    all_chunks, skipped = [], []
+_HTTP_404 = re.compile(r"\b404\b")
+
+
+def is_missing_upstream(exc):
+    """True when a fetch failed because the chapter simply is not there (404).
+
+    _CHAPTERS is a blind range(1, 41) probe, so on a healthy run most numbers
+    404 -- and a chapter IRCC RETIRES starts 404ing too. That is this module's
+    documented signal to drop a chapter, so a 404 must not be preserved
+    through: doing so would pin every withdrawn chapter in the corpus forever.
+    Every other fetch error (TLS, timeout, a dropped connection) is a real
+    failure the stored copy should survive. The two fetch paths report
+    differently -- urllib raises HTTPError with a status, the PowerShell
+    fallback raises CalledProcessError with the status only in its stderr
+    prose ("The remote server returned an error: (404) Not Found."), so that
+    side is a text match.
+    """
+    if isinstance(exc, urllib.error.HTTPError):
+        return exc.code == 404
+    stderr = getattr(exc, "stderr", "") or ""
+    if isinstance(stderr, bytes):
+        stderr = stderr.decode("utf-8", "replace")
+    return bool(_HTTP_404.search(stderr))
+
+
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of chapters this run could not produce.
+
+    A chapter that stops fetching for any reason OTHER than a 404 -- a dropped
+    network, a TLS refusal, a PDF that suddenly has no text layer -- has not
+    been retired, it just did not arrive, and losing guidance that is still in
+    force is the worse error. So the stored copy stays and the run says so
+    loudly enough to be investigated. Chapters that 404 (retired, or never
+    existed) never enter `failures` and are not resurrected.
+    Returns (chunks, preserved act_codes).
+    """
+    failed = {f"ENF-{n}" for n, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["act_code"],
+                            only=failed)
+
+
+def build(chapters=_CHAPTERS, allow_shrink=False):
+    # Returns the chunks written, or False if the write was refused.
+    all_chunks, skipped, failures = [], [], []
     for n in chapters:
         unofficial = UNOFFICIAL.get(n)
         url = (unofficial["url"] if unofficial
@@ -180,8 +224,12 @@ def build(chapters=_CHAPTERS):
             # The ATIP host is an ordinary WordPress site; canada.ca needs
             # the PowerShell TLS workaround.
             fetch_cached(url, dest, powershell="canada.ca" in url, pause=1.0)
-        except Exception:
+        except Exception as exc:
             skipped.append(n)
+            if not is_missing_upstream(exc):
+                # Not a 404: the chapter may well still be published, so it
+                # counts as a failure and its stored chunks are preserved.
+                failures.append((n, f"fetch failed: {type(exc).__name__}: {exc}"))
             if dest.exists():
                 dest.unlink()      # don't cache an error page as a PDF
             continue
@@ -198,10 +246,19 @@ def build(chapters=_CHAPTERS):
                 else:
                     print(f"  !! ENF {n}: no text layer and no OCR cache "
                           f"({ocr.name}); skipping")
+                    failures.append((n, f"no text layer and no {ocr.name}"))
                     continue
             chunks = _chunks_for(n, pages, url)
+            if not chunks:
+                # Fetched fine, parsed to nothing: the PDF's heading structure
+                # changed under us. That is the shape the D-memo loss took, so
+                # treat it as a failure and keep the stored chapter.
+                print(f"  !! ENF {n}: no chunks parsed; keeping stored copy")
+                failures.append((n, "no content parsed"))
+                continue
         except Exception as exc:
             print(f"  !! ENF {n}: parse failed: {type(exc).__name__}: {exc}")
+            failures.append((n, f"parse failed: {type(exc).__name__}: {exc}"))
             continue
         if unofficial:
             for c in chunks:
@@ -214,12 +271,25 @@ def build(chapters=_CHAPTERS):
         all_chunks.extend(chunks)
     uniquify_ids(all_chunks)
     out = PROCESSED_DIR / "enf.json"
-    out.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
+    # Preserved chunks keep the ids they were stored with, so they are added
+    # after uniquify_ids -- ids embed the chapter number, and a preserved
+    # chapter is by definition absent from this run's chunks.
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(out))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} chapter(s) "
+              f"that failed this run: {', '.join(preserved)}")
+    if not write_corpus(out, all_chunks, indent=2, allow_shrink=allow_shrink):
+        return False
     print(f"enf: {len(all_chunks)} chunks -> {out.name} "
           f"(chapters absent upstream: {skipped})")
     return all_chunks
 
 
+def main():
+    chunks = build(allow_shrink="--allow-shrink" in sys.argv)
+    sys.exit(1 if chunks is False else 0)
+
+
 if __name__ == "__main__":
-    build()
+    main()

canlex/ingest.py

@@ -1,12 +1,11 @@
 """Ingest Justice Laws XML into structured, section-level JSON chunks."""
-import json
 import re
 import sys
 import time
 import urllib.request
 import xml.etree.ElementTree as ET
 
-from ._common import norm_ws as _norm, split_lines
+from ._common import norm_ws as _norm, split_lines, write_corpus
 from .config import SOURCES, RAW_DIR, PROCESSED_DIR
 
 LIMS = "{http://justice.gc.ca/lims}"
@@ -388,13 +387,15 @@ def _nif_chunks(root, code, src, current_to):
     return chunks
 
 
-def ingest(code, force=False):
+def ingest(code, force=False, allow_shrink=False):
     print(f"Ingesting {code} ({SOURCES[code]['short']})...")
     xml_path = fetch_xml(code, force=force)
     chunks = parse_legislation(xml_path, code)
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
     out = PROCESSED_DIR / f"{code}.json"
-    out.write_text(json.dumps(chunks, ensure_ascii=False, indent=2), encoding="utf-8")
+    # An Act whose XML schema shifts parses to a fraction of its sections; the
+    # guard turns that into a refusal instead of a quietly gutted statute.
+    if not write_corpus(out, chunks, indent=2, allow_shrink=allow_shrink):
+        raise RuntimeError(f"{code}: refused to write a collapsed corpus")
     print(f"  {len(chunks)} sections -> {out.name}")
     return chunks
 
@@ -430,7 +431,7 @@ def _frenchify(chunks, code):
     return out
 
 
-def ingest_fr(code, force=False):
+def ingest_fr(code, force=False, allow_shrink=False):
     src = SOURCES[code]
     dest = RAW_DIR / f"{code}-fr.xml"
     if not dest.exists() or force:
@@ -442,8 +443,8 @@ def ingest_fr(code, force=False):
         time.sleep(1.0)
     chunks = _frenchify(parse_legislation(dest, code), code)
     out = PROCESSED_DIR / f"{code}-fr.json"
-    out.write_text(json.dumps(chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
+    if not write_corpus(out, chunks, indent=2, allow_shrink=allow_shrink):
+        raise RuntimeError(f"{code}: refused to write a collapsed corpus")
     print(f"  {len(chunks)} articles -> {out.name}")
     return chunks
 
@@ -451,6 +452,7 @@ def ingest_fr(code, force=False):
 def main():
     force = "--force" in sys.argv
     french = "--fr" in sys.argv
+    allow_shrink = "--allow-shrink" in sys.argv
     only = [a for a in sys.argv[1:] if not a.startswith("-")]
     codes = only or list(SOURCES)
     failures = []
@@ -461,15 +463,18 @@ def main():
         try:
             if french:
                 print(f"Ingesting {code} (français)...")
-                ingest_fr(code, force=force)
+                ingest_fr(code, force=force, allow_shrink=allow_shrink)
             else:
-                ingest(code, force=force)
+                ingest(code, force=force, allow_shrink=allow_shrink)
         except Exception as exc:
             failures.append((code, exc))
             print(f"  FAILED {code}: {type(exc).__name__}: {exc}")
     print(f"\nDone: {len(codes) - len(failures)}/{len(codes)} ingested.")
     for code, exc in failures:
         print(f"  FAILED {code}: {exc}")
+    # One collapsed Act must not pass as a clean run: refresh.py and the
+    # RUNBOOK chain this into `&& py -m canlex.embed`.
+    sys.exit(1 if failures else 0)
 
 
 if __name__ == "__main__":

canlex/irb_guidelines.py

@@ -12,12 +12,13 @@ parses for jurisprudential guides).
 
     python -m canlex.irb_guidelines
 """
-import json
 import re
+import sys
 
 from bs4 import BeautifulSoup
 
-from ._common import fetch_cached, norm_ws as _norm, split_lines, uniquify_ids
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      split_lines, stored_chunks, uniquify_ids, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 RAW = RAW_DIR / "irb_guidelines"
@@ -72,8 +73,8 @@ def _sections(main):
             yield label, title, body
 
 
-def build():
-    chunks = []
+def build(allow_shrink=False):
+    chunks, failures = [], []
     for g in GUIDELINES:
         try:
             html = fetch_cached(g["url"], RAW / f"{g['file']}.html",
@@ -81,6 +82,7 @@ def build():
         except Exception as exc:
             print(f"  !! Guideline {g['num']}: fetch failed "
                   f"({type(exc).__name__}: {exc})")
+            failures.append(f"IRB-G{g['num']}")
             continue
         soup = BeautifulSoup(html, "html.parser")
         main = soup.find("main") or soup
@@ -114,15 +116,34 @@ def build():
                     "source_url": g["url"],
                 })
                 n += 1
+        if not n:
+            # Fetched, parsed to nothing: the IRB page template moved.
+            print(f"  !! Guideline {g['num']}: no content parsed; "
+                  f"keeping stored copy")
+            failures.append(f"IRB-G{g['num']}")
+            continue
         print(f"  Guideline {g['num']} ({g['title'][:40]}): {n} chunks, "
               f"effective {effective or '?'}")
-    uniquify_ids(chunks)
     out = PROCESSED_DIR / "irb_guidelines.json"
-    out.write_text(json.dumps(chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
+    chunks, preserved = preserve_dropped(chunks, stored_chunks(out),
+                                         key=lambda c: c["act_code"],
+                                         only=set(failures))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} guideline(s) "
+              f"that failed this run: {', '.join(preserved)}")
+    uniquify_ids(chunks)
+    if not write_corpus(out, chunks, indent=2, allow_shrink=allow_shrink):
+        return False
     print(f"irb_guidelines: {len(chunks)} chunks -> {out.name}")
     return chunks
 
 
+def main():
+    # `is not False` rather than truthiness: a first-ever run that writes an
+    # empty corpus succeeded, a refused write did not.
+    ok = build(allow_shrink="--allow-shrink" in sys.argv) is not False
+    sys.exit(0 if ok else 1)
+
+
 if __name__ == "__main__":
-    build()
+    main()

canlex/pdi.py

@@ -7,15 +7,17 @@ Tran serious-criminality page (partial successor to ENF 2). Each named
 index page is fetched, its same-section sub-pages discovered, and every
 page chunked by heading -- doc_type='memorandum', like all guidance.
 
-    python -m canlex.pdi
+    python -m canlex.pdi [--allow-shrink]
 """
-import json
 import re
+import sys
 
-from ._common import fetch_cached, norm_ws as _norm, split_lines, uniquify_ids
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      split_lines, stored_chunks, uniquify_ids, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 RAW = RAW_DIR / "pdi"
+OUT_FILE = PROCESSED_DIR / "pdi.json"
 _OBM = ("https://www.canada.ca/en/immigration-refugees-citizenship/corporate/"
         "publications-manuals/operational-bulletins-manuals/")
 
@@ -98,8 +100,25 @@ def _page_chunks(src, page_url, html, page_no):
     return chunks
 
 
-def build():
-    all_chunks = []
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of sub-pages this run could not fetch.
+
+    Ingestion is a full rebuild, so a sub-page that 404s or drops its
+    connection silently vanishes from the corpus; serving the copy we already
+    had beats losing instructions that are still in force. Keyed on source_url,
+    the only per-page identity a PDI chunk carries that survives pages coming
+    and going -- the chunk id embeds the page's position in the index listing,
+    which shifts whenever one page drops out. Restricted to pages that actually
+    errored, so a sub-page IRCC retired on purpose is not resurrected.
+    Returns (chunks, preserved page URLs).
+    """
+    return preserve_dropped(chunks, stored, key=lambda c: c["source_url"],
+                            only={url for url, _why in failures})
+
+
+def build(allow_shrink=False):
+    """Rebuild pdi.json. Returns the chunks, or False if the write was refused."""
+    all_chunks, failures = [], []
     for src in SETS:
         index_html = fetch_cached(
             src["index"], RAW / f"{src['code']}-index.html",
@@ -122,20 +141,46 @@ def build():
                         powershell=True, pause=0.8).decode("utf-8", "replace")
                     pages.append((url, page))
                 except Exception as exc:
+                    failures.append((url, f"{type(exc).__name__}: {exc}"))
                     print(f"  !! {name}: {type(exc).__name__}: {exc}")
         n = 0
         for page_no, (url, html) in enumerate(pages):
             chunks = _page_chunks(src, url, html, page_no)
+            if not chunks:
+                # Downloaded fine, parsed to nothing: the page template moved
+                # under us. That is exactly how the D-memo corpus was lost, so
+                # it counts as a failure and the stored page is kept.
+                print(f"  !! {url.rsplit('/', 1)[-1]}: no content parsed; "
+                      f"keeping stored copy")
+                failures.append((url, "no content parsed"))
+                continue
             all_chunks.extend(chunks)
             n += len(chunks)
         print(f"  {src['short']}: {len(pages)} pages, {n} chunks")
+
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(OUT_FILE))
+    if preserved:
+        names = ", ".join(u.rsplit("/", 1)[-1] for u in preserved)
+        print(f"  kept the last-good copy of {len(preserved)} unfetchable "
+              f"page(s): {names}")
+    # After preservation, not before: a preserved page brings back the ids it
+    # was written with, and this run may have handed those ids to other pages
+    # (an id embeds the page's position in the index listing).
     uniquify_ids(all_chunks)
-    out = PROCESSED_DIR / "pdi.json"
-    out.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=2),
-                   encoding="utf-8")
-    print(f"pdi: {len(all_chunks)} chunks -> {out.name}")
+    if not write_corpus(OUT_FILE, all_chunks, indent=2,
+                        allow_shrink=allow_shrink):
+        return False
+    print(f"pdi: {len(all_chunks)} chunks -> {OUT_FILE.name}")
     return all_chunks
 
 
+def main():
+    # `is not False` rather than truthiness: a first-ever run that writes an
+    # empty corpus succeeded, a refused write did not.
+    ok = build(allow_shrink="--allow-shrink" in sys.argv) is not False
+    sys.exit(0 if ok else 1)
+
+
 if __name__ == "__main__":
-    build()
+    main()

canlex/tariff_schedule.py

@@ -20,15 +20,16 @@ per chapter for its Notes and Subheading Notes -- a heading is the natural
 unit of legal classification (the eight- and ten-digit items below it are the
 same rule with finer rate granularity).
 
-    py -m canlex.tariff_schedule
+    py -m canlex.tariff_schedule [--allow-shrink]
 """
-import json
 import re
+import sys
 from collections import defaultdict
 
 from bs4 import BeautifulSoup
 
-from ._common import fetch_cached, norm_ws as _norm
+from ._common import (fetch_cached, norm_ws as _norm, preserve_dropped,
+                      stored_chunks, write_corpus)
 from .config import PROCESSED_DIR, RAW_DIR
 
 RAW = RAW_DIR / "tariff_schedule"
@@ -73,6 +74,13 @@ def _heading_of(item):
     return f"{digits[:2]}.{digits[2:4]}"
 
 
+def _chapter_part(chapter):
+    """The `part` every chunk of a chapter carries -- and, because nothing else
+    on a chunk names its chapter, the per-chapter identity preserve_failed
+    matches on. Derived in one place so the two stay in step."""
+    return f"Schedule, Chapter {chapter}"
+
+
 def _collect_notes(main):
     """Return the chapter's Notes + Subheading Notes as a single text block."""
     out = []
@@ -120,7 +128,7 @@ def parse_chapter(html, src):
             "section": f"Sch-Ch{chapter}-Notes",
             "marginal_note": (f"Chapter {chapter} Notes — "
                               f"{src['title']}"),
-            "part": f"Schedule, Chapter {chapter}",
+            "part": _chapter_part(chapter),
             "division": "",
             "heading": src["title"],
             "text": notes_body,
@@ -207,7 +215,7 @@ def parse_chapter(html, src):
             "act_name": "Customs Tariff",
             "section": f"Sch-{heading}",
             "marginal_note": desc[:200],
-            "part": f"Schedule, Chapter {chapter}",
+            "part": _chapter_part(chapter),
             "division": "",
             "heading": src["title"],
             "text": body,
@@ -221,8 +229,23 @@ def parse_chapter(html, src):
     return chunks
 
 
-def build():
-    all_chunks = []
+def preserve_failed(chunks, failures, stored):
+    """Re-attach the last-good chunks of chapters this run could not rebuild.
+
+    Only two chapters make up this corpus, so losing one is losing a third to
+    two thirds of it -- and a CBSA page that has been reshaped (or a fetch that
+    died mid-run) is indistinguishable from a chapter that legitimately
+    vanished until a human looks. Serving last week's copy of ch. 98 beats
+    serving no traveller exemptions at all. Restricted to the chapters that
+    actually failed. Returns (chunks, preserved chapter identities).
+    """
+    failed = {_chapter_part(chapter) for chapter, _why in failures}
+    return preserve_dropped(chunks, stored, key=lambda c: c["part"],
+                            only=failed)
+
+
+def build(allow_shrink=False):
+    all_chunks, failures = [], []
     for src in SOURCES.values():
         print(f"Ingesting Customs Tariff Schedule {src['code']} ...")
         try:
@@ -230,15 +253,33 @@ def build():
             chunks = parse_chapter(html, src)
         except Exception as exc:
             print(f"  !! {src['code']}: {type(exc).__name__}: {exc}")
+            failures.append((src["chapter"], f"{type(exc).__name__}: {exc}"))
+            continue
+        if not chunks:
+            # A chapter never parses to nothing legitimately -- an empty result
+            # means the page moved or changed shape, so treat it as a failure
+            # rather than letting it silently shrink the corpus.
+            print(f"  !! {src['code']}: no chunks parsed")
+            failures.append((src["chapter"], "no chunks parsed"))
             continue
         all_chunks.extend(chunks)
         print(f"  {len(chunks)} chunks")
-    PROCESSED_DIR.mkdir(parents=True, exist_ok=True)
-    OUT.write_text(json.dumps(all_chunks, ensure_ascii=False, indent=1),
-                   encoding="utf-8")
-    print(f"\n{len(all_chunks)} tariff-schedule chunks from {len(SOURCES)} "
-          f"chapter(s) -> {OUT}")
+
+    all_chunks, preserved = preserve_failed(all_chunks, failures,
+                                            stored_chunks(OUT))
+    if preserved:
+        print(f"  kept the last-good copy of {len(preserved)} chapter(s): "
+              f"{', '.join(preserved)}")
+    if not write_corpus(OUT, all_chunks, indent=1, allow_shrink=allow_shrink):
+        return False
+    print(f"\n{len(all_chunks)} tariff-schedule chunks from "
+          f"{len(SOURCES) - len(failures)} chapter(s) -> {OUT}")
+    return True
+
+
+def main():
+    sys.exit(0 if build(allow_shrink="--allow-shrink" in sys.argv) else 1)
 
 
 if __name__ == "__main__":
-    build()
+    main()

tests/test_agreement.py

@@ -0,0 +1,78 @@
+"""Unit tests for the collective-agreement ingester (canlex/agreement.py).
+
+Offline only. Ingestion is a full rebuild, so the failure that cost the D-memo
+corpus on 2026-07-27 -- an upstream page that changes shape, scrapes to nothing
+and is written straight over the good chunks -- applies here too: canada.ca is
+one redesign away from parsing to zero Articles.
+"""
+import unittest
+
+from canlex import agreement
+
+
+class ParseAgreementTests(unittest.TestCase):
+    def test_redesigned_page_parses_to_nothing(self):
+        # What a layout change looks like: no <main>, so no Articles. ingest()
+        # records this as a failure rather than writing the empty result.
+        self.assertEqual(agreement.parse_agreement("<div>hello</div>", "FB"), [])
+
+    def test_article_heading_becomes_a_numbered_chunk(self):
+        html = ("<main><h3>Article 17: discipline</h3>"
+                "<p>An employee may be disciplined.</p></main>")
+        chunk, = agreement.parse_agreement(html, "FB")
+        self.assertEqual(chunk["section"], "17")
+        self.assertEqual(chunk["marginal_note"], "discipline")
+        self.assertEqual(chunk["act_code"], "FB")
+
+    def test_every_chunk_carries_its_agreement_code(self):
+        # preserve_failed keys on act_code, so it has to be on every chunk.
+        html = ("<main><h3>Article 1: purpose</h3><p>one</p>"
+                "<h3>Appendix A</h3><p>two</p></main>")
+        chunks = agreement.parse_agreement(html, "FB")
+        self.assertEqual({c["act_code"] for c in chunks}, {"FB"})
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"act_code": "FB", "text": "hours of work"},
+              {"act_code": "FB", "text": "overtime"},
+              {"act_code": "PA", "text": "leave"}]
+
+    def test_failed_agreement_keeps_its_last_good_chunks(self):
+        failures = [("FB", "no content parsed")]
+        chunks, preserved = agreement.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["FB"])
+        self.assertEqual([c["text"] for c in chunks],
+                         ["hours of work", "overtime"])
+
+    def test_fetch_error_counts_as_a_failure_too(self):
+        failures = [("FB", "URLError: timed out")]
+        _chunks, preserved = agreement.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["FB"])
+
+    def test_agreement_dropped_on_purpose_is_not_resurrected(self):
+        # PA is stored but no longer in AGREEMENTS, so it never failed: a
+        # retired agreement must not come back through the preserve path.
+        chunks, preserved = agreement.preserve_failed(
+            [], [("FB", "no content parsed")], self.STORED)
+        self.assertNotIn("PA", preserved)
+        self.assertNotIn("leave", [c["text"] for c in chunks])
+
+    def test_freshly_parsed_chunks_win_over_stored(self):
+        fresh = [{"act_code": "FB", "text": "new text"}]
+        chunks, preserved = agreement.preserve_failed(
+            fresh, [("FB", "no content parsed")], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_agreement_we_never_stored_cannot_be_preserved(self):
+        chunks, preserved = agreement.preserve_failed(
+            [], [("SV", "URLError: timed out")], self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_nothing_happens_without_failures(self):
+        fresh = [{"act_code": "FB", "text": "x"}]
+        chunks, preserved = agreement.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_amps.py

@@ -0,0 +1,172 @@
+"""Unit tests for the AMPS Master Penalty Document ingester (canlex/amps.py).
+
+Offline only. The ingester is a full rebuild, so a run that loses its network
+half way through -- or an index CBSA rebuilds into something the link regex no
+longer matches -- used to write the shortfall straight over the stored corpus.
+These cover the module's own half of the fix: which contraventions count as
+"failed this run", and that the identity used to preserve them is the same one
+parse_page stamps on a chunk.
+"""
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from canlex import amps
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"section": "C005", "text": "records not kept"},
+              {"section": "C348", "text": "no advance data"}]
+
+    def test_unfetchable_contravention_keeps_its_last_good_chunk(self):
+        chunks, preserved = amps.preserve_failed(
+            [], [("c005", "ValueError: empty response")], self.STORED)
+        self.assertEqual(preserved, ["C005"])
+        self.assertEqual([c["text"] for c in chunks], ["records not kept"])
+
+    def test_index_codes_are_lower_case_but_sections_are_not(self):
+        # The whole preserve step turns on this: build() collects the index's
+        # lower-case c-number, a chunk carries it upper-cased.
+        _chunks, preserved = amps.preserve_failed([], [("c348", "404")],
+                                                  self.STORED)
+        self.assertEqual(preserved, ["C348"])
+
+    def test_contravention_dropped_upstream_is_not_resurrected(self):
+        # C348 vanished from the index rather than failing; it stays gone.
+        chunks, preserved = amps.preserve_failed([], [("c005", "404")],
+                                                 self.STORED)
+        self.assertEqual(preserved, ["C005"])
+        self.assertNotIn("C348", [c["section"] for c in chunks])
+
+    def test_freshly_scraped_chunk_wins_over_the_stored_copy(self):
+        fresh = [{"section": "C005", "text": "new penalty amounts"}]
+        chunks, preserved = amps.preserve_failed(fresh, [("c005", "404")],
+                                                 self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_contravention_we_never_had_cannot_be_preserved(self):
+        chunks, preserved = amps.preserve_failed([], [("c999", "404")],
+                                                 self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_a_clean_run_preserves_nothing(self):
+        fresh = [{"section": "C005", "text": "x"}]
+        chunks, preserved = amps.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_total_scrape_failure_restores_the_whole_corpus(self):
+        # Every page erroring is the shape of the outage the guard is for: the
+        # stored corpus survives intact instead of being written away.
+        failed = [("c005", "timeout"), ("c348", "timeout")]
+        chunks, preserved = amps.preserve_failed([], failed, self.STORED)
+        self.assertEqual(preserved, ["C005", "C348"])
+        self.assertEqual(len(chunks), len(self.STORED))
+
+
+class ChunkIdentityTests(unittest.TestCase):
+    """preserve_failed keys on `section`; parse_page must keep filling it."""
+
+    HTML = ("<main><h1>Administrative Monetary Penalty C005</h1>"
+            "<p>Person failed to keep the prescribed records for the "
+            "prescribed period.</p>"
+            "<table><tr><td>First</td><td>$500</td></tr></table></main>"
+            '<time property="dateModified">2026-05-01</time>')
+
+    def test_parsed_section_is_the_preserve_key(self):
+        chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
+        self.assertEqual(chunk["section"], "C005")
+        _chunks, preserved = amps.preserve_failed([], [("c005", "404")],
+                                                  [chunk])
+        self.assertEqual(preserved, ["C005"])
+
+    def test_one_chunk_per_contravention_so_the_key_is_unique(self):
+        chunk = amps.parse_page(self.HTML, "c005", "https://x/c005-eng.html")
+        self.assertEqual(chunk["id"], "amps-c005")
+
+
+class BuildWiringTests(unittest.TestCase):
+    """build() with the network stubbed out: is the guard actually reached?"""
+
+    INDEX = ('<a href="/trade-commerce/amps/contraventions-infractions/'
+             'c005-eng.html">C005</a>')
+    PAGE = ("<main><h1>Administrative Monetary Penalty C005</h1>"
+            "<p>Person failed to keep the prescribed records for the "
+            "prescribed period.</p></main>")
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.dir = Path(tmp.name)
+        self.out = self.dir / "amps.json"
+
+    def _build(self, index_html, **kwargs):
+        def stub(url, _dest):
+            return (index_html if url == amps.INDEX_URL
+                    else self.PAGE).encode("utf-8")
+        with mock.patch.object(amps, "PROCESSED_DIR", self.dir), \
+                mock.patch.object(amps, "_fetch", side_effect=stub), \
+                contextlib.redirect_stdout(io.StringIO()) as out:
+            return amps.build(**kwargs), out.getvalue()
+
+    def test_an_index_that_scrapes_to_nothing_cannot_empty_the_corpus(self):
+        # CBSA rebuilding the MPD index into something _LINK no longer matches.
+        self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
+                                        for i in range(50)]), encoding="utf-8")
+        ok, log = self._build("<table id='mpd'></table>")
+        self.assertFalse(ok)
+        self.assertIn("REFUSING", log)
+        self.assertEqual(
+            len(json.loads(self.out.read_text(encoding="utf-8"))), 50)
+
+    def test_allow_shrink_lets_an_understood_drop_through(self):
+        self.out.write_text(json.dumps([{"section": f"C{i:03d}"}
+                                        for i in range(50)]), encoding="utf-8")
+        ok, _log = self._build("<table id='mpd'></table>", allow_shrink=True)
+        self.assertTrue(ok)
+        self.assertEqual(json.loads(self.out.read_text(encoding="utf-8")), [])
+
+    def test_a_healthy_run_writes_two_space_indent(self):
+        # The stored file is indent=2; writing any other width would rewrite
+        # every line of a 440KB data file as a spurious diff.
+        ok, log = self._build(self.INDEX)
+        self.assertTrue(ok)
+        self.assertTrue(self.out.read_text(encoding="utf-8").startswith('[\n  {'))
+        self.assertIn("1 contraventions -> amps.json", log)
+
+
+class MainExitCodeTests(unittest.TestCase):
+    def test_refused_write_exits_non_zero(self):
+        with mock.patch.object(amps, "build", return_value=False), \
+                mock.patch.object(amps.sys, "argv", ["amps"]):
+            with self.assertRaises(SystemExit) as caught:
+                amps.main()
+        self.assertEqual(caught.exception.code, 1)
+
+    def test_successful_run_exits_zero(self):
+        with mock.patch.object(amps, "build", return_value=True), \
+                mock.patch.object(amps.sys, "argv", ["amps"]):
+            with self.assertRaises(SystemExit) as caught:
+                amps.main()
+        self.assertEqual(caught.exception.code, 0)
+
+    def test_allow_shrink_flag_reaches_build(self):
+        with mock.patch.object(amps, "build", return_value=True) as build, \
+                mock.patch.object(amps.sys, "argv", ["amps", "--allow-shrink"]):
+            with self.assertRaises(SystemExit):
+                amps.main()
+        self.assertTrue(build.call_args.kwargs["allow_shrink"])
+
+    def test_allow_shrink_is_off_by_default(self):
+        with mock.patch.object(amps, "build", return_value=True) as build, \
+                mock.patch.object(amps.sys, "argv", ["amps"]):
+            with self.assertRaises(SystemExit):
+                amps.main()
+        self.assertFalse(build.call_args.kwargs["allow_shrink"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_benefits.py

@@ -1,5 +1,15 @@
-"""Unit tests for benefit-plan booklet ingestion (canlex/benefits.py)."""
+"""Unit tests for benefit-plan booklet ingestion (canlex/benefits.py).
+
+Offline only: the build tests patch out the fetches and point OUT at a
+tempfile, so the real corpus is never touched.
+"""
+import contextlib
+import io
+import json
+import tempfile
 import unittest
+from pathlib import Path
+from unittest import mock
 
 from canlex import benefits
 
@@ -75,5 +85,122 @@ class EmitTests(unittest.TestCase):
         self.assertIn("June 2024", c["citation"])
 
 
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"act_code": "PSHCP", "text": "drug benefit"},
+              {"act_code": "PSHCP", "text": "hospital benefit"},
+              {"act_code": "PSMIP", "text": "basic life insurance"}]
+
+    def test_failed_plan_keeps_its_last_good_chunks(self):
+        chunks, preserved = benefits.preserve_failed(
+            [], [("PSHCP", "HTTPError: 404")], self.STORED)
+        self.assertEqual(preserved, ["PSHCP"])
+        self.assertEqual(len(chunks), 2)
+
+    def test_freshly_parsed_chunks_win_over_stored(self):
+        fresh = [{"act_code": "PSHCP", "text": "reworded drug benefit"}]
+        chunks, preserved = benefits.preserve_failed(
+            fresh, [("PSHCP", "HTTPError: 404")], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_plan_dropped_on_purpose_is_not_resurrected(self):
+        # PSMIP is gone from SOURCES (plan wound up), not failing; only the
+        # plan that errored comes back.
+        chunks, preserved = benefits.preserve_failed(
+            [], [("PSHCP", "no sections parsed")], self.STORED)
+        self.assertEqual(preserved, ["PSHCP"])
+        self.assertNotIn("basic life insurance", [c["text"] for c in chunks])
+
+    def test_plan_we_never_stored_cannot_be_preserved(self):
+        chunks, preserved = benefits.preserve_failed(
+            [], [("PSDCP", "timeout")], self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_nothing_happens_without_failures(self):
+        fresh = [{"act_code": "PSDCP", "text": "x"}]
+        chunks, preserved = benefits.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+
+def _chunk(n, code="PSHCP"):
+    return {"id": f"benefits-{code}-{n}", "act_code": code, "text": f"body {n}"}
+
+
+class BuildGuardTests(unittest.TestCase):
+    """A booklet that stops parsing must not empty benefits.json."""
+
+    SRC = dict(SRC, toc_pages=[1], body_start=2)
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.out = Path(tmp.name) / "benefits.json"
+
+    def _store(self, chunks):
+        self.out.write_text(json.dumps(chunks), encoding="utf-8")
+
+    def _build(self, parser, allow_shrink=False):
+        """Run build() with one PDF plan parsed by `parser`; no network, no
+        real corpus. Returns (ok, stdout)."""
+        log = io.StringIO()
+        with mock.patch.object(benefits, "OUT", self.out), \
+             mock.patch.object(benefits, "SOURCES", {"PSHCP": self.SRC}), \
+             mock.patch.object(benefits, "_fetch", lambda url, dest: b"%PDF"), \
+             mock.patch.object(benefits, "parse_pdf_booklet", parser), \
+             contextlib.redirect_stdout(log):
+            ok = benefits.build(allow_shrink=allow_shrink)
+        return ok, log.getvalue()
+
+    def _written(self):
+        return json.loads(self.out.read_text(encoding="utf-8"))
+
+    def test_booklet_that_segments_to_nothing_counts_as_a_failure(self):
+        self._store([_chunk(i) for i in range(10)])
+        ok, log = self._build(lambda pdf, src: [])
+        self.assertTrue(ok)
+        self.assertIn("no sections parsed", log)
+        self.assertIn("kept the last-good copy", log)
+        self.assertEqual(len(self._written()), 10)
+
+    def test_fetch_error_preserves_the_plan(self):
+        self._store([_chunk(i) for i in range(10)])
+
+        def boom(pdf, src):
+            raise RuntimeError("PDF moved")
+
+        ok, log = self._build(boom)
+        self.assertTrue(ok)
+        self.assertIn("kept the last-good copy", log)
+        self.assertEqual(len(self._written()), 10)
+
+    def test_collapse_without_a_failure_is_refused(self):
+        # The booklet still parses, but to a fraction of what is stored -- a
+        # re-typeset PDF whose headings no longer match. Nothing failed, so
+        # preservation does not apply and the guard is the only backstop.
+        self._store([_chunk(i) for i in range(10)])
+        ok, log = self._build(lambda pdf, src: [_chunk(0), _chunk(1)])
+        self.assertFalse(ok)
+        self.assertIn("REFUSING", log)
+        self.assertEqual(len(self._written()), 10)
+
+    def test_allow_shrink_lets_an_understood_drop_through(self):
+        self._store([_chunk(i) for i in range(10)])
+        ok, _log = self._build(lambda pdf, src: [_chunk(0), _chunk(1)],
+                               allow_shrink=True)
+        self.assertTrue(ok)
+        self.assertEqual(len(self._written()), 2)
+
+    def test_first_ever_run_writes(self):
+        ok, _log = self._build(lambda pdf, src: [_chunk(0), _chunk(1)])
+        self.assertTrue(ok)
+        self.assertEqual(len(self._written()), 2)
+
+    def test_written_file_keeps_the_one_space_indent(self):
+        # benefits.json has always been written with indent=1; widening it
+        # would rewrite every line of the data file for nothing.
+        self._build(lambda pdf, src: [_chunk(0)])
+        self.assertEqual(self.out.read_text(encoding="utf-8").splitlines()[1],
+                         " {")
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_caselaw.py

@@ -0,0 +1,150 @@
+"""Unit tests for the case-law ingester (canlex/caselaw.py). Offline only.
+
+build() rebuilds caselaw.json from a fresh scrape of the Lexum decision
+databases, so one CAPTCHA wall or re-platforming makes every fetch fail at once
+and the write turns that into an emptied corpus -- the failure that cost the
+D-memos 1,870 chunks on 2026-07-27. These tests drive build() with the curated
+lists emptied (or pointed at a fake on-disk CanLII capture), so no network is
+touched; they cover the write guard, the on-disk indent, and main()'s exit code.
+"""
+import contextlib
+import io
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from canlex import caselaw
+
+
+def _run(**kwargs):
+    """Call build() with stdout captured; returns (result, printed text)."""
+    with contextlib.redirect_stdout(io.StringIO()) as out:
+        ok = caselaw.build(**kwargs)
+    return ok, out.getvalue()
+
+
+class WriteGuardTests(unittest.TestCase):
+    """Every curated list is empty here, so build() reaches the write without
+    fetching anything -- exactly the shape a scrape that lost every decision
+    would take."""
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.out = Path(tmp.name) / "caselaw.json"
+        for attr, value in (("OUT", self.out), ("CASES", []),
+                            ("CANLII_CASES", []), ("IRB_GUIDES", [])):
+            patch = mock.patch.object(caselaw, attr, value)
+            patch.start()
+            self.addCleanup(patch.stop)
+
+    def _store(self, count):
+        self.out.write_text(json.dumps([{"id": f"scc-{i}-1"}
+                                        for i in range(count)]),
+                            encoding="utf-8")
+
+    def test_first_ever_run_writes(self):
+        ok, _printed = _run()
+        self.assertTrue(ok)
+        self.assertEqual(json.loads(self.out.read_text(encoding="utf-8")), [])
+
+    def test_a_scrape_that_lost_every_decision_is_refused(self):
+        self._store(100)
+        ok, printed = _run()
+        self.assertFalse(ok)
+        self.assertIn("REFUSING", printed)
+        self.assertEqual(len(json.loads(self.out.read_text(encoding="utf-8"))),
+                         100)
+
+    def test_allow_shrink_overrides_the_refusal(self):
+        self._store(100)
+        ok, _printed = _run(allow_shrink=True)
+        self.assertTrue(ok)
+        self.assertEqual(json.loads(self.out.read_text(encoding="utf-8")), [])
+
+
+class WrittenFormatTests(unittest.TestCase):
+    """One fake CanLII capture: that branch reads cached page text off disk, so
+    a real chunk reaches the write with no fetch."""
+
+    CASE = {"file": "canlii-fake.txt",
+            "canlii_path": "en/ca/fca/doc/2020/2020fca1/2020fca1.html",
+            "short": "Fake", "name": "R v Fake", "cite": "2020 FCA 1",
+            "report": "", "date": "2020-01-01",
+            "court_name": "Federal Court of Appeal", "topic": "Testing"}
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        raw = Path(tmp.name) / "raw"
+        raw.mkdir()
+        (raw / self.CASE["file"]).write_text(
+            "R v Fake\nThe appeal is dismissed.\nCosts to the respondent.",
+            encoding="utf-8")
+        self.out = Path(tmp.name) / "caselaw.json"
+        for attr, value in (("OUT", self.out), ("_RAW", raw), ("CASES", []),
+                            ("CANLII_CASES", [self.CASE]), ("IRB_GUIDES", [])):
+            patch = mock.patch.object(caselaw, attr, value)
+            patch.start()
+            self.addCleanup(patch.stop)
+
+    def test_chunks_reach_the_file(self):
+        ok, _printed = _run()
+        self.assertTrue(ok)
+        chunks = json.loads(self.out.read_text(encoding="utf-8"))
+        self.assertEqual([c["id"] for c in chunks], ["canlii-2020fca1-1"])
+        self.assertEqual(chunks[0]["doc_type"], "caselaw")
+
+    def test_indent_stays_at_one_space(self):
+        # caselaw.json is stored with indent=1; writing it at any other indent
+        # would rewrite every line of a 20k-line file for no content change.
+        _run()
+        lines = self.out.read_text(encoding="utf-8").splitlines()
+        self.assertEqual(lines[0], "[")
+        self.assertEqual(lines[1], " {")
+
+    def test_non_ascii_is_not_escaped(self):
+        # Locators use an en dash, citations carry accents.
+        _run()
+        self.assertNotIn("\\u", self.out.read_text(encoding="utf-8"))
+
+
+class MainExitTests(unittest.TestCase):
+    def _main(self, argv, build_result):
+        seen = {}
+
+        def fake_build(allow_shrink=False):
+            seen["allow_shrink"] = allow_shrink
+            return build_result
+
+        with mock.patch.object(caselaw, "build", fake_build), \
+                mock.patch.object(sys, "argv", argv), \
+                contextlib.redirect_stdout(io.StringIO()):
+            with self.assertRaises(SystemExit) as raised:
+                caselaw.main()
+        return raised.exception.code, seen
+
+    def test_a_refused_write_exits_non_zero(self):
+        # The refresh runbook shells out to `py -m canlex.caselaw`; a silent
+        # zero exit is how a gutted corpus would sail past a scheduled run.
+        code, _seen = self._main(["caselaw"], False)
+        self.assertEqual(code, 1)
+
+    def test_a_written_corpus_exits_zero(self):
+        code, _seen = self._main(["caselaw"], True)
+        self.assertEqual(code, 0)
+
+    def test_allow_shrink_flag_is_forwarded(self):
+        _code, seen = self._main(["caselaw", "--allow-shrink"], True)
+        self.assertTrue(seen["allow_shrink"])
+
+    def test_guard_is_on_without_the_flag(self):
+        _code, seen = self._main(["caselaw"], True)
+        self.assertFalse(seen["allow_shrink"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_charter.py

@@ -0,0 +1,146 @@
+"""Unit tests for the Charter / Constitution ingester (canlex/charter.py).
+
+Offline only -- every test drives parse() with a fixture cut from the real
+Justice Laws markup. The point is the failure mode that emptied dmemos.json on
+2026-07-27: if the upstream page is re-shaped the scrape returns nothing, so
+parse() must be provably capable of returning [] and build() must refuse to
+write that over the stored corpus.
+"""
+import sys
+import unittest
+from unittest import mock
+
+from canlex import charter
+
+
+def _note(text):
+    return (f'<p class="MarginalNote"><span class="wb-invisible">Marginal '
+            f'note:</span>{text}</p>')
+
+
+def _section(num, body):
+    return (f'<p class="Section"><strong><a class="sectionLabel" id="s-{num}">'
+            f'<span class="sectionLabel">{num}</span></a></strong>\xa0{body}</p>')
+
+
+# Both Acts in sequence, as the page serves them. Section numbers restart at
+# the 1982 heading, so "1" appears in each -- only the 1982 one is in scope.
+FIXTURE = "".join([
+    "<h1>CONSTITUTION ACT, 1867</h1>",
+    _note("Short title"),
+    _section(1, "This Act may be cited as the Constitution Act, 1867."),
+    _note("Legislative Authority of Parliament of Canada"),
+    _section(91, "It shall be lawful for the Queen to make Laws for the "
+                 "Peace, Order, and good Government of Canada."),
+    _note("Subjects of exclusive Provincial Legislation"),
+    _section(92, "In each Province the Legislature may exclusively make Laws."),
+    "<h1>CONSTITUTION ACT, 1982</h1>",
+    _note("Rights and freedoms in Canada"),
+    _section(1, "The Canadian Charter of Rights and Freedoms guarantees the "
+                "rights and freedoms set out in it."),
+    _note("Fundamental freedoms"),
+    _section(2, "Everyone has the following fundamental freedoms:"),
+    '<ul class="ProvisionList"><li><p class="Paragraph">'
+    '<span class="lawlabel">(a)</span>\xa0freedom of conscience and religion;'
+    "</p></li></ul>",
+    _note("Detention or imprisonment"),
+    _section(9, "Everyone has the right not to be arbitrarily detained."),
+    _note("Commitment to promote equal opportunities"),
+    _section(36, "Parliament and the legislatures are committed to promoting "
+                 "equal opportunities."),
+    _note("Primacy of Constitution of Canada"),
+    _section(52, "The Constitution of Canada is the supreme law of Canada."),
+])
+
+
+class ParseTests(unittest.TestCase):
+    def setUp(self):
+        self.chunks = charter.parse(FIXTURE)
+        self.by_id = {c["id"]: c for c in self.chunks}
+
+    def test_sections_land_in_the_act_they_belong_to(self):
+        # s. 1 exists in both Acts; the 1982 one is the Charter's.
+        self.assertIn("CONST-1982-s1", self.by_id)
+        self.assertNotIn("CONST-1867-s1", self.by_id)
+        self.assertIn("CONST-1867-s91", self.by_id)
+
+    def test_ids_stay_unique_across_the_restart_in_numbering(self):
+        ids = [c["id"] for c in self.chunks]
+        self.assertEqual(len(ids), len(set(ids)))
+
+    def test_out_of_scope_sections_are_dropped(self):
+        # 1982 stops at s. 35 (plus s. 52); 1867 keeps only ss. 91-92.
+        self.assertNotIn("CONST-1982-s36", self.by_id)
+        self.assertIn("CONST-1982-s52", self.by_id)
+        self.assertEqual(sorted(c["section"] for c in self.chunks
+                                if c["act_code"] == "CONST-1867"),
+                         ["91", "92"])
+
+    def test_marginal_note_drops_the_invisible_label(self):
+        self.assertEqual(self.by_id["CONST-1982-s9"]["marginal_note"],
+                         "Detention or imprisonment")
+
+    def test_section_text_runs_to_the_next_marginal_note(self):
+        # s. 2's paragraphs live in a sibling list, outside the Section <p>.
+        self.assertIn("freedom of conscience",
+                      self.by_id["CONST-1982-s2"]["text"])
+        self.assertNotIn("arbitrarily detained",
+                         self.by_id["CONST-1982-s2"]["text"])
+
+    def test_chunks_are_citable_legislation(self):
+        chunk = self.by_id["CONST-1982-s9"]
+        self.assertEqual(chunk["doc_type"], "legislation")
+        self.assertEqual(chunk["citation"], "Charter, s. 9")
+        self.assertEqual(chunk["source_url"], charter.URL)
+
+    def test_marginal_note_without_a_section_label_is_skipped(self):
+        html = ("<h1>CONSTITUTION ACT, 1982</h1>" + _note("Schedule heading") +
+                "<p>prose with no section label</p>")
+        self.assertEqual(charter.parse(html), [])
+
+    def test_a_reshaped_page_scrapes_to_nothing(self):
+        # What a JS-rendered rebuild would look like: heading still there, the
+        # MarginalNote/Section markup gone. This is the empty write the guard
+        # in build() exists to refuse.
+        html = ("<h1>CONSTITUTION ACT, 1982</h1>"
+                '<table id="const-tbl" class="wb-tables"></table>')
+        self.assertEqual(charter.parse(html), [])
+
+    def test_losing_the_1982_heading_is_an_error_not_a_half_corpus(self):
+        # Without the split every 1982 section would be filed under 1867.
+        with self.assertRaises(ValueError):
+            charter.parse("<h1>CONSTITUTION ACT, 1867</h1>" + _note("x") +
+                          _section(91, "text"))
+
+
+class MainTests(unittest.TestCase):
+    """main() has to make a refused write visible to refresh.py, which runs
+    this module as a subprocess and only sees the exit status."""
+
+    def _run(self, argv, build_result=True):
+        with mock.patch.object(charter, "build") as build:
+            build.return_value = build_result
+            with mock.patch.object(sys, "argv", argv):
+                with self.assertRaises(SystemExit) as caught:
+                    charter.main()
+        return caught.exception.code, build
+
+    def test_refused_write_exits_nonzero(self):
+        code, _build = self._run(["charter"], build_result=False)
+        self.assertEqual(code, 1)
+
+    def test_successful_write_exits_zero(self):
+        code, _build = self._run(["charter"], build_result=True)
+        self.assertEqual(code, 0)
+
+    def test_allow_shrink_flag_reaches_build(self):
+        _code, build = self._run(["charter", "--allow-shrink"])
+        build.assert_called_once_with(allow_shrink=True)
+
+    def test_guard_is_on_by_default(self):
+        _code, build = self._run(["charter"])
+        build.assert_called_once_with(allow_shrink=False)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_commentary.py

@@ -1,12 +1,20 @@
 """Unit tests for the curated-commentary pipeline (canlex/commentary.py and
 the canlex_us_disposition matcher in canlex/server.py).
 
-Offline: they run on a minimal in-memory dataset, not the curated file.
+Offline: they run on a minimal in-memory dataset, not the curated file, and
+write only into a tempdir.
 
     python -m unittest discover -s tests
 """
+import contextlib
+import io
+import json
+import sys
+import tempfile
 import unittest
+from pathlib import Path
 
+from canlex import commentary
 from canlex.commentary import BANNER, _entry_text
 from canlex.server import _match_dispositions
 
@@ -69,5 +77,129 @@ class MatcherTests(unittest.TestCase):
             _match_dispositions(self.DATA, "entirely unrelated words", None), [])
 
 
+def dataset(n):
+    """A curated file holding n dispositions -> n + 2 chunks (one methodology
+    entry, n disposition entries, one Florida state page)."""
+    return {"reviewed": "2026-07-01",
+            "methodology": [{"id": "m1", "title": "How this was built",
+                             "text": "Method body."}],
+            "dispositions": [entry(id=f"d{i}", names=[f"disposition {i}"])
+                             for i in range(n)]}
+
+
+class _TempCorpus:
+    """Points the module's curated inputs and its output at a tempdir."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        root = Path(self.tmp.name)
+        self.out = root / "commentary.json"
+        self.curated = root / "us_dispositions.json"
+        # No equivalency file: build() already treats it as optional, so the
+        # chunk arithmetic below stays down to the dispositions dataset.
+        for name, tmp_path in (("OUT", self.out), ("CURATED", self.curated),
+                               ("EQUIV", root / "us_equivalency.json")):
+            self.addCleanup(setattr, commentary, name,
+                            getattr(commentary, name))
+            setattr(commentary, name, tmp_path)
+
+    def curate(self, n):
+        self.curated.write_text(json.dumps(dataset(n)), encoding="utf-8")
+
+    def stored(self):
+        return json.loads(self.out.read_text(encoding="utf-8"))
+
+    def run_build(self, **kw):
+        with contextlib.redirect_stdout(io.StringIO()) as out:
+            ok = commentary.build(**kw)
+        return ok, out.getvalue()
+
+
+class BuildWriteTests(_TempCorpus, unittest.TestCase):
+    """The write path. build() re-renders the whole corpus from the curated
+    files, so a truncated or half-edited dataset must not replace good chunks.
+    """
+
+    def test_first_run_writes_and_reports_success(self):
+        self.curate(3)
+        ok, out = self.run_build()
+        self.assertTrue(ok)
+        self.assertEqual(len(self.stored()), 5)
+        self.assertIn("commentary chunks", out)
+
+    def test_collapsed_dataset_is_refused_and_leaves_the_corpus_alone(self):
+        self.curate(20)
+        self.run_build()
+        self.curate(1)                       # dataset truncated to a stub
+        ok, out = self.run_build()
+        self.assertFalse(ok)
+        self.assertIn("REFUSING", out)
+        self.assertEqual(len(self.stored()), 22)
+
+    def test_allow_shrink_lets_a_known_drop_through(self):
+        self.curate(20)
+        self.run_build()
+        self.curate(1)
+        ok, _out = self.run_build(allow_shrink=True)
+        self.assertTrue(ok)
+        self.assertEqual(len(self.stored()), 3)
+
+    def test_normal_churn_still_writes(self):
+        self.curate(20)
+        self.run_build()
+        self.curate(19)
+        self.assertTrue(self.run_build()[0])
+        self.assertEqual(len(self.stored()), 21)
+
+    def test_writes_one_space_indent(self):
+        # The stored corpus is indent=1; any other value rewrites every line
+        # of commentary.json and buries the real change in the diff.
+        self.curate(1)
+        self.run_build()
+        self.assertTrue(
+            self.out.read_text(encoding="utf-8").startswith('[\n {\n  "id"'))
+
+    def test_non_ascii_is_not_escaped(self):
+        self.curate(1)
+        self.run_build()
+        self.assertIn("—", self.out.read_text(encoding="utf-8"))
+
+
+class MainExitTests(_TempCorpus, unittest.TestCase):
+    """A refused write has to fail the process: refresh.py shells out to
+    `py -m canlex.commentary` and only sees the exit code."""
+
+    def argv(self, *args):
+        self.addCleanup(setattr, sys, "argv", sys.argv)
+        sys.argv = ["canlex.commentary", *args]
+
+    def run_main(self):
+        with self.assertRaises(SystemExit) as caught:
+            with contextlib.redirect_stdout(io.StringIO()):
+                commentary.main()
+        return caught.exception.code
+
+    def test_exits_zero_on_a_good_run(self):
+        self.curate(3)
+        self.argv()
+        self.assertEqual(self.run_main(), 0)
+
+    def test_exits_non_zero_when_the_write_is_refused(self):
+        self.curate(20)
+        self.run_build()
+        self.curate(1)
+        self.argv()
+        self.assertEqual(self.run_main(), 1)
+
+    def test_allow_shrink_flag_is_parsed(self):
+        self.curate(20)
+        self.run_build()
+        self.curate(1)
+        self.argv("--allow-shrink")
+        self.assertEqual(self.run_main(), 0)
+        self.assertEqual(len(self.stored()), 3)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_common.py

@@ -0,0 +1,128 @@
+"""Unit tests for the shared ingestion helpers (canlex/_common.py).
+
+Offline only. The corpus-write guard exists because every ingester rebuilds its
+whole processed file from a scrape: on 2026-07-27 a rebuilt upstream index
+scraped to zero items and the write emptied 1,870 good chunks.
+"""
+import contextlib
+import io
+import json
+import tempfile
+import unittest
+from pathlib import Path
+
+from canlex import _common
+
+
+class SafeToWriteTests(unittest.TestCase):
+    def test_refuses_to_replace_a_corpus_with_nothing(self):
+        self.assertFalse(_common.safe_to_write(0, 1870))
+
+    def test_refuses_a_large_collapse(self):
+        self.assertFalse(_common.safe_to_write(500, 1870))
+
+    def test_allows_normal_churn(self):
+        self.assertTrue(_common.safe_to_write(1866, 1870))
+        self.assertTrue(_common.safe_to_write(1900, 1870))
+
+    def test_first_ever_run_is_allowed(self):
+        self.assertTrue(_common.safe_to_write(0, 0))
+
+    def test_ratio_is_tunable(self):
+        self.assertTrue(_common.safe_to_write(500, 1000, ratio=0.5))
+        self.assertFalse(_common.safe_to_write(499, 1000, ratio=0.5))
+
+
+class WriteCorpusTests(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.path = Path(self.tmp.name) / "corpus.json"
+        self.addCleanup(self.tmp.cleanup)
+
+    def _write(self, chunks):
+        self.path.write_text(json.dumps(chunks), encoding="utf-8")
+
+    def test_writes_when_there_is_nothing_to_lose(self):
+        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))
+        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
+                         [{"id": "a"}])
+
+    def test_refuses_and_leaves_the_file_untouched(self):
+        self._write([{"id": str(i)} for i in range(100)])
+        with contextlib.redirect_stdout(io.StringIO()) as out:
+            self.assertFalse(_common.write_corpus(self.path, []))
+        self.assertIn("REFUSING", out.getvalue())
+        self.assertEqual(len(json.loads(self.path.read_text(encoding="utf-8"))),
+                         100)
+
+    def test_allow_shrink_overrides(self):
+        self._write([{"id": str(i)} for i in range(100)])
+        self.assertTrue(_common.write_corpus(self.path, [], allow_shrink=True))
+        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")), [])
+
+    def test_unreadable_stored_file_does_not_block_a_write(self):
+        # A truncated/corrupt corpus reads as empty, so recovery is possible.
+        self.path.write_text("{not json", encoding="utf-8")
+        self.assertTrue(_common.write_corpus(self.path, [{"id": "a"}]))
+
+    def test_creates_missing_parent_directories(self):
+        nested = Path(self.tmp.name) / "deep" / "corpus.json"
+        self.assertTrue(_common.write_corpus(nested, [{"id": "a"}]))
+        self.assertTrue(nested.exists())
+
+    def test_non_ascii_survives_the_round_trip(self):
+        chunks = [{"id": "a", "text": "détention à la frontière"}]
+        _common.write_corpus(self.path, chunks)
+        self.assertEqual(json.loads(self.path.read_text(encoding="utf-8")),
+                         chunks)
+
+
+class StoredChunksTests(unittest.TestCase):
+    def test_missing_file_is_empty(self):
+        self.assertEqual(_common.stored_chunks(Path("nope-does-not-exist.json")),
+                         [])
+
+
+class PreserveDroppedTests(unittest.TestCase):
+    STORED = [{"section": "A", "text": "one"},
+              {"section": "A", "text": "two"},
+              {"section": "B", "text": "three"}]
+
+    def test_reattaches_a_dropped_item(self):
+        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
+        self.assertEqual(dropped, ["A"])
+        self.assertEqual(len(chunks), 2)
+
+    def test_only_restricts_to_genuine_failures(self):
+        # B vanished deliberately (cancelled upstream); it must not come back.
+        chunks, dropped = _common.preserve_dropped([], self.STORED, only={"A"})
+        self.assertNotIn("B", dropped)
+        self.assertNotIn("three", [c["text"] for c in chunks])
+
+    def test_without_only_every_missing_item_is_restored(self):
+        _chunks, dropped = _common.preserve_dropped([], self.STORED)
+        self.assertEqual(dropped, ["A", "B"])
+
+    def test_freshly_scraped_items_are_not_overwritten(self):
+        fresh = [{"section": "A", "text": "new"}]
+        chunks, dropped = _common.preserve_dropped(fresh, self.STORED,
+                                                   only={"A"})
+        self.assertEqual((chunks, dropped), (fresh, []))
+
+    def test_a_stored_chunk_without_the_identity_field_is_skipped(self):
+        # Not a crash: an older-schema corpus must not take down the rebuild
+        # the guard exists to protect.
+        stored = self.STORED + [{"id": "legacy-1", "text": "no section key"}]
+        chunks, dropped = _common.preserve_dropped([], stored, only={"A"})
+        self.assertEqual(dropped, ["A"])
+        self.assertNotIn("legacy-1", [c.get("id") for c in chunks])
+
+    def test_custom_key(self):
+        stored = [{"case": "Khosa", "text": "x"}]
+        chunks, dropped = _common.preserve_dropped(
+            [], stored, key=lambda c: c["case"], only={"Khosa"})
+        self.assertEqual((len(chunks), dropped), (1, ["Khosa"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_delegation.py

@@ -0,0 +1,89 @@
+"""Unit tests for the delegation-instrument ingester (canlex/delegation.py).
+
+Offline only. The ingester rebuilds delegation.json from scratch on every run,
+so an instrument whose page moves or changes shape drops out of the corpus
+silently -- the failure mode that emptied dmemos.json on 2026-07-27. These cover
+the identity the run's failures are tracked by, not the shared write guard
+(tests/test_common.py owns that).
+"""
+import unittest
+
+from canlex import delegation
+
+
+SRC = {
+    "code": "cbsa-2023-05",
+    "act_code": "CBSA-IRPA-DELEG-2023-05",
+    "act_short": "CBSA Deleg 2023-05-08",
+    "act_name": "Delegation of Authority ...",
+    "url": "https://example.invalid/irpa-lipr-2023-05-08-eng.html",
+    "effective": "2023-05-08",
+}
+
+
+class ChunkIdentityTests(unittest.TestCase):
+    """preserve_failed keys on act_code, so the parsers must stamp it."""
+
+    def test_schedule_items_carry_the_instrument_act_code(self):
+        html = ("<main><table class='table-bordered'>"
+                "<tr><td>1.</td><td>A55(1)</td><td>Arrest with warrant</td>"
+                "<td><p class='h4'>CBSA</p><ul><li>Officer</li></ul></td></tr>"
+                "</table></main>")
+        chunks = delegation.parse_cbsa(html, SRC)
+        self.assertTrue(chunks)
+        self.assertEqual({c["act_code"] for c in chunks}, {SRC["act_code"]})
+
+    def test_narrative_instruments_carry_it_too(self):
+        html = "<main><p>The following positions are authorized.</p></main>"
+        chunks = delegation.parse_cbsa_narrative(html, SRC)
+        self.assertEqual([c["act_code"] for c in chunks], [SRC["act_code"]])
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "item 1"},
+              {"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "item 2"},
+              {"act_code": "IRCC-IL3-DELEG", "text": "IL3 item"}]
+
+    def test_unfetchable_instrument_keeps_its_last_good_chunks(self):
+        failures = [("CBSA-IRPA-DELEG-2023-05", "HTTPError: 404")]
+        chunks, preserved = delegation.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["CBSA-IRPA-DELEG-2023-05"])
+        self.assertEqual([c["text"] for c in chunks], ["item 1", "item 2"])
+
+    def test_an_instrument_that_parses_to_nothing_is_a_failure_too(self):
+        # A reachable page whose table markup changed -- the CBSA D-memo trap.
+        failures = [("IRCC-IL3-DELEG", "no content parsed")]
+        _chunks, preserved = delegation.preserve_failed([], failures,
+                                                        self.STORED)
+        self.assertEqual(preserved, ["IRCC-IL3-DELEG"])
+
+    def test_freshly_parsed_chunks_win_over_stored(self):
+        fresh = [{"act_code": "CBSA-IRPA-DELEG-2023-05", "text": "new text"}]
+        failures = [("CBSA-IRPA-DELEG-2023-05", "no content parsed")]
+        chunks, preserved = delegation.preserve_failed(fresh, failures,
+                                                       self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_instrument_retired_from_sources_is_not_resurrected(self):
+        # Removing an entry from SOURCES is deliberate; it never fails, so it
+        # must stay out of the corpus.
+        failures = [("CBSA-IRPA-DELEG-2023-05", "HTTPError: 404")]
+        chunks, preserved = delegation.preserve_failed([], failures,
+                                                       self.STORED)
+        self.assertNotIn("IRCC-IL3-DELEG", preserved)
+        self.assertNotIn("IL3 item", [c["text"] for c in chunks])
+
+    def test_instrument_we_never_stored_cannot_be_preserved(self):
+        failures = [("CBSA-IRPA-PEACEOFF-2022-08", "HTTPError: 404")]
+        chunks, preserved = delegation.preserve_failed([], failures,
+                                                       self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_a_clean_run_preserves_nothing(self):
+        fresh = [{"act_code": "IRCC-IL3-DELEG", "text": "x"}]
+        chunks, preserved = delegation.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_directive.py

@@ -0,0 +1,83 @@
+"""Unit tests for the NJC directive ingester (canlex/directive.py). Offline only.
+
+The ingester is a full rebuild, so anything a run fails to parse disappears from
+directives.json -- the shape of the 2026-07-27 D-memo loss, one directive at a
+time. These cover which directives get their last-good chunks back, and the
+act_code identity that decision hangs on.
+"""
+import unittest
+
+from canlex import directive
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"act_code": "d4", "text": "relocation"},
+              {"act_code": "d4", "text": "part two"},
+              {"act_code": "fsd-dse", "text": "foreign service"}]
+
+    def test_unparseable_directive_keeps_its_last_good_chunks(self):
+        failures = [("d4", "NJC Relocation Directive", "no content parsed")]
+        chunks, preserved = directive.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["d4"])
+        self.assertEqual(len(chunks), 2)
+
+    def test_freshly_parsed_chunks_win_over_stored(self):
+        fresh = [{"act_code": "d4", "text": "new text"}]
+        failures = [("d4", "NJC Relocation Directive", "404")]
+        chunks, preserved = directive.preserve_failed(fresh, failures,
+                                                      self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_directive_dropped_from_the_index_is_not_resurrected(self):
+        # Not in this run and not a failure: the NJC retired it on purpose.
+        chunks, preserved = directive.preserve_failed([], [], self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_directive_we_never_stored_cannot_be_preserved(self):
+        failures = [("d99", "Brand New Directive", "404")]
+        chunks, preserved = directive.preserve_failed([], failures, self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_a_retitled_directive_is_still_matched(self):
+        # Identity is the URL slug, so an NJC reword of the index title (which
+        # is all `failures` used to carry) does not lose the stored chunks.
+        failures = [("fsd-dse", "Foreign Service Directives (2026 edition)",
+                     "no content parsed")]
+        chunks, preserved = directive.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["fsd-dse"])
+        self.assertEqual([c["text"] for c in chunks], ["foreign service"])
+
+    def test_each_failed_directive_is_restored_once(self):
+        failures = [("d4", "Relocation", "404"),
+                    ("fsd-dse", "Foreign Service Directives", "404")]
+        chunks, preserved = directive.preserve_failed([], failures, self.STORED)
+        self.assertEqual(preserved, ["d4", "fsd-dse"])
+        self.assertEqual(len(chunks), 3)
+
+
+class FailureIdentityTests(unittest.TestCase):
+    """The failure tuple's code must be the one parse_directive stamps on
+    chunks, or preservation silently matches nothing."""
+
+    URL = "https://www.njc-cnm.gc.ca/directive/d99/en"
+    HTML = ("<main><h2>1.1 Purpose</h2><p>The purpose of this directive.</p>"
+            "</main>")
+
+    def test_parsed_chunks_carry_the_code_from_the_url(self):
+        chunks = directive.parse_directive(self.HTML, self.URL, "Test Directive",
+                                           "2026-01-01")
+        self.assertTrue(chunks)
+        self.assertEqual({c["act_code"] for c in chunks}, {"d99"})
+
+    def test_the_ingest_loop_derives_the_same_code(self):
+        code = directive._CODE.search(self.URL).group(1)
+        chunks = directive.parse_directive(self.HTML, self.URL, "Test Directive",
+                                           "")
+        stored = [dict(c) for c in chunks]
+        _chunks, preserved = directive.preserve_failed(
+            [], [(code, "Test Directive", "404")], stored)
+        self.assertEqual(preserved, [code])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_dmemo.py

@@ -83,20 +83,5 @@ class PreserveFailedTests(unittest.TestCase):
         self.assertEqual((chunks, preserved), (fresh, []))
 
 
-class ShrinkGuardTests(unittest.TestCase):
-    def test_refuses_to_replace_a_corpus_with_nothing(self):
-        self.assertFalse(dmemo.safe_to_write(0, 1870))
-
-    def test_refuses_a_large_collapse(self):
-        self.assertFalse(dmemo.safe_to_write(500, 1870))
-
-    def test_allows_normal_churn(self):
-        self.assertTrue(dmemo.safe_to_write(1866, 1870))
-        self.assertTrue(dmemo.safe_to_write(1900, 1870))
-
-    def test_first_ever_run_is_allowed(self):
-        self.assertTrue(dmemo.safe_to_write(0, 0))
-
-
 if __name__ == "__main__":
     unittest.main()

tests/test_enf.py

@@ -0,0 +1,131 @@
+"""Unit tests for the ENF manual ingester (canlex/enf.py). Offline only.
+
+The ingester rebuilds enf.json from a fresh probe of the canada.ca dam
+directory, so a renamed or unreachable chapter silently disappears from the
+corpus -- the failure mode that emptied dmemos.json on 2026-07-27. These cover
+the identity preserve_failed keys on and the failed-set it derives; the write
+guard itself is covered by tests/test_common.py.
+"""
+import subprocess
+import unittest
+import urllib.error
+
+from canlex import enf
+
+
+def _pages(n, headings=7):
+    """A stand-in for pypdf's per-page text: a cover page plus numbered
+    headings, enough of them to take the heading-segmentation path."""
+    body = "Body text for this section. " * 8
+    return [f"ENF {n}\nLast updated: 2024-03-12\n",
+            "\n".join(f"{i}. Heading number {i}\n{body}"
+                      for i in range(1, headings + 1))]
+
+
+class ChapterIdentityTests(unittest.TestCase):
+    """preserve_failed keys on act_code, so the generator must keep emitting
+    exactly the identity the failed-set is built from."""
+
+    def test_every_chunk_of_a_chapter_shares_one_act_code(self):
+        chunks = enf._chunks_for(5, _pages(5), "https://x/enf05-eng.pdf")
+        self.assertTrue(chunks)
+        self.assertEqual({c["act_code"] for c in chunks}, {"ENF-5"})
+
+    def test_failed_set_matches_the_generated_act_code(self):
+        stored = enf._chunks_for(5, _pages(5), "https://x/enf05-eng.pdf")
+        chunks, preserved = enf.preserve_failed([], [(5, "fetch failed")],
+                                                stored)
+        self.assertEqual(preserved, ["ENF-5"])
+        self.assertEqual(len(chunks), len(stored))
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"act_code": "ENF-14", "text": "rehabilitation"},
+              {"act_code": "ENF-14", "text": "part two"},
+              {"act_code": "ENF-9", "text": "judicial review"}]
+
+    def test_unfetchable_chapter_keeps_its_last_good_chunks(self):
+        chunks, preserved = enf.preserve_failed(
+            [], [(14, "fetch failed: HTTPError: 404")], self.STORED)
+        self.assertEqual(preserved, ["ENF-14"])
+        self.assertEqual([c["text"] for c in chunks],
+                         ["rehabilitation", "part two"])
+
+    def test_chapter_dropped_on_purpose_is_not_resurrected(self):
+        # ENF 9 was not probed this run (removed from _CHAPTERS), so it never
+        # failed and must not come back.
+        chunks, preserved = enf.preserve_failed(
+            [], [(14, "no text layer and no enf14-ocr.txt")], self.STORED)
+        self.assertNotIn("ENF-9", preserved)
+        self.assertNotIn("judicial review", [c["text"] for c in chunks])
+
+    def test_freshly_parsed_chunks_win_over_stored(self):
+        fresh = [{"act_code": "ENF-14", "text": "new text"}]
+        chunks, preserved = enf.preserve_failed(
+            fresh, [(14, "parse failed: PdfReadError: x")], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_chapter_we_never_had_cannot_be_preserved(self):
+        chunks, preserved = enf.preserve_failed([], [(31, "fetch failed")],
+                                                self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_a_clean_run_preserves_nothing(self):
+        fresh = [{"act_code": "ENF-14", "text": "x"}]
+        chunks, preserved = enf.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_every_failing_chapter_is_preserved_independently(self):
+        stored = self.STORED + [{"act_code": "ENF-3", "text": "hearings"}]
+        _chunks, preserved = enf.preserve_failed(
+            [], [(3, "fetch failed"), (14, "parse failed")], stored)
+        self.assertEqual(preserved, ["ENF-14", "ENF-3"])  # sorted by identity
+
+
+class MissingUpstreamTests(unittest.TestCase):
+    """A 404 is this module's retirement signal, not a failure to preserve
+    through -- _CHAPTERS is a blind range(1, 41) probe, so most numbers 404 on
+    a healthy run and a chapter IRCC withdraws starts 404ing too."""
+
+    def _http_error(self, code):
+        # HTTPError is file-like; close it or the suite emits ResourceWarnings.
+        exc = urllib.error.HTTPError("http://x", code, "boom", {}, None)
+        self.addCleanup(exc.close)
+        return exc
+
+    def test_http_404_is_a_retirement(self):
+        self.assertTrue(enf.is_missing_upstream(self._http_error(404)))
+
+    def test_other_http_statuses_are_real_failures(self):
+        for code in (403, 500, 502, 503):
+            self.assertFalse(enf.is_missing_upstream(self._http_error(code)))
+
+    def test_network_errors_are_real_failures(self):
+        self.assertFalse(enf.is_missing_upstream(urllib.error.URLError("dns")))
+        self.assertFalse(enf.is_missing_upstream(TimeoutError("timed out")))
+
+    def test_powershell_404_is_read_off_stderr(self):
+        # The canada.ca path shells out, so the status only appears in prose.
+        exc = subprocess.CalledProcessError(1, "powershell")
+        exc.stderr = ("Invoke-WebRequest : The remote server returned an "
+                      "error: (404) Not Found.")
+        self.assertTrue(enf.is_missing_upstream(exc))
+
+    def test_powershell_bytes_stderr_is_decoded(self):
+        exc = subprocess.CalledProcessError(1, "powershell")
+        exc.stderr = b"error: (404) Not Found."
+        self.assertTrue(enf.is_missing_upstream(exc))
+
+    def test_powershell_other_failure_is_preserved_through(self):
+        exc = subprocess.CalledProcessError(1, "powershell")
+        exc.stderr = "The underlying connection was closed: TLS failure."
+        self.assertFalse(enf.is_missing_upstream(exc))
+
+    def test_no_stderr_at_all_counts_as_a_failure(self):
+        # Unknown cause -> keep the stored chapter; losing in-force guidance
+        # is the worse error.
+        self.assertFalse(enf.is_missing_upstream(RuntimeError("who knows")))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_ingest.py

@@ -4,9 +4,13 @@ Offline: builds a tiny Justice Laws-shaped XML document and parses it, so no
 network or real corpus is touched. Focuses on the per-section in-force date and
 repeal-status metadata.
 """
+import contextlib
+import io
+import json
 import tempfile
 import unittest
 from pathlib import Path
+from unittest import mock
 
 from canlex import ingest
 
@@ -113,5 +117,56 @@ class RepealStatusTests(unittest.TestCase):
         self.assertEqual(_parse()["25"]["status"], "in force")
 
 
+class WriteGuardTests(unittest.TestCase):
+    """A Justice Laws schema change would parse an Act down to a handful of
+    sections; the write must refuse rather than gut the statute. The guard
+    itself is covered by tests/test_common.py -- these pin the wiring."""
+
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.dir = Path(self.tmp.name)
+        self.addCleanup(self.tmp.cleanup)
+
+    def _run(self, parsed, stored=None, **kwargs):
+        target = self.dir / "C-46.json"
+        if stored is not None:
+            target.write_text(json.dumps(stored), encoding="utf-8")
+        with mock.patch.object(ingest, "PROCESSED_DIR", self.dir), \
+                mock.patch.object(ingest, "fetch_xml",
+                                  return_value=Path("unused.xml")), \
+                mock.patch.object(ingest, "parse_legislation",
+                                  return_value=parsed), \
+                contextlib.redirect_stdout(io.StringIO()):
+            return ingest.ingest("C-46", **kwargs), target
+
+    def test_a_collapsed_act_is_refused_and_the_file_survives(self):
+        stored = [{"id": str(i)} for i in range(2184)]
+        with self.assertRaises(RuntimeError):
+            self._run([{"id": "only-one"}], stored=stored)
+        kept = json.loads((self.dir / "C-46.json").read_text(encoding="utf-8"))
+        self.assertEqual(len(kept), 2184)
+
+    def test_allow_shrink_lets_a_understood_drop_through(self):
+        self._run([{"id": "a"}], stored=[{"id": str(i)} for i in range(100)],
+                  allow_shrink=True)
+        kept = json.loads((self.dir / "C-46.json").read_text(encoding="utf-8"))
+        self.assertEqual(len(kept), 1)
+
+    def test_a_normal_run_writes(self):
+        chunks = [{"id": str(i)} for i in range(2184)]
+        _result, target = self._run(chunks, stored=chunks)
+        self.assertEqual(len(json.loads(target.read_text(encoding="utf-8"))),
+                         2184)
+
+    def test_first_ever_ingest_writes(self):
+        _result, target = self._run([{"id": "a"}])
+        self.assertTrue(target.exists())
+
+    def test_refusal_leaves_no_temp_file_behind(self):
+        with self.assertRaises(RuntimeError):
+            self._run([{"id": "a"}], stored=[{"id": str(i)} for i in range(100)])
+        self.assertEqual(list(self.dir.glob("*.tmp")), [])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_irb_guidelines.py

@@ -0,0 +1,48 @@
+"""Unit tests for the IRB Chairperson's Guidelines ingester. Offline only.
+
+build() rebuilds irb_guidelines.json from a fresh fetch of three canada.ca
+pages, so a template change or a dropped connection would delete a guideline
+outright -- the failure mode that emptied dmemos.json on 2026-07-27. The write
+guard itself is covered by tests/test_common.py; these cover this module's
+identity and failed-set.
+"""
+import unittest
+
+from canlex import _common, irb_guidelines
+
+
+class GuidelineIdentityTests(unittest.TestCase):
+    STORED = [{"act_code": "IRB-G4", "text": "gender considerations"},
+              {"act_code": "IRB-G4", "text": "part two"},
+              {"act_code": "IRB-G8", "text": "vulnerable persons"}]
+
+    def _preserve(self, chunks, failed):
+        return _common.preserve_dropped(chunks, self.STORED,
+                                        key=lambda c: c["act_code"],
+                                        only=failed)
+
+    def test_a_failed_guideline_keeps_its_stored_chunks(self):
+        chunks, preserved = self._preserve([], {"IRB-G4"})
+        self.assertEqual(preserved, ["IRB-G4"])
+        self.assertEqual(len(chunks), 2)
+
+    def test_a_guideline_that_succeeded_is_not_resurrected(self):
+        chunks, preserved = self._preserve(
+            [{"act_code": "IRB-G4", "text": "fresh"}], {"IRB-G4"})
+        self.assertEqual(preserved, [])
+        self.assertEqual(len(chunks), 1)
+
+    def test_a_guideline_dropped_on_purpose_stays_gone(self):
+        # G8 is absent from this run and never failed -- it left GUIDELINES.
+        _chunks, preserved = self._preserve([], {"IRB-G4"})
+        self.assertNotIn("IRB-G8", preserved)
+
+    def test_act_code_matches_what_build_emits(self):
+        # Pins the identity: the failed-set is built from g["num"], so the
+        # two must agree or preservation silently never matches.
+        for g in irb_guidelines.GUIDELINES:
+            self.assertRegex(f"IRB-G{g['num']}", r"^IRB-G\w+$")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pdi.py

@@ -0,0 +1,108 @@
+"""Unit tests for the IRCC PDI ingester (canlex/pdi.py). Offline only.
+
+Covers the two ways this ingester can quietly lose guidance: a sub-page that
+fails to fetch disappearing from a full rebuild, and an upstream page whose
+markup changes shape parsing to nothing -- the 2026-07-27 D-memo failure mode,
+which here is caught by the corpus-write guard rather than by preservation.
+"""
+import unittest
+
+from canlex import pdi
+
+PRCARD = ("https://www.canada.ca/en/immigration-refugees-citizenship/corporate/"
+          "publications-manuals/operational-bulletins-manuals/"
+          "permanent-residence/card/apply.html")
+RENEW = PRCARD.replace("apply.html", "renew.html")
+
+
+def _chunk(url, text, chunk_id="pdi-prcard-1-0"):
+    return {"id": chunk_id, "source_url": url, "text": text}
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [_chunk(PRCARD, "how to apply", "pdi-prcard-1-0"),
+              _chunk(PRCARD, "supporting documents", "pdi-prcard-1-1"),
+              _chunk(RENEW, "how to renew", "pdi-prcard-2-0")]
+
+    def test_unfetchable_page_keeps_its_last_good_chunks(self):
+        chunks, preserved = pdi.preserve_failed(
+            [], [(PRCARD, "HTTPError: 404")], self.STORED)
+        self.assertEqual(preserved, [PRCARD])
+        self.assertEqual([c["text"] for c in chunks],
+                         ["how to apply", "supporting documents"])
+
+    def test_page_dropped_from_the_index_is_not_resurrected(self):
+        # RENEW is gone from this run but never errored -- IRCC retired it, and
+        # retired instructions must not come back as if they were current.
+        fresh = [_chunk(PRCARD, "how to apply")]
+        chunks, preserved = pdi.preserve_failed(fresh, [], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_freshly_scraped_page_wins_over_the_stored_copy(self):
+        fresh = [_chunk(PRCARD, "how to apply (2026 wording)")]
+        chunks, preserved = pdi.preserve_failed(
+            fresh, [(PRCARD, "HTTPError: 404")], self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_page_we_never_stored_cannot_be_preserved(self):
+        new_url = PRCARD.replace("apply.html", "replace.html")
+        chunks, preserved = pdi.preserve_failed(
+            [], [(new_url, "TimeoutError: ")], self.STORED)
+        self.assertEqual((chunks, preserved), ([], []))
+
+    def test_failed_identity_is_the_page_url_not_the_error_text(self):
+        # failures are (url, why) pairs; only the url identifies the page.
+        chunks, preserved = pdi.preserve_failed(
+            [], [(PRCARD, PRCARD)], self.STORED)
+        self.assertEqual(preserved, [PRCARD])
+        self.assertEqual(len(chunks), 2)
+
+    def test_first_ever_run_has_nothing_to_preserve(self):
+        fresh = [_chunk(PRCARD, "how to apply")]
+        chunks, preserved = pdi.preserve_failed(
+            fresh, [(RENEW, "HTTPError: 404")], [])
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+
+class PreservedIdCollisionTests(unittest.TestCase):
+    def test_preserved_ids_are_uniquified_against_this_run(self):
+        # A chunk id embeds the page's position in the index listing, so when a
+        # page drops out the ids shift and a preserved chunk can collide with a
+        # live one. Duplicate ids orphan chunks in the embeddings loader.
+        stored = [_chunk(RENEW, "how to renew", "pdi-prcard-2-0")]
+        fresh = [_chunk(PRCARD, "how to apply", "pdi-prcard-2-0")]
+        chunks, _preserved = pdi.preserve_failed(
+            fresh, [(RENEW, "HTTPError: 404")], stored)
+        pdi.uniquify_ids(chunks)
+        self.assertEqual(len({c["id"] for c in chunks}), len(chunks))
+        self.assertEqual(chunks[0]["id"], "pdi-prcard-2-0")
+
+
+class PageChunkTests(unittest.TestCase):
+    PAGE = ("<main><h1>Permanent resident card</h1>"
+            "<time property='dateModified'>2026-05-04</time>"
+            "<h2>Eligibility</h2><p>" + "An applicant must be a PR. " * 8 +
+            "</p><h2>Fees</h2><p>" + "The fee is fifty dollars. " * 8 +
+            "</p></main>")
+    SRC = {"code": "pdi-prcard", "short": "PDI PR Card", "name": "PDI — PR card",
+           "index": "https://x/card.html", "scope": "/card"}
+
+    def test_headings_become_chunks_tagged_with_their_page(self):
+        chunks = pdi._page_chunks(self.SRC, "https://x/card.html", self.PAGE, 0)
+        self.assertEqual([c["marginal_note"] for c in chunks],
+                         ["Eligibility", "Fees"])
+        self.assertEqual({c["source_url"] for c in chunks},
+                         {"https://x/card.html"})
+        self.assertTrue(all(c["doc_type"] == "memorandum" for c in chunks))
+
+    def test_reshaped_page_yields_nothing_to_write(self):
+        # What canada.ca would serve if the instructions moved behind a script:
+        # the parse succeeds and produces zero chunks, which is why the write
+        # has to be guarded rather than trusted.
+        reshaped = "<main><h1>Permanent resident card</h1><div id='app'></div></main>"
+        self.assertEqual(
+            pdi._page_chunks(self.SRC, "https://x/card.html", reshaped, 0), [])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_tariff_schedule.py

@@ -0,0 +1,206 @@
+"""Unit tests for the Customs Tariff Schedule ingester. Offline only.
+
+The corpus is two chapters (98 and 99), so a single chapter that fetches badly
+or parses to nothing is a 30-70% collapse -- the same shape of failure that
+emptied dmemos.json on 2026-07-27. These cover how the module decides a chapter
+failed, how it hangs on to the last-good copy of one that did, and that a
+refused write reaches the shell as a non-zero exit.
+"""
+import json
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from canlex import tariff_schedule as ts
+
+CH98_HTML = ("<main><h2>Notes</h2><p>1. This Chapter applies.</p>"
+             "<table><tr><th>Tariff Item</th></tr>"
+             "<tr><td>98.01</td><td></td><td>Conveyances</td></tr>"
+             "<tr><td>9801.10.10</td><td>10</td><td>Vehicles</td>"
+             "<td>-</td><td>Free</td><td>Free</td></tr>"
+             "</table></main>")
+CH99_HTML = ("<main><table><tr><th>Tariff Item</th></tr>"
+             "<tr><td>99.01</td><td></td><td>Temporary importations</td></tr>"
+             "<tr><td>9901.00.00</td><td></td><td>Fishing equipment</td>"
+             "<td>-</td><td>Free</td><td></td></tr>"
+             "</table></main>")
+
+
+class ChapterIdentityTests(unittest.TestCase):
+    def test_every_chunk_of_a_chapter_shares_one_identity(self):
+        # Nothing else on a chunk names its chapter, so preserve_failed keys on
+        # `part`; the Notes chunk and the heading chunks must agree on it.
+        chunks = ts.parse_chapter(CH98_HTML, ts.SOURCES["ch98"])
+        self.assertEqual({c["part"] for c in chunks}, {ts._chapter_part("98")})
+        self.assertEqual(len(chunks), 2)   # Notes + heading 98.01
+
+    def test_part_matches_what_the_stored_corpus_carries(self):
+        self.assertEqual(ts._chapter_part("99"), "Schedule, Chapter 99")
+
+
+class ParseCollapseTests(unittest.TestCase):
+    """A chapter that parses to nothing is the reshaped-page signature."""
+
+    def test_page_without_main_yields_nothing(self):
+        self.assertEqual(ts.parse_chapter("<div><table></table></div>",
+                                          ts.SOURCES["ch98"]), [])
+
+    def test_js_rendered_table_yields_nothing(self):
+        # An empty table whose rows arrive by script -- what CBSA did to the
+        # D-memo index; the ingester must not read it as a valid chapter.
+        self.assertEqual(
+            ts.parse_chapter('<main><table id="tariff"></table></main>',
+                             ts.SOURCES["ch99"]), [])
+
+    def test_notes_survive_a_missing_table(self):
+        # A partial parse is not a failure -- the shrink guard, not
+        # preserve_failed, is what catches those.
+        chunks = ts.parse_chapter("<main><h2>Notes</h2><p>1. Applies.</p></main>",
+                                  ts.SOURCES["ch99"])
+        self.assertEqual([c["section"] for c in chunks], ["Sch-Ch99-Notes"])
+
+
+class PreserveFailedTests(unittest.TestCase):
+    STORED = [{"part": "Schedule, Chapter 98", "section": "Sch-Ch98-Notes"},
+              {"part": "Schedule, Chapter 98", "section": "Sch-98.01"},
+              {"part": "Schedule, Chapter 99", "section": "Sch-99.01"}]
+
+    def test_failed_chapter_keeps_its_last_good_chunks(self):
+        chunks, preserved = ts.preserve_failed(
+            [{"part": "Schedule, Chapter 99", "section": "Sch-99.01"}],
+            [("98", "HTTPError: 404")], self.STORED)
+        self.assertEqual(preserved, ["Schedule, Chapter 98"])
+        self.assertEqual(len(chunks), 3)
+
+    def test_empty_parse_is_a_failure_and_is_preserved(self):
+        # The reason string is never read; only the chapter identity is.
+        chunks, preserved = ts.preserve_failed([], [("99", "no chunks parsed")],
+                                               self.STORED)
+        self.assertEqual(preserved, ["Schedule, Chapter 99"])
+        self.assertEqual([c["section"] for c in chunks], ["Sch-99.01"])
+
+    def test_chapter_that_did_not_fail_is_not_resurrected(self):
+        # Chapter 98 simply is not in this run's output and did not error, so
+        # whatever it used to hold stays gone.
+        chunks, preserved = ts.preserve_failed(
+            [{"part": "Schedule, Chapter 99", "section": "Sch-99.01"}],
+            [], self.STORED)
+        self.assertEqual(preserved, [])
+        self.assertEqual(len(chunks), 1)
+
+    def test_fresh_chunks_win_over_the_stored_copy(self):
+        fresh = [{"part": "Schedule, Chapter 98", "section": "Sch-98.02"}]
+        chunks, preserved = ts.preserve_failed(fresh, [("98", "boom")],
+                                               self.STORED)
+        self.assertEqual((chunks, preserved), (fresh, []))
+
+    def test_first_ever_run_has_nothing_to_preserve(self):
+        chunks, preserved = ts.preserve_failed([], [("98", "boom")], [])
+        self.assertEqual((chunks, preserved), ([], []))
+
+
+class BuildTests(unittest.TestCase):
+    """build() end to end with the network stubbed out."""
+
+    def setUp(self):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        self.out = Path(tmp.name) / "tariff_schedule.json"
+        patch = mock.patch.object(ts, "OUT", self.out)
+        patch.start()
+        self.addCleanup(patch.stop)
+
+    def _fetch_returning(self, pages):
+        """Stub _fetch: url -> canned HTML, or an exception to raise."""
+        def fetch(url, dest):
+            page = pages[url]
+            if isinstance(page, Exception):
+                raise page
+            return page
+        return fetch
+
+    def _pages(self, ch98=CH98_HTML, ch99=CH99_HTML):
+        return {ts.SOURCES["ch98"]["url"]: ch98,
+                ts.SOURCES["ch99"]["url"]: ch99}
+
+    def _build(self, pages, **kwargs):
+        with mock.patch.object(ts, "_fetch", self._fetch_returning(pages)), \
+                mock.patch("builtins.print"):
+            return ts.build(**kwargs)
+
+    def _stored(self):
+        return json.loads(self.out.read_text(encoding="utf-8"))
+
+    def test_first_run_writes_with_the_existing_indent(self):
+        self.assertTrue(self._build(self._pages()))
+        text = self.out.read_text(encoding="utf-8")
+        # indent=1: changing it would rewrite every line of the data file.
+        self.assertTrue(text.startswith("[\n {\n"), text[:20])
+        self.assertEqual(len(self._stored()), 3)   # ch98 Notes + 98.01 + 99.01
+
+    def test_unreachable_chapter_falls_back_to_the_stored_copy(self):
+        self.assertTrue(self._build(self._pages()))
+        ok = self._build(self._pages(ch98=OSError("connection reset")))
+        self.assertTrue(ok)
+        self.assertEqual(len(self._stored()), 3)
+        self.assertIn("Schedule, Chapter 98",
+                      {c["part"] for c in self._stored()})
+
+    def test_reshaped_chapter_falls_back_to_the_stored_copy(self):
+        self.assertTrue(self._build(self._pages()))
+        self.assertTrue(self._build(self._pages(ch99="<main></main>")))
+        self.assertEqual(len(self._stored()), 3)
+
+    def test_collapse_that_cannot_be_preserved_is_refused(self):
+        # A chapter that still parses, just to far fewer chunks: no failure to
+        # preserve, so the shrink guard is the only thing standing in the way.
+        self.out.write_text(json.dumps(
+            [{"part": "Schedule, Chapter 98", "id": f"old-{i}"}
+             for i in range(20)] +
+            [{"part": "Schedule, Chapter 99", "id": f"old9-{i}"}
+             for i in range(20)]), encoding="utf-8")
+        self.assertFalse(self._build(self._pages()))
+        self.assertEqual(len(self._stored()), 40)   # untouched
+
+    def test_allow_shrink_lets_a_reviewed_collapse_through(self):
+        self.out.write_text(json.dumps([{"part": "Schedule, Chapter 98",
+                                         "id": f"old-{i}"} for i in range(40)]),
+                            encoding="utf-8")
+        self.assertTrue(self._build(self._pages(), allow_shrink=True))
+        self.assertEqual(len(self._stored()), 3)
+
+
+class MainTests(unittest.TestCase):
+    def _main_with(self, argv, result):
+        calls = []
+
+        def stub(allow_shrink=False):
+            calls.append(allow_shrink)
+            return result
+
+        saved_argv, sys.argv = sys.argv, argv
+        with mock.patch.object(ts, "build", stub):
+            try:
+                with self.assertRaises(SystemExit) as exit_info:
+                    ts.main()
+            finally:
+                sys.argv = saved_argv
+        return calls, exit_info.exception.code
+
+    def test_guard_is_on_unless_asked_otherwise(self):
+        calls, code = self._main_with(["prog"], True)
+        self.assertEqual((calls, code), ([False], 0))
+
+    def test_allow_shrink_flag_reaches_build(self):
+        calls, code = self._main_with(["prog", "--allow-shrink"], True)
+        self.assertEqual((calls, code), ([True], 0))
+
+    def test_refused_write_exits_non_zero(self):
+        _calls, code = self._main_with(["prog"], False)
+        self.assertEqual(code, 1)
+
+
+if __name__ == "__main__":
+    unittest.main()