Competitive-gap closure: validation suites, fallback policy, ER accuracy evidence, 25M benchmarks, CI honesty

.github/workflows/ci.yml

@@ -66,7 +66,12 @@ jobs:
     if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
     runs-on: ubuntu-latest
     timeout-minutes: 45
-    continue-on-error: true
+    # NO continue-on-error here: when this lane runs, a failure is a real
+    # failure. Visibility is guaranteed by the alert step below (a pinned
+    # issue), so a red nightly can't scroll by unnoticed.
+    permissions:
+      issues: write
+      contents: read
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -79,6 +84,63 @@ jobs:
           pip install -e ".[dev,ml]"
       - name: Nightly online/large drift checks
         run: pytest -m "online or large or tier1"
+      - name: Open/refresh alert issue on failure
+        if: failure()
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const title = "nightly large-data lane failing";
+            const body = `The scheduled online/large test lane failed.\n\n` +
+              `Run: ${context.serverUrl}/${context.repo.owner}/${context.repo.repo}` +
+              `/actions/runs/${context.runId}\n\n` +
+              `This lane does not run on push/PR — a failure here is only ` +
+              `visible through this issue and the Actions tab.`;
+            const open = await github.rest.issues.listForRepo({
+              ...context.repo, state: "open", labels: "nightly-failure",
+            });
+            const existing = open.data.find(i => i.title === title);
+            if (existing) {
+              await github.rest.issues.createComment({
+                ...context.repo, issue_number: existing.number, body,
+              });
+            } else {
+              await github.rest.issues.create({
+                ...context.repo, title, body, labels: ["nightly-failure"],
+              });
+            }
+
+  # Runs on every push/PR: makes the *existence and status* of the large-data
+  # lane visible where a green checkmark would otherwise hide it. Fails if the
+  # large/online markers stop collecting any tests (a silently-rotted lane).
+  large-data-status:
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    steps:
+      - uses: actions/checkout@v4
+      - uses: actions/setup-python@v5
+        with:
+          python-version: "3.12"
+          cache: pip
+      - name: Install
+        run: |
+          python -m pip install --upgrade pip
+          pip install -e ".[dev,ml]"
+      - name: Assert the large/online lane still collects tests
+        run: |
+          N=$(pytest -m "online or large or tier1" --collect-only -q -o addopts="" \
+              2>/dev/null | grep -c "::" || true)
+          echo "large/online/tier1 tests collected: $N"
+          if [ "$N" -eq 0 ]; then
+            echo "::error::the large/online test lane collects zero tests — it has rotted"
+            exit 1
+          fi
+          {
+            echo "## Large-data test lane"
+            echo ""
+            echo "**$N** large/online/tier1 test(s) exist. They do **not** run on this"
+            echo "push/PR — they run on the nightly schedule (see the *nightly-online-large*"
+            echo "job in scheduled runs; failures open a pinned \`nightly-failure\` issue)."
+          } >> "$GITHUB_STEP_SUMMARY"
 
   build:
     needs: quality-fast

.github/workflows/cleanbench-full.yml

@@ -9,6 +9,9 @@ jobs:
   cleanbench:
     runs-on: ubuntu-latest
     timeout-minutes: 30
+    permissions:
+      issues: write
+      contents: read
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -38,3 +41,25 @@ jobs:
           name: cleanbench-results
           path: benchmarks/cleanbench/results/latest.*
           retention-days: 30
+      - name: Open/refresh alert issue on failure
+        if: failure()
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const title = "nightly CleanBench suite failing";
+            const body = `The scheduled CleanBench full suite failed.\n\n` +
+              `Run: ${context.serverUrl}/${context.repo.owner}/${context.repo.repo}` +
+              `/actions/runs/${context.runId}`;
+            const open = await github.rest.issues.listForRepo({
+              ...context.repo, state: "open", labels: "nightly-failure",
+            });
+            const existing = open.data.find(i => i.title === title);
+            if (existing) {
+              await github.rest.issues.createComment({
+                ...context.repo, issue_number: existing.number, body,
+              });
+            } else {
+              await github.rest.issues.create({
+                ...context.repo, title, body, labels: ["nightly-failure"],
+              });
+            }

.github/workflows/perf-regression.yml

@@ -16,6 +16,9 @@ jobs:
   perf:
     runs-on: ubuntu-latest
     timeout-minutes: 20
+    permissions:
+      issues: write
+      contents: read
     steps:
       - uses: actions/checkout@v4
       - uses: actions/setup-python@v5
@@ -42,3 +45,26 @@ jobs:
           name: perf-results
           path: benchmarks/cleanbench/results/latest.*
           retention-days: 30
+      - name: Open/refresh alert issue on failure
+        if: failure()
+        uses: actions/github-script@v7
+        with:
+          script: |
+            const title = "nightly perf-regression gate failing";
+            const body = `The scheduled T5 performance-regression gate failed ` +
+              `(runtime > 120% or memory > 115% of baseline).\n\n` +
+              `Run: ${context.serverUrl}/${context.repo.owner}/${context.repo.repo}` +
+              `/actions/runs/${context.runId}`;
+            const open = await github.rest.issues.listForRepo({
+              ...context.repo, state: "open", labels: "nightly-failure",
+            });
+            const existing = open.data.find(i => i.title === title);
+            if (existing) {
+              await github.rest.issues.createComment({
+                ...context.repo, issue_number: existing.number, body,
+              });
+            } else {
+              await github.rest.issues.create({
+                ...context.repo, title, body, labels: ["nightly-failure"],
+              });
+            }

benchmarks/bench_entity_resolution.py

@@ -156,15 +156,197 @@ def evaluate(df: pd.DataFrame, backend: str) -> dict:
     }
 
 
+# =====================================================================
+# Labelled-dataset mode (benchmarks/data/er_labelled_*.csv)
+# =====================================================================
+
+
+def _labelled_config(null_policy: str, mode: str) -> EntityResolutionConfig:
+    """Linkage config for the labelled dataset schema (see gen_er_labelled.py)."""
+    return EntityResolutionConfig(
+        enabled=True,
+        backend="pandas",
+        unique_id_column="id",
+        blocking_rules=(
+            BlockingRule("lower(l.email) = lower(r.email)", "same email"),
+            BlockingRule(
+                "l.dob = r.dob and substr(lower(l.last_name), 1, 2) = "
+                "substr(lower(r.last_name), 1, 2)",
+                "same dob + surname prefix",
+            ),
+            BlockingRule("l.phone = r.phone", "same phone"),
+        ),
+        comparisons=(
+            ComparisonLevel("full_name", "token_set", weight=2.0),
+            ComparisonLevel("full_name", "jaro_winkler", threshold=0.85, weight=1.5),
+            ComparisonLevel("last_name", "metaphone", weight=1.0),
+            ComparisonLevel("email", "jaro_winkler", threshold=0.90, weight=2.5),
+            ComparisonLevel("dob", "exact", weight=2.0),
+            ComparisonLevel("first_name", "jaro_winkler", threshold=0.85, weight=1.0),
+        ),
+        null_policy=null_policy,  # type: ignore[arg-type]
+        mode=mode,  # type: ignore[arg-type]
+    )
+
+
+def _false_merge_count(report, df: pd.DataFrame) -> int:
+    """Clusters that merged more than one true entity (the costly failure)."""
+    label_of = dict(zip(df["id"], df["entity_id"]))
+    return sum(
+        1
+        for c in report.clusters
+        if len({label_of[rid] for rid in c.record_ids if rid in label_of}) > 1
+    )
+
+
+def _peak_rss_mb() -> float | None:
+    try:
+        import resource
+
+        peak = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
+        return round(peak / (1 << 20) if sys.platform == "darwin" else peak / 1024, 1)
+    except ImportError:  # pragma: no cover
+        return None
+
+
+def evaluate_labelled(df: pd.DataFrame, null_policy: str, mode: str) -> dict:
+    config = _labelled_config(null_policy, mode)
+    t0 = time.perf_counter()
+    _, report = fd.resolve_entities(df, config=config)
+    runtime = time.perf_counter() - t0
+
+    n = len(df)
+    total_pairs = n * (n - 1) // 2
+    predicted = {_norm(p.left_id, p.right_id) for p in report.pairs if p.decision == "match"}
+    truth = truth_pairs(df.rename(columns={"entity_id": "entity_label"}))
+    tp, fp, fn = (
+        len(predicted & truth),
+        len(predicted - truth),
+        len(truth - predicted),
+    )
+    precision = tp / (tp + fp) if (tp + fp) else 0.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
+    return {
+        "config": f"null_policy={null_policy}, mode={mode}",
+        "null_policy": null_policy,
+        "mode": mode,
+        "n_records": n,
+        "candidate_pairs": report.n_candidate_pairs,
+        "reduction_ratio": round(1.0 - report.n_candidate_pairs / total_pairs, 6)
+        if total_pairs
+        else 0.0,
+        "true_pairs": len(truth),
+        "tp": tp,
+        "fp": fp,
+        "fn": fn,
+        "precision": round(precision, 4),
+        "recall": round(recall, 4),
+        "f1": round(f1, 4),
+        "false_merged_clusters": _false_merge_count(report, df),
+        "n_clusters": report.n_clusters,
+        "runtime_s": round(runtime, 4),
+        "peak_rss_mb": _peak_rss_mb(),
+    }
+
+
+def evaluate_exact_baseline(df: pd.DataFrame) -> dict:
+    """pandas drop_duplicates on identity fields — the honesty baseline."""
+    fields = ["full_name", "email", "phone", "dob"]
+    t0 = time.perf_counter()
+    dup_mask = df.duplicated(subset=fields, keep=False)
+    runtime = time.perf_counter() - t0
+    predicted: set[tuple] = set()
+    for _, group in df[dup_mask].groupby(fields, dropna=False):
+        ids = sorted(group["id"].tolist())
+        for a, b in combinations(ids, 2):
+            predicted.add((a, b))
+    truth = truth_pairs(df.rename(columns={"entity_id": "entity_label"}))
+    tp, fp, fn = (
+        len(predicted & truth),
+        len(predicted - truth),
+        len(truth - predicted),
+    )
+    precision = tp / (tp + fp) if (tp + fp) else 1.0
+    recall = tp / (tp + fn) if (tp + fn) else 0.0
+    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) else 0.0
+    return {
+        "config": "pandas exact drop_duplicates (baseline)",
+        "n_records": len(df),
+        "true_pairs": len(truth),
+        "tp": tp,
+        "fp": fp,
+        "fn": fn,
+        "precision": round(precision, 4),
+        "recall": round(recall, 4),
+        "f1": round(f1, 4),
+        "false_merged_clusters": 0,
+        "runtime_s": round(runtime, 4),
+    }
+
+
+_SWEEP = [
+    ("penalize", "balanced"),
+    ("neutral", "balanced"),
+    ("neutral", "precision"),
+    ("neutral", "recall"),
+]
+
+
+def run_labelled(path: str, write_results: str | None) -> list[dict]:
+    df = pd.read_csv(path)
+    results = [evaluate_exact_baseline(df)]
+    results += [evaluate_labelled(df, np_, m) for np_, m in _SWEEP]
+    if write_results:
+        import platform
+
+        payload = {
+            "dataset": path,
+            "generator": "benchmarks/gen_er_labelled.py",
+            "machine": f"{platform.system()} {platform.machine()}",
+            "python": platform.python_version(),
+            "pandas": pd.__version__,
+            "freshdata": fd.__version__,
+            "method": "rule-weighted linkage (normalized weighted average; "
+            "no EM / Fellegi-Sunter)",
+            "results": results,
+        }
+        with open(write_results, "w", encoding="utf-8") as fh:
+            json.dump(payload, fh, indent=2)
+    return results
+
+
 def main(argv: list[str] | None = None) -> int:
     ap = argparse.ArgumentParser(description=__doc__)
     ap.add_argument("--rows", type=int, default=2000)
     ap.add_argument("--dup-rate", type=float, default=0.3)
     ap.add_argument("--seed", type=int, default=0)
     ap.add_argument("--backend", choices=["pandas", "duckdb", "both"], default="both")
     ap.add_argument("--json", action="store_true", help="emit machine-readable JSON")
+    ap.add_argument(
+        "--labelled",
+        metavar="CSV",
+        help="run the labelled-dataset accuracy sweep (see gen_er_labelled.py)",
+    )
+    ap.add_argument(
+        "--write-results", metavar="JSON", help="with --labelled: write results here"
+    )
     args = ap.parse_args(argv)
 
+    if args.labelled:
+        results = run_labelled(args.labelled, args.write_results)
+        if args.json:
+            print(json.dumps(results, indent=2))
+        else:
+            for r in results:
+                print(
+                    f"[{r['config']:<44}] P={r['precision']:.4f} "
+                    f"R={r['recall']:.4f} F1={r['f1']:.4f} "
+                    f"FP={r['fp']} FN={r['fn']} "
+                    f"false-merges={r['false_merged_clusters']}"
+                )
+        return 0
+
     df = synth(args.rows, args.dup_rate, seed=args.seed)
     backends = ["pandas", "duckdb"] if args.backend == "both" else [args.backend]