Persist ngmix epoch-cut counts; tile cleaning no longer needs module logs

tests/workflow/params_pin.json

@@ -3,7 +3,7 @@
   "rules": {
     "all": "572122d8d1901e12ff591b30adf405f8920be8183129befececbb819f3392ed8",
     "clean_exposure": "22cb76b13a5205d20a02a9bd3b8c8bea5ea2801555e24dd2b84b11145f7e79d9",
-    "clean_tile": "a5c07b0461526ed407df36a291deb866d4181524b4c8e0fbc3dd047fd9d28479",
+    "clean_tile": "c6a91af666e86bd9735effa005dee862e7d3682079eac23c56a877505bdbe76e",
     "exp_get_images": "a9e9eca0a99a348e43e0cd44d0c0aae0f890773d03e35007fb77f886b7d21f5f",
     "exp_maps": "6114b2bb508217eb1a87c89a295485c4cc17f329391a27832548f525a3ef1d2e",
     "exp_persist": "302e2837542bc1102430c27c81c600b7cda32e8bddcb5fd60d33950987609fff",
@@ -18,15 +18,15 @@
     "tile_find_exposures": "c883495a7ec196c61ef770507865f43834b08523793d337e8ddeaef0ad62f117",
     "tile_get_catalogue": "84ffd03c4bde248c06aa58a0f4b7f399390a0ff3ecbcf07e03d718df4fe0cd7f",
     "tile_get_images": "96e235dad9cdcde8ecf11f4516bc2303dbba1c8115ad343cdb26d3d1513da671",
-    "tile_make_cat": "831f2b104d20326c3836e29d252118b8d2dd9a277be26857f47162b8250e3ca2",
+    "tile_make_cat": "28e38d428333b7998511e91a2ae5f71beb18da4e8342633a045a92ff706033c2",
     "tile_merge_cats": "0783c93deeb7f90bdb9772bbc4117851dff1a671518c81f67edd4c7c79cc7e6a",
     "tile_merge_headers": "3e674f961fd5bb93ce1f3b0e2bb072a25a421f2828315a3db45a960c44a42111",
     "tile_ngmix": "a35a6326f345e2fac77444724e917f5a36c3c19377a6c3c66750c8ee32250e62",
     "tile_uncompress": "4a013cdd04bc3e3c1082e477161d2b03e0eefe280b1aabb9af25c149a977f55b",
     "tile_vignets": "1d434b4a6a4bcdeb8f1e5e9a5e4f228b67ade400245b41b033a2c883fb6a3267"
   },
   "schema": 1,
-  "sha256": "8065298d1fb835b0346c75c9946bbe2f25f4b99131f025662341e388a71c7fbd",
+  "sha256": "6c1a7cf52778e4423ab1e3d4f1424dc7f039a8173266216206556a52491f2935",
   "unit_pre": {
     "exp_get_images": "dac6685a207dae3ea81d296ce53e9ba75d4636cf1f81e6399f2dda452680c992",
     "exp_psf": "87fc8ea1153a709ab7ba56542cfacc041f5f8b7f6cf2f38a9bc9d8004d1cecb4",

workflow/README.md

@@ -345,6 +345,28 @@ profiles/nibi/config.yaml  SLURM executor; apptainer SDM; per-user jobs cap; kee
   since the missing tombstones schedule exactly the outstanding clean jobs.
   The tombstone is written *before* anything is deleted, so a crash can cost
   disk but never the record.
+- **Tile epoch-cut diagnostics survive cleaning.**
+  `tile_ngmix` completeness checks capture PR #922's five counters before each chunk's temporary directory goes away.
+  `tile_make_cat` sums all configured chunks and publishes `tile_make_cat.json` beside the tile's persistent `final_cat`.
+  Its `epoch_cuts` field contains `schema_version`, `n_chunks`, per-chunk `chunks`, and tile `totals` for `considered`, `masked_fraction`, `central_veto`, `failed`, and `objects_emptied`.
+  Counts are sequential veto counts, not independent cut incidences; `objects_emptied` counts objects whose considered epochs were all removed before fitting.
+  Missing, malformed, or duplicate summaries fail completeness rather than becoming zeros.
+  `clean_tile` validates the durable count record before touching scratch, so `clean_tiles: true` needs no module-log retention override.
+  This requires an ngmix build with PR #922's end-of-loop `epoch cuts:` summary, including on empty chunks.
+  Older code without that summary fails the ngmix completeness check and leaves scratch intact.
+
+  For an **idle existing campaign** with kept logs, use `workflow/scripts/backfill_epoch_cuts.py --run-dir <run> --products-dir <products> --n-chunks <N> --dry-run` first.
+  Alternatively pass `--config <run_config.yaml>` to resolve products and chunk count from the run's frozen `workflow/config.yaml` plus its run config.
+  The script skips tiles without a nonempty persisted catalogue and refuses missing or ambiguous chunk evidence.
+  Without `--dry-run`, it publishes the same manifest schema and reclaims through `clean_tile.reclaim`, including its survivor checks and tombstone.
+  Don't use it while jobs or another cleaner are writing the campaign.
+  Dry-run reads metadata and small logs only, writes nothing, and reports logical regular-file bytes (not allocated blocks), without following symlinks.
+
+  **Deploy at a campaign boundary.**
+  The completeness/helper hash is part of every ShapePipe rule's `params`; this change also alters make-cat's params, outputs and shell, and clean-tile's inputs and shell.
+  An ordinary resume against old completed tiles can therefore schedule science stages again, even after their exposure stores have been reclaimed.
+  Backfill old campaigns without changing their frozen code or resuming them against this workflow; removing the log-retention override is appropriate for new campaigns running this contract.
+
 - **A finished tile declares no reclaimed exposures.** Deleting an exposure's
   manifests would otherwise rerun every other tile that reads it, and those
   reruns spread across the exposure-overlap component. So a tile whose

workflow/Snakefile

@@ -493,10 +493,13 @@ def path_hash(path):
                 f"that needs the file will say so when it runs.")
         return "missing"
 
-SCRIPT_HASH = script_hash("completeness.py")
+# Epoch-count parsing/aggregation is part of the completeness contract too.
+SCRIPT_HASH = ":".join((script_hash("completeness.py"),
+                        script_hash("epoch_cuts.py")))
 FOREST_HASH = script_hash("build_forest.py")
 CLEAN_HASH  = script_hash("clean_exposure.py")
-CLEAN_TILE_HASH = script_hash("clean_tile.py")
+CLEAN_TILE_HASH = ":".join((script_hash("clean_tile.py"),
+                            script_hash("epoch_cuts.py")))
 PERSIST_HASH = script_hash("persist_exp.py")
 EXP_MAPS_HASH = script_hash("exp_maps.py")
 MERGE_MAPS_HASH = script_hash("merge_exposure_maps.py")

workflow/config.yaml

@@ -181,8 +181,9 @@ machines:
 # selection studies and cost an order of magnitude more.
 #
 # THIS LIST IS EXPOSURE-SIDE ONLY. Tile-side retention is not configurable: the
-# only tile product that persists today is final_cat, written by tile_make_cat
-# straight to products_dir. A tile keep list is #844 follow-up.
+# tile products that persist are final_cat and tile_make_cat.json, written by
+# tile_make_cat straight to products_dir. The manifest contains epoch-cut
+# counts summed over all ngmix chunks. A tile keep list is #844 follow-up.
 #
 # NOTE ON products_dir DEFAULTING TO run_dir (a fixture or smoke test): the tar
 # then lands beside the store on the same filesystem and buys nothing, and
@@ -198,7 +199,10 @@ clean: true
 
 # Per-tile store reclamation after the tile's final_cat. Needed above ~850
 # tiles (scratch quota); deletes the per-tile data sp_tilecost/sp_costmodel
-# read. Set both false for a small control run you want to re-measure.
+# read. Epoch-cut counts survive in products_dir/tiles/<shard>/<tile>/
+# tile_make_cat.json; keeping module logs does NOT require disabling cleaning.
+# Cleaning refuses a tile without this complete durable count record.
+# Set both false for a small control run you want to re-measure timings on.
 clean_tiles: true
 
 # Failed tiles that should not keep their exposures from being cleaned.

workflow/rules/tile.smk

@@ -759,9 +759,9 @@ rule tile_ngmix:
         # reclamation ON the tombstoned tile has lost tile_detect.json and the
         # structural "Input files updated by another job" propagation puts
         # tile_vignets back in the group (measured at clean_tile below), so the
-        # store is rebuilt and nothing trips. The SHIPPED DEFAULT
-        # clean_tiles: false is the dangerous configuration -- the opposite of
-        # how reclamation reads everywhere else in this file.
+        # store is rebuilt and nothing trips. A campaign override setting
+        # clean_tiles: false leaves the dangerous configuration; the shipped
+        # default is true. Epoch-cut diagnostics no longer need that override.
         #
         # So: land this hash, and every later edit to ngmix_range.py, at a
         # campaign boundary on a fresh root. The same rule and direct command as
@@ -908,10 +908,12 @@ rule tile_make_cat:
     output:
         manifest  = f"{TILE_DIR}/manifests/tile_make_cat.json",
         final_cat = f"{PROD_TILE_DIR}/final_cat-{{tile}}.hdf5",
+        stats = f"{PROD_TILE_DIR}/tile_make_cat.json",
     log:
         f"{TILE_DIR}/logs/tile_make_cat.json"
     params:
         pre = lambda wc: unit_pre("tile_make_cat", wc.tile,
+                                  env={"NGMIX_N_CHUNKS": NGMIX_CHUNKS},
                                   pre_run=[tile_local(wc.tile), TILE_VIGNET_REQUIRED,
                                            TILE_CLEAN]),
         script_hash = SCRIPT_HASH
@@ -930,6 +932,8 @@ rule tile_make_cat:
                  post="if [ $rc -eq 0 ]; then\n"
                       '  cp -f "$(ls -1 "$SP_RUN"/output/run_sp_tile_Mc/make_cat_runner'
                       '/output/final_cat*.hdf5 | head -1)" {output.final_cat}\n'
+                      '  cp -f {output.manifest} {output.stats}.tmp\n'
+                      '  mv -f {output.stats}.tmp {output.stats}\n'
                       "fi\n")
 
 
@@ -1008,7 +1012,8 @@ rule clean_tile:
         # the PROD_TILE_DIR pattern so there is exactly one definition of this
         # path (final_cat() in the Snakefile), the same way clean_exposure keys
         # off wildcards.exp alone.
-        lambda wc: final_cat(wc.tile)
+        cat = lambda wc: final_cat(wc.tile),
+        stats = rules.tile_make_cat.output.stats,
     output:
         tombstone = f"{TILE_DIR}/cleaned.json"
     params:
@@ -1022,7 +1027,7 @@ rule clean_tile:
     shell:
         f"python {SCRIPTS}/clean_tile.py"
         " --tile-dir $(dirname {output.tombstone}) --tile {wildcards.tile}"
-        " --tombstone {output.tombstone}"
+        " --tombstone {output.tombstone} --final-cat {input.cat}"
 
 
 # --- the campaign's shear catalogue -----------------------------------------

workflow/scripts/backfill_epoch_cuts.py

@@ -0,0 +1,197 @@
+#!/usr/bin/env python3
+"""Persist epoch-cut counts and reclaim finished tiles of an IDLE campaign.
+
+Run --dry-run first. No DAG or Snakemake resume is needed. Products must be
+outside scratch's run root. Both historical .fits and .hdf5 catalogues are
+supported; a missing, empty or ambiguous persisted catalogue is never cleaned.
+The parser and schema are shared with the live completeness checks, and
+clean_tile owns all keep rules, the dry-run traversal and actual reclamation.
+Do not run while jobs or another cleaner are writing this campaign.
+"""
+
+import argparse
+import json
+import re
+import sys
+from pathlib import Path
+
+import clean_tile
+from completeness import write_if_changed
+from epoch_cuts import (
+    aggregate_chunks,
+    read_chunk_logs,
+    require_durable_manifest,
+    validate_counts,
+)
+
+
+def resolve_config(config_path):
+    """Resolve frozen base defaults plus the run's overlay."""
+    import yaml
+    from run_config import apply_defaults, merge
+
+    base = config_path.parent / "workflow" / "config.yaml"
+    config = apply_defaults(
+        merge(
+            yaml.safe_load(base.read_text()),
+            yaml.safe_load(config_path.read_text()),
+        )
+    )
+    return (
+        Path(config["outputs"]["run_dir"]),
+        Path(config["outputs"]["products_dir"]),
+        int(config["ngmix_chunks"]),
+    )
+
+
+def catalogue_path(products_dir, tile):
+    """Return one nonempty persisted catalogue, or None if none exists."""
+    root = products_dir / "tiles" / tile[:2] / tile
+    paths = [root / f"final_cat-{tile}.{ext}" for ext in ("fits", "hdf5")]
+    paths = [p for p in paths if p.is_file() and p.stat().st_size > 0]
+    if len(paths) > 1:
+        raise ValueError(f"ambiguous persisted final_cat for {tile}: {paths}")
+    return paths[0] if paths else None
+
+
+def backfill_manifest(tile_dir, final_cat, n_chunks):
+    """Build the durable manifest, or validate and reuse an existing one."""
+    durable = final_cat.parent / "tile_make_cat.json"
+    if durable.exists():
+        return require_durable_manifest(final_cat, tile_dir.name, n_chunks)
+    path = tile_dir / "manifests" / "tile_make_cat.json"
+    manifest = json.loads(path.read_text())
+    if (
+        manifest.get("stage") != "tile_make_cat"
+        or manifest.get("unit") != tile_dir.name
+        or manifest.get("status") != "complete"
+    ):
+        raise ValueError(f"not a successful make-cat manifest: {path}")
+    chunks = {}
+    for k in range(1, n_chunks + 1):
+        chunk_path = tile_dir / "manifests" / f"tile_ngmix_{k}.json"
+        chunk = json.loads(chunk_path.read_text())
+        if (
+            chunk.get("stage") != "tile_ngmix"
+            or chunk.get("unit") != tile_dir.name
+            or chunk.get("status") != "complete"
+        ):
+            raise ValueError(f"not a successful chunk manifest: {chunk_path}")
+        if "epoch_cuts" in chunk:
+            chunks[str(k)] = validate_counts(chunk["epoch_cuts"])
+        else:
+            name = f"run_sp_tile_ngmix_Ng{k}u"
+            root = tile_dir / "logs" / "modules" / name
+            if not root.exists():
+                root = tile_dir / "output" / name
+            chunks[str(k)] = read_chunk_logs(root)
+    manifest["epoch_cuts"] = aggregate_chunks(chunks, n_chunks)
+    return manifest
+
+
+def process_tile(tile_dir, products_dir, n_chunks, dry_run):
+    """Validate one tile, report reclaimable bytes and optionally reclaim."""
+    tile = tile_dir.name
+    final_cat = catalogue_path(products_dir, tile)
+    if final_cat is None:
+        print(f"[backfill] {tile}: SKIP no persisted final_cat")
+        return "skipped", 0, None
+    survivors = clean_tile.survivor_paths(tile_dir, tile)
+    clean_tile.require_survivors(survivors, tile)
+    manifest = backfill_manifest(tile_dir, final_cat, n_chunks)
+    tombstone = tile_dir / "cleaned.json"
+    freed = clean_tile.prune(
+        tile_dir, set(survivors.values()) | {tombstone}, [], dry_run=True
+    )
+    record = manifest["epoch_cuts"]
+    action = "WOULD FREE" if dry_run else "FREE"
+    totals_text = json.dumps(record["totals"], sort_keys=True)
+    print(
+        f"[backfill] {tile}: {action} {freed} bytes; "
+        f"{n_chunks} chunks; totals={totals_text}"
+    )
+    if not dry_run:
+        # Publish atomically BEFORE the cleaner can touch scratch. Never run a
+        # Snakemake science stage merely to retrofit this small durable record.
+        write_if_changed(
+            final_cat.parent / "tile_make_cat.json",
+            json.dumps(manifest, indent=2, sort_keys=True) + "\n",
+        )
+        clean_tile.reclaim(tile_dir, tile, tombstone, final_cat)
+    return "ready", freed, record["totals"]
+
+
+def main(argv=None):
+    """Run the backfill CLI; refusals leave the affected tile intact."""
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--run-dir", type=Path, required=True)
+    parser.add_argument("--products-dir", type=Path)
+    parser.add_argument("--n-chunks", type=int)
+    parser.add_argument(
+        "--config",
+        type=Path,
+        help="run overlay beside frozen workflow/config.yaml",
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args(argv)
+    if args.config:
+        run, products, chunks = resolve_config(args.config)
+        if run.resolve() != args.run_dir.resolve():
+            parser.error("--run-dir disagrees with the run config")
+        if (
+            args.products_dir
+            and args.products_dir.resolve() != products.resolve()
+        ):
+            parser.error("--products-dir disagrees with the run config")
+        if args.n_chunks is not None and args.n_chunks != chunks:
+            parser.error("--n-chunks disagrees with the run config")
+        args.products_dir, args.n_chunks = products, chunks
+    if args.products_dir is None or args.n_chunks is None or args.n_chunks < 1:
+        parser.error(
+            "provide --config, or both --products-dir and positive --n-chunks"
+        )
+    run = args.run_dir.resolve()
+    products = args.products_dir.resolve()
+    if products == run or run in products.parents:
+        parser.error("products must be outside the scratch run root")
+    if not (run / "tiles").is_dir():
+        parser.error("run has no tiles directory")
+    totals = {}
+    counts = dict(ready=0, skipped=0, refused=0)
+    freed = 0
+    for tile_dir in sorted((run / "tiles").glob("*/*")):
+        if not re.fullmatch(r"\d+\.\d+", tile_dir.name):
+            continue
+        try:
+            if (
+                tile_dir.is_symlink()
+                or tile_dir.parent.is_symlink()
+                or tile_dir.parent.name != tile_dir.name[:2]
+            ):
+                raise ValueError(f"unexpected tile/shard path: {tile_dir}")
+            status, size, record = process_tile(
+                tile_dir, products, args.n_chunks, args.dry_run
+            )
+            counts[status] += 1
+            freed += size
+            if record:
+                for key, value in record.items():
+                    totals[key] = totals.get(key, 0) + value
+        except (OSError, ValueError, SystemExit) as exc:
+            counts["refused"] += 1
+            print(
+                f"[backfill] {tile_dir.name}: REFUSED {exc}", file=sys.stderr
+            )
+    action = "DRY RUN" if args.dry_run else "DONE"
+    print(
+        f"[backfill] {action}: {counts['ready']} ready, "
+        f"{counts['skipped']} skipped, {counts['refused']} refused; "
+        f"{freed} bytes ({freed / 2**30:.3f} GiB) "
+        f"{'would be freed' if args.dry_run else 'reclaimed'}"
+    )
+    print(f"[backfill] campaign totals={json.dumps(totals, sort_keys=True)}")
+    return 1 if counts["refused"] else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())