Methodology
Probe implementation
The actual source that drives every probe in this report — read straight from scripts/ at build time. 4 files, 865 lines. This is the real logic, not a paraphrase.
How the probe engine works
Two-phase discovery-then-finalize
Each table is probed in two phases. First a *discovery* pass probes candidate filter-value maps once and picks the winning set (the one returning the most rows with no error) — the default map is tried first and short-circuits if it succeeds, so the common case is a single discovery probe. Then the canonical `minimal` / `realistic` / `edge` variants run against only that winning value set. Exploration failures (non-existent alternate paths, wrong repo names) are never counted against the final verdict, so a working table is not dragged to 'partial' by a bad alternate value.
VALUE_ALTERNATES
A per-filter dictionary of real candidate values (`org`: anomalyco/sst, `owner`: FiscalMindset/anomalyco/sst, `repo`: opencode/linux/Hello-World, `path`: ''/README.md/src, plus `branch`, `installation_id`, `app_slug`). The discovery pass walks these so a table passes if ANY real value set returns rows. The actual GitHub App for this project is installed on the `FiscalMindset` user account (installation 158026446), and the `anomalyco` / `sst` orgs exist.
429 rate-limit retry + backoff
GitHub applies aggressive secondary rate limits; a burst of probes trips 429s that would otherwise falsely fail a probe. `run_coral_sql` detects a 'rate limit' error and retries the query with a 60-second backoff (2 retries), and the final verdict computes on the canonical variants after discovery. This makes a pass verdict meaningful — it reflects the endpoint, not transient throttling.
Version-preserving reports
Every re-probe archives the previous canonical result to `reports/<source>/tables/<name>.history/<timestamp>.json` (append-only) and writes the new result to the canonical file, bumping `history.retry_count`. Nothing is ever deleted — the site shows conversion progress over time and versions.
probe_batch.py
The probe engine. Builds filter-value probes, runs them through `coral sql`, and writes the version-preserving per-table report (canonical + `.history/`).
/opt/render/project/src/scripts/probe_batch.py
| 1 | #!/usr/bin/env python3 |
| 2 | """In-process version of probe_batch: reuses one `coral sql` subprocess |
| 3 | per worker, no per-table spawn overhead. Each table does 3 variants × |
| 4 | N trials = 6 coral calls (~10s each on github). |
| 5 | |
| 6 | For 364 tables with 4 workers ≈ 30 min (vs ~4 hours via probe_table.py). |
| 7 | |
| 8 | Usage: |
| 9 | python3 scripts/probe_batch.py --source github [--workers 4] |
| 10 | """ |
| 11 | import argparse, json, os, statistics, subprocess, sys, time |
| 12 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 13 | from pathlib import Path |
| 14 | |
| 15 | DEFAULT_FILTER_VALUES = { |
| 16 | "owner": "anomalyco", "repo": "opencode", "org": "anomalyco", |
| 17 | "username": "FiscalMindset", "enterprise": "github", |
| 18 | "plan_id": "1", "assignment_id": "1", "app_slug": "coral-benchmarks-connector", |
| 19 | "gist_id": "1", "subject_digest": "sha256:0000", |
| 20 | "branch": "main", "ref": "main", "issue_number": "1", |
| 21 | "pull_number": "1", "alert_number": "1", "attempt_number": "1", |
| 22 | "run_id": "1", "check_run_id": "1", "hook_id": "1", |
| 23 | "runner_group_id": "1", "environment_name": "production", |
| 24 | "commit_sha": "main", "repository_id": "1296269", |
| 25 | "file_sha": "main", "tag_name": "v1", "build_id": "1", |
| 26 | "analysis_id": "1", "sarif_id": "1", "delivery_id": "1", |
| 27 | "deployment_id": "1", "job_id": "1", "secret_name": "1", |
| 28 | "variable_name": "1", "key": "1", "page": "1", |
| 29 | "installation_id": "158026446", "rule_id": "1", "rule_suite_id": "1", |
| 30 | "comment_id": "1", "event_id": "1", "label_name": "bug", |
| 31 | "artifact_id": "1", "delivery_id": "1", |
| 32 | "workflow_file_name": "test.yml", |
| 33 | } |
| 34 | |
| 35 | # Alternate candidate values applied per-filter when the default set 404s. |
| 36 | # Each key maps to a list of fallback values tried before giving up. |
| 37 | VALUE_ALTERNATES = { |
| 38 | "org": ["anomalyco", "sst"], |
| 39 | "owner": ["anomalyco", "FiscalMindset", "sst"], |
| 40 | "repo": ["opencode", "linux", "Hello-World"], |
| 41 | "username": ["FiscalMindset", "anomalyco"], |
| 42 | "path": ["", "README.md", "src"], |
| 43 | "installation_id": ["158026446", "1"], |
| 44 | "app_slug": ["coral-benchmarks-connector", "opencode"], |
| 45 | } |
| 46 | |
| 47 | # Map: git/runs/etc. categorical keys to a discovery file list. |
| 48 | # Loaded from scripts/data/github_discovered_values.json if present. |
| 49 | DISCOVERED_FILTERS = { |
| 50 | "issue_number": "issue_numbers", |
| 51 | "pull_number": "pr_numbers", |
| 52 | "run_id": "workflow_runs", |
| 53 | "check_run_id": "check_run_ids", |
| 54 | "artifact_id": "artifact_ids", |
| 55 | "job_id": "job_ids", |
| 56 | "commit_sha": "commit_shas", |
| 57 | "tag_name": "tags", |
| 58 | "environment_name": "environments", |
| 59 | "label_name": "labels", |
| 60 | "comment_id": "comment_ids", |
| 61 | "event_id": "event_ids", |
| 62 | "delivery_id": "event_ids", |
| 63 | } |
| 64 | |
| 65 | def load_discovered_values(path="scripts/data/github_discovered_values.json"): |
| 66 | """Load discovered real filter values (e.g. issue numbers, run IDs) and |
| 67 | return them as a per-filter list. Appends to VALUE_ALTERNATES at import time.""" |
| 68 | p = Path(path) |
| 69 | if not p.exists(): |
| 70 | return |
| 71 | try: |
| 72 | data = json.loads(p.read_text()) |
| 73 | except Exception: |
| 74 | return |
| 75 | if data.get("scope") and "FiscalMindset" not in data.get("scope", ""): |
| 76 | return # wrong scope |
| 77 | for filter_key, json_key in DISCOVERED_FILTERS.items(): |
| 78 | vals = data.get(json_key, []) |
| 79 | if not vals: |
| 80 | continue |
| 81 | # values may be ints or strings; coerce to str |
| 82 | coerced = [str(v) for v in vals if v] |
| 83 | if not coerced: |
| 84 | continue |
| 85 | # discovered values beat hand-rolled alternates — put them first |
| 86 | existing = list(VALUE_ALTERNATES.get(filter_key, [])) |
| 87 | merged = coerced + [v for v in existing if v not in coerced] |
| 88 | VALUE_ALTERNATES[filter_key] = merged[:20] |
| 89 | # also surface a couple extra useful mappings |
| 90 | if data.get("branches"): |
| 91 | VALUE_ALTERNATES["branch"] = list(dict.fromkeys(list(data["branches"]) + ["main"]))[:12] |
| 92 | if data.get("workflow_files"): |
| 93 | # store under workflow_file_name so table-specific alternates can use it |
| 94 | VALUE_ALTERNATES["workflow_file_name"] = list(data["workflow_files"])[:15] |
| 95 | |
| 96 | load_discovered_values() |
| 97 | |
| 98 | MAX_ROWS_RETURNED = 200 |
| 99 | |
| 100 | def coral_version() -> str: |
| 101 | try: |
| 102 | out = subprocess.run(["coral", "--version"], capture_output=True, text=True, timeout=10) |
| 103 | return (out.stdout or out.stderr).strip().splitlines()[0] if out.returncode == 0 else "unknown" |
| 104 | except Exception: |
| 105 | return "unknown" |
| 106 | |
| 107 | def parse_md_table(out): |
| 108 | lines = [l.rstrip() for l in out.splitlines() if l.strip()] |
| 109 | header = None; rows = [] |
| 110 | for ln in lines: |
| 111 | if ln.startswith("+"): continue |
| 112 | if ln.startswith("|") and ln.endswith("|"): |
| 113 | cells = [c.strip() for c in ln.strip("|").split("|")] |
| 114 | if not any(cells): continue |
| 115 | if header is None: header = cells; continue |
| 116 | if len(cells) != len(header) or sum(1 for c in cells if c) == 0: continue |
| 117 | rows.append(dict(zip(header, cells))) |
| 118 | return header, rows |
| 119 | |
| 120 | def run_coral_sql(query, workspace, rate_limit_retries=2, rate_limit_backoff=60): |
| 121 | """Run one coral sql query. If it hits a GitHub (secondary) rate limit (429), |
| 122 | wait `rate_limit_backoff` seconds and retry up to `rate_limit_retries` times so |
| 123 | transient throttling does not falsely fail a probe. Timeouts on a single call |
| 124 | are also retried once (big-list tables can exceed 60s on first slow call).""" |
| 125 | cmd = ["coral", "sql", "--workspace", workspace, query] |
| 126 | last = None |
| 127 | for attempt in range(rate_limit_retries + 1): |
| 128 | t0 = time.perf_counter() |
| 129 | try: |
| 130 | p = subprocess.run(cmd, capture_output=True, text=True, timeout=120) |
| 131 | except subprocess.TimeoutExpired: |
| 132 | last = (-1, True, "timeout", [], None) |
| 133 | if attempt < rate_limit_retries: |
| 134 | time.sleep(2) |
| 135 | continue |
| 136 | wall = int((time.perf_counter() - t0) * 1000) |
| 137 | header, rows = parse_md_table(p.stdout) |
| 138 | is_err = p.returncode != 0 or "Error" in (p.stdout or "")[:200] |
| 139 | err_msg = ((p.stderr or "") + (p.stdout or ""))[:500] if is_err else None |
| 140 | if is_err and err_msg and "rate limit" in err_msg.lower() and attempt < rate_limit_retries: |
| 141 | time.sleep(rate_limit_backoff) |
| 142 | continue |
| 143 | return wall, is_err, err_msg, rows[:MAX_ROWS_RETURNED], header |
| 144 | return last if last else (0, True, "exhausted retries", [], None) |
| 145 | |
| 146 | def quote(v): return "'" + v.replace("'", "''") + "'" |
| 147 | |
| 148 | # Per-repo discovered (id_key -> value) maps. Loaded from |
| 149 | # scripts/data/github_discovered_values.json: {repo_full_name: {id_key: [values]}}. |
| 150 | _REPO_DISCOVERED = None # lazy-loaded by _build_combo_candidates |
| 151 | |
| 152 | def _build_combo_candidates(required): |
| 153 | """Return a list of {filter_key: value} dicts that pair owner+repo with a |
| 154 | per-repo ID. Each entry is a coherent tuple the install can actually see. |
| 155 | Falls back to single-filter alternates if no combos match.""" |
| 156 | global _REPO_DISCOVERED |
| 157 | if _REPO_DISCOVERED is None: |
| 158 | _REPO_DISCOVERED = {} |
| 159 | p = Path("scripts/data/github_discovered_values.json") |
| 160 | if p.exists(): |
| 161 | try: |
| 162 | data = json.loads(p.read_text()) |
| 163 | except Exception: |
| 164 | data = None |
| 165 | if data: |
| 166 | # Hand-curated per-repo value mappings (from the discovery run). |
| 167 | # Only include IDs we actually verified come from that repo. |
| 168 | _REPO_DISCOVERED["FiscalMindset/Blindfold"] = { |
| 169 | "issue_number": [str(n) for n in data.get("issue_numbers", [])[:8]], |
| 170 | "pull_number": [str(n) for n in data.get("pr_numbers", [])[:8]], |
| 171 | "comment_id": [str(n) for n in data.get("comment_ids", [])[:3]], |
| 172 | "event_id": [str(n) for n in data.get("event_ids", [])[:5]], |
| 173 | "tag_name": data.get("tags", [])[:5], |
| 174 | "label_name": data.get("labels", [])[:5], |
| 175 | } |
| 176 | _REPO_DISCOVERED["FiscalMindset/coral"] = { |
| 177 | "run_id": [str(n) for n in data.get("workflow_runs", [])[:5]], |
| 178 | "check_run_id": [str(n) for n in data.get("check_run_ids", [])[:5]], |
| 179 | "artifact_id": [str(n) for n in data.get("artifact_ids", [])[:5]], |
| 180 | "job_id": [str(n) for n in data.get("job_ids", [])[:5]], |
| 181 | "commit_sha": data.get("commit_shas", [])[:3], |
| 182 | "environment_name": data.get("environments", [])[:3], |
| 183 | "branch": data.get("branches", [])[:5], |
| 184 | } |
| 185 | _REPO_DISCOVERED["FiscalMindset/coral_specs_testing"] = { |
| 186 | "tag_name": [r["tag"] for r in data.get("releases", []) if r.get("repo", "").endswith("coral_specs_testing")], |
| 187 | } |
| 188 | _REPO_DISCOVERED["FiscalMindset/opencode"] = { |
| 189 | "branch": ["main"] + data.get("branches", [])[:5], |
| 190 | } |
| 191 | |
| 192 | # If the table's required filters include owner+repo+<id>, build tuples |
| 193 | # by joining repo-specific ID lists with the matching owner/repo pair. |
| 194 | has_owner = "owner" in required |
| 195 | has_repo = "repo" in required |
| 196 | if not (has_owner and has_repo): |
| 197 | return [] |
| 198 | id_keys = [rf for rf in required if rf not in {"owner", "repo"}] |
| 199 | combos = [] |
| 200 | for repo_full, repo_ids in _REPO_DISCOVERED.items(): |
| 201 | owner, repo = repo_full.split("/", 1) |
| 202 | for ik in id_keys: |
| 203 | for id_val in repo_ids.get(ik, []): |
| 204 | c = {"owner": owner, "repo": repo, ik: id_val} |
| 205 | combos.append(c) |
| 206 | return combos[:30] # cap to keep discovery cheap |
| 207 | |
| 208 | def build_probes(source, table, required): |
| 209 | """Build probe variants, trying alternate candidate filter values so a |
| 210 | table passes if ANY real value set returns rows. Mirrors retry_batch.py's |
| 211 | intent but bakes value-alternation directly into the probe.""" |
| 212 | candidate_maps = [dict(DEFAULT_FILTER_VALUES)] |
| 213 | for rf in required: |
| 214 | if rf in VALUE_ALTERNATES: |
| 215 | for alt in VALUE_ALTERNATES[rf]: |
| 216 | cand = dict(DEFAULT_FILTER_VALUES) |
| 217 | cand[rf] = alt |
| 218 | if cand not in candidate_maps: |
| 219 | candidate_maps.append(cand) |
| 220 | |
| 221 | probes = [] |
| 222 | seen_sql = set() |
| 223 | for idx, fmap in enumerate(candidate_maps): |
| 224 | fmap = {k: fmap.get(k, "1") for k in required} |
| 225 | where = " AND ".join(f"{k} = {quote(v)}" for k, v in fmap.items()) |
| 226 | if not where: |
| 227 | continue # no required filters handled below |
| 228 | if idx == 0: |
| 229 | limits = [(1, "minimal"), (25, "realistic"), (200, "edge")] |
| 230 | else: |
| 231 | limits = [(1, f"alternate-{idx}")] |
| 232 | for limit, label in limits: |
| 233 | sql = f"SELECT * FROM {source}.{table} WHERE {where} LIMIT {limit}" |
| 234 | if sql in seen_sql: |
| 235 | continue |
| 236 | seen_sql.add(sql) |
| 237 | probes.append((label, sql, fmap, f"{label} probe")) |
| 238 | # No required filters: emit limit variants only |
| 239 | if not required: |
| 240 | for label, limit in [("minimal", 1), ("realistic", 25), ("edge", 200)]: |
| 241 | sql = f"SELECT * FROM {source}.{table} LIMIT {limit}" |
| 242 | if sql in seen_sql: |
| 243 | continue |
| 244 | seen_sql.add(sql) |
| 245 | probes.append((label, sql, {}, f"{label} probe (no filters)")) |
| 246 | return probes[:12] |
| 247 | |
| 248 | def run_table(workspace, source, table, required, runs, catalog_rec, prev_history): |
| 249 | table_full = f"{source}.{table}" |
| 250 | now_iso = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) |
| 251 | |
| 252 | # ---- Phase 1: discover a working filter-value set ----------------------- |
| 253 | # 1) Try the default map first — if it returns ≥ 1 row, it's the winner |
| 254 | # (the common fast path, e.g. owner='anomalyco'+repo='opencode' succeeding immediately). |
| 255 | # 2) Only if the default returns 0 rows (or errors), probe alternate value maps. |
| 256 | # Exploration failures are NOT counted against the final verdict. |
| 257 | # IMPORTANT: "no error" is not "winning". Only ≥ 1 row counts as a win, |
| 258 | # so we don't lock in a default-value map that quietly returns 0 rows |
| 259 | # and then fail the canonical variants against it. |
| 260 | win_map = None |
| 261 | win_score = 0 # 0 is the "no winner found yet" sentinel |
| 262 | if required: |
| 263 | # Build combo-aware candidate maps. The discovered-values JSON groups real |
| 264 | # (owner, repo, id) tuples that the install can actually see (e.g. |
| 265 | # Blindfold + issue 54; coral + run 31300020216). A single-filter |
| 266 | # alternation can never find such combos, so we probe coherent tuples first. |
| 267 | combo_candidates = _build_combo_candidates(required) |
| 268 | default_map = dict(DEFAULT_FILTER_VALUES) |
| 269 | for rf in required: |
| 270 | default_map.setdefault(rf, "1") |
| 271 | default_map = {k: default_map.get(k, "1") for k in required} |
| 272 | # ---- Phase 1a: try coherent (owner, repo, id) combos ----------------- |
| 273 | for combo in combo_candidates: |
| 274 | where = " AND ".join(f"{k} = {quote(v)}" for k, v in combo.items()) |
| 275 | sql = f"SELECT * FROM {source}.{table} WHERE {where} LIMIT 1" |
| 276 | wall, is_err, err_msg, rows, _hdr = run_coral_sql(sql, workspace) |
| 277 | if not is_err and len(rows) >= 1: |
| 278 | win_map, win_score = dict(combo), len(rows) |
| 279 | break |
| 280 | # ---- Phase 1b: fall back to single-filter alternation if no combo won |
| 281 | if win_map is None: |
| 282 | where0 = " AND ".join(f"{k} = {quote(v)}" for k, v in default_map.items()) |
| 283 | sql0 = f"SELECT * FROM {source}.{table} WHERE {where0} LIMIT 1" |
| 284 | wall0, is_err0, err0, rows0, _hdr0 = run_coral_sql(sql0, workspace) |
| 285 | if not is_err0 and len(rows0) >= 1: |
| 286 | win_map, win_score = dict(default_map), len(rows0) |
| 287 | else: |
| 288 | for rf in required: |
| 289 | for alt in VALUE_ALTERNATES.get(rf, []): |
| 290 | cand = dict(default_map) |
| 291 | cand[rf] = alt |
| 292 | where = " AND ".join(f"{k} = {quote(v)}" for k, v in cand.items()) |
| 293 | sql = f"SELECT * FROM {source}.{table} WHERE {where} LIMIT 1" |
| 294 | wall, is_err, err_msg, rows, _hdr = run_coral_sql(sql, workspace) |
| 295 | score = (len(rows) if (not is_err and len(rows) >= 1) else -1) |
| 296 | if score > win_score: |
| 297 | win_score = score |
| 298 | win_map = dict(cand) |
| 299 | if win_map is not None: |
| 300 | break |
| 301 | if win_map is not None: |
| 302 | break |
| 303 | if win_map is None: |
| 304 | win_map = dict(default_map) |
| 305 | else: |
| 306 | win_map = {} |
| 307 | |
| 308 | # ---- Phase 2: finalize on the discovered value set only ----------------- |
| 309 | # Run the canonical minimal/realistic/edge variants against the winning map. |
| 310 | variants = [] |
| 311 | if required: |
| 312 | where = " AND ".join(f"{k} = {quote(v)}" for k, v in win_map.items()) |
| 313 | for limit, label in [(1, "minimal"), (25, "realistic"), (200, "edge")]: |
| 314 | sql = f"SELECT * FROM {source}.{table} WHERE {where} LIMIT {limit}" |
| 315 | variants.append((label, sql, dict(win_map), f"{label} probe")) |
| 316 | else: |
| 317 | for label, limit in [("minimal", 1), ("realistic", 25), ("edge", 200)]: |
| 318 | sql = f"SELECT * FROM {source}.{table} LIMIT {limit}" |
| 319 | variants.append((label, sql, {}, f"{label} probe (no filters)")) |
| 320 | |
| 321 | |
| 322 | |
| 323 | if prev_history: |
| 324 | history = { |
| 325 | "first_run_at": prev_history.get("first_run_at"), |
| 326 | "last_retried_at": now_iso, |
| 327 | "retry_count": (prev_history.get("retry_count") or 0) + 1, |
| 328 | } |
| 329 | else: |
| 330 | history = { |
| 331 | "first_run_at": now_iso, |
| 332 | "last_retried_at": now_iso, |
| 333 | "retry_count": 0, |
| 334 | } |
| 335 | |
| 336 | report = { |
| 337 | "table": table_full, |
| 338 | "schema": source, |
| 339 | "captured_at": now_iso, |
| 340 | "required_filters": required, |
| 341 | "guide": catalog_rec.get("guide"), |
| 342 | "description": catalog_rec.get("description"), |
| 343 | "probes": [], |
| 344 | "verdict": {"pass": None, "notes": None}, |
| 345 | "discovery": {"winning_filters": win_map, "note": "winning filter-value set chosen by discovery probes"}, |
| 346 | "history": history, |
| 347 | "metadata": { |
| 348 | "coral_version": coral_version(), |
| 349 | "probe_runner": f"probe_batch.py {runs}x", |
| 350 | "host": os.uname().nodename if hasattr(os, "uname") else "laptop-A", |
| 351 | }, |
| 352 | } |
| 353 | |
| 354 | for label, sql, args_used, intent in variants: |
| 355 | per = {"label": label, "sql": sql, "intent": intent, "args": args_used, "trials": []} |
| 356 | for trial in range(1, runs + 1): |
| 357 | wall, is_err, err_msg, rows, header = run_coral_sql(sql, workspace) |
| 358 | t_rec = { |
| 359 | "trial": trial, "wall_ms": wall, "is_error": is_err, |
| 360 | "row_count": len(rows), |
| 361 | "rows": rows, |
| 362 | "truncated": len(rows) > MAX_ROWS_RETURNED, |
| 363 | "columns": header, |
| 364 | "first_row": rows[0] if rows else None, |
| 365 | } |
| 366 | if err_msg and is_err: |
| 367 | t_rec["error"] = err_msg |
| 368 | per["trials"].append(t_rec) |
| 369 | kept = [t for t in per["trials"] if t["trial"] > 0] |
| 370 | if kept: |
| 371 | walls = [t["wall_ms"] for t in kept if t["wall_ms"] >= 0] |
| 372 | if walls: |
| 373 | per["latency_ms"] = {"min": min(walls), "median": int(statistics.median(walls)), "max": max(walls)} |
| 374 | per["error_count"] = sum(1 for t in kept if t["is_error"]) |
| 375 | per["row_counts"] = [t["row_count"] for t in kept] |
| 376 | report["probes"].append(per) |
| 377 | |
| 378 | errors = sum(p.get("error_count", 0) for p in report["probes"]) |
| 379 | rows_ok = sum(1 for p in report["probes"] if any(t["row_count"] > 0 for t in p["trials"])) |
| 380 | if errors == 0 and rows_ok >= 2: |
| 381 | report["verdict"]["pass"] = True |
| 382 | report["verdict"]["notes"] = f"All variants returned rows on ≥2 trials, 0 errors." |
| 383 | elif errors > 0 and rows_ok > 0: |
| 384 | report["verdict"]["pass"] = "partial" |
| 385 | report["verdict"]["notes"] = f"{errors} errors across trials but {rows_ok} variants returned data." |
| 386 | else: |
| 387 | report["verdict"]["pass"] = False |
| 388 | report["verdict"]["notes"] = f"All variants failed: {errors} errors, {rows_ok} successful." |
| 389 | return report |
| 390 | |
| 391 | def main(): |
| 392 | ap = argparse.ArgumentParser() |
| 393 | ap.add_argument("--source", required=True) |
| 394 | ap.add_argument("--only", default=None) |
| 395 | ap.add_argument("--skip-done", action="store_true") |
| 396 | ap.add_argument("--workers", type=int, default=4) |
| 397 | ap.add_argument("--runs", type=int, default=2) |
| 398 | ap.add_argument("--workspace", default=os.environ.get("CORAL_WORKSPACE", "default")) |
| 399 | args = ap.parse_args() |
| 400 | |
| 401 | cat = json.loads(Path(f"reports/{args.source}/catalog.json").read_text()) |
| 402 | cat_map = {t["name"]: t for t in cat["tables"]} |
| 403 | tables = list(cat_map.keys()) |
| 404 | if args.only: |
| 405 | only = set(args.only.split(",")) |
| 406 | tables = [t for t in tables if t in only] |
| 407 | if args.skip_done: |
| 408 | done = Path(f"reports/{args.source}/tables") |
| 409 | tables = [t for t in tables if not (done / f"{t}.json").exists()] |
| 410 | |
| 411 | print(f"probing {len(tables)} tables (in-process, workers={args.workers})", file=sys.stderr) |
| 412 | |
| 413 | def worker(table): |
| 414 | rec_cat = cat_map[table] |
| 415 | required = rec_cat.get("required_filters", []) |
| 416 | # Load previous history if file exists |
| 417 | fp = Path(f"reports/{args.source}/tables/{table}.json") |
| 418 | prev_history = None |
| 419 | if fp.exists(): |
| 420 | try: |
| 421 | prev_history = json.loads(fp.read_text()).get("history") |
| 422 | except Exception: |
| 423 | pass |
| 424 | rep = run_table(args.workspace, args.source, table, required, args.runs, rec_cat, prev_history) |
| 425 | fp.parent.mkdir(parents=True, exist_ok=True) |
| 426 | fp.write_text(json.dumps(rep, indent=2, default=str) + "\n") |
| 427 | # P2 / new feature: also archive this run to tables/<name>.history/<ISO>.json |
| 428 | # so users can compare versions over time. The canonical JSON stays the latest; |
| 429 | # the history directory is append-only. |
| 430 | try: |
| 431 | hist_dir = Path(f"reports/{args.source}/tables/{table}.history") |
| 432 | hist_dir.mkdir(parents=True, exist_ok=True) |
| 433 | ts = rep["captured_at"].replace(":", "-").replace(".", "-") |
| 434 | (hist_dir / f"{ts}.json").write_text(json.dumps(rep, indent=2, default=str) + "\n") |
| 435 | except Exception as e: |
| 436 | print(f"warn: could not archive history for {table}: {e}", file=sys.stderr) |
| 437 | return table, rep["verdict"]["pass"] |
| 438 | |
| 439 | ok = 0 |
| 440 | with ThreadPoolExecutor(max_workers=args.workers) as ex: |
| 441 | futs = [ex.submit(worker, t) for t in tables] |
| 442 | for i, fut in enumerate(as_completed(futs)): |
| 443 | try: |
| 444 | table, verdict = fut.result() |
| 445 | marker = "✓" if verdict is True else ("~" if verdict == "partial" else "✗") |
| 446 | print(f"{marker} [{i+1}/{len(tables)}] {table}: verdict={verdict}", file=sys.stderr) |
| 447 | if verdict is True: ok += 1 |
| 448 | except Exception as e: |
| 449 | print(f"✗ exception: {e}", file=sys.stderr) |
| 450 | |
| 451 | print(f"\n{ok}/{len(tables)} tables pass", file=sys.stderr) |
| 452 | |
| 453 | if __name__ == "__main__": |
| 454 | main() |
reprobe_github.py
v1.0.0Version-preserving targeted re-probe of github tables. Selects a table set (--all / --failing / --set), optionally refreshes the token, and hands off to probe_batch.py.
/opt/render/project/src/scripts/reprobe_github.py
| 1 | #!/usr/bin/env python3 |
| 2 | """Version-preserving re-probe of github tables using the expanded GitHub App token. |
| 3 | |
| 4 | Why this exists |
| 5 | --------------- |
| 6 | A full github re-probe (364 tables) is expensive and long. This runner lets you |
| 7 | re-probe a **targeted set** of tables (by name, or only the currently failing / |
| 8 | partial ones) while keeping the coral source token fresh and preserving version |
| 9 | history (each prior result is archived to `tables/<name>.history/` by probe_batch). |
| 10 | |
| 11 | Versioning / spec |
| 12 | ----------------- |
| 13 | - Version: 1.0.0 |
| 14 | - Conventions: writes canonical JSON to `reports/github/tables/<name>.json` |
| 15 | (latest wins) and appends each prior run to `reports/github/tables/<name>.history/` |
| 16 | (append-only, version comparison). The Next.js `/report` surface renders the |
| 17 | `coverage_pct`, `snapshots.json`, and the .history file count. |
| 18 | - Does NOT delete any existing report data. |
| 19 | |
| 20 | Selection |
| 21 | --------- |
| 22 | --all re-probe every table (canonical + history) |
| 23 | --failing only tables whose verdict is not True (incl partial) |
| 24 | --include-partial include partial verdicts (default excludes them) |
| 25 | --set=a,b,c explicit comma list of tables |
| 26 | --category=RATE_LIMIT only tables whose last error matches a classifier tag |
| 27 | (see scripts/analyze_failures.py for tag names) |
| 28 | |
| 29 | Usage |
| 30 | ----- |
| 31 | python3 scripts/reprobe_github.py --failing --include-partial --workers 3 |
| 32 | python3 scripts/reprobe_github.py --set=contents,issues,labels --workers 3 |
| 33 | python3 scripts/reprobe_github.py --all --workers 4 |
| 34 | |
| 35 | Rate-limits |
| 36 | ----------- |
| 37 | `probe_batch.py` now retries 429 rate-limit errors with backoff, so transient |
| 38 | GitHub secondary throttling does not falsely fail a probe. |
| 39 | """ |
| 40 | import argparse, json, glob, os, subprocess, sys |
| 41 | from pathlib import Path |
| 42 | |
| 43 | REPO = Path(__file__).resolve().parent.parent |
| 44 | VERSION = "1.0.0" |
| 45 | |
| 46 | def target_tables(selection, include_partial, all_tables): |
| 47 | tables_dir = REPO / "reports/github/tables" |
| 48 | if all_tables: |
| 49 | return sorted(Path(f).stem for f in glob.glob(str(tables_dir / "*.json"))) |
| 50 | if selection: |
| 51 | sel = set(selection) |
| 52 | return sorted(t for t in sel if (tables_dir / f"{t}.json").exists()) |
| 53 | out = [] |
| 54 | for f in glob.glob(str(tables_dir / "*.json")): |
| 55 | d = json.load(open(f)) |
| 56 | v = d.get("verdict", {}) |
| 57 | vp = v.get("pass") if isinstance(v, dict) else v |
| 58 | if vp is True: |
| 59 | continue |
| 60 | if vp == "partial" and not include_partial: |
| 61 | continue |
| 62 | out.append(Path(f).stem) |
| 63 | return sorted(out) |
| 64 | |
| 65 | def main(): |
| 66 | ap = argparse.ArgumentParser(description="Version-preserving targeted github re-probe") |
| 67 | g = ap.add_mutually_exclusive_group() |
| 68 | g.add_argument("--all", action="store_true") |
| 69 | g.add_argument("--failing", action="store_true") |
| 70 | g.add_argument("--set", default="") |
| 71 | ap.add_argument("--include-partial", action="store_true") |
| 72 | ap.add_argument("--workers", type=int, default=3) |
| 73 | ap.add_argument("--runs", type=int, default=1) |
| 74 | ap.add_argument("--refresh-token", action="store_true", |
| 75 | help="refresh the github install token before probing (uses GITHUB_APP_* env)") |
| 76 | args = ap.parse_args() |
| 77 | |
| 78 | if args.refresh_token: |
| 79 | r = subprocess.run([sys.executable, str(REPO / "scripts/refresh_github_token.py")], |
| 80 | cwd=str(REPO)) |
| 81 | if r.returncode != 0: |
| 82 | print("token refresh failed", file=sys.stderr); sys.exit(1) |
| 83 | |
| 84 | if args.set: |
| 85 | selection = [s for s in args.set.split(",") if s] |
| 86 | else: |
| 87 | selection = None |
| 88 | tables = target_tables(selection, args.include_partial, args.all) |
| 89 | if not tables: |
| 90 | print("no tables matched", file=sys.stderr); sys.exit(0) |
| 91 | print(f"[reprobe_github {VERSION}] re-probing {len(tables)} tables", file=sys.stderr) |
| 92 | |
| 93 | # Ensure GITHUB_TOKEN is present in env (coral source may already be installed) |
| 94 | env = dict(os.environ) |
| 95 | argv = ["python3", str(REPO / "scripts/probe_batch.py"), "--source", "github", |
| 96 | "--runs", str(args.runs), "--workers", str(args.workers), "--only", ",".join(tables)] |
| 97 | r = subprocess.run(argv, cwd=str(REPO), env=env, text=True) |
| 98 | sys.exit(r.returncode) |
| 99 | |
| 100 | if __name__ == "__main__": |
| 101 | main() |
| 102 |
refresh_github_token.py
v1.0.0Refreshes the GitHub App installation access token and reinstalls the coral github source, so long probe runs never expire mid-batch.
/opt/render/project/src/scripts/refresh_github_token.py
| 1 | #!/usr/bin/env python3 |
| 2 | """Refresh the coral github source's GitHub App installation token and reinstall it. |
| 3 | |
| 4 | Why this exists |
| 5 | --------------- |
| 6 | The repo probes `github` tables through Coral's bundled source, which authenticates |
| 7 | with a single `GITHUB_TOKEN`. The highest-budget credential available is a GitHub |
| 8 | App **installation access token** (ghs_), which the app owner regenerates from the |
| 9 | app's private key + App ID + installation ID. These tokens live only ~1 hour, so a |
| 10 | long probe run needs them refreshed before expiry. |
| 11 | |
| 12 | This script: |
| 13 | 1. Signs an App JWT (RS256) from the app private key. |
| 14 | 2. POSTs to create a fresh installation access token. |
| 15 | 3. Re-runs `coral source remove github` + `coral source add github` with that token. |
| 16 | |
| 17 | Versioning |
| 18 | ---------- |
| 19 | - Version: 1.0.0 |
| 20 | - Requires: the app private key PEM (path via --pem), App ID (--app-id), |
| 21 | installation ID (--installation). Kept OUT of git (secrets). |
| 22 | - The refresh is idempotent and safe to call repeatedly. |
| 23 | |
| 24 | Usage |
| 25 | ----- |
| 26 | GITHUB_APP_PEM=/path/to/private-key.pem \ |
| 27 | GITHUB_APP_ID=4783650 \ |
| 28 | GITHUB_APP_INSTALLATION=158026446 \ |
| 29 | python3 scripts/refresh_github_token.py |
| 30 | |
| 31 | # or with flags: |
| 32 | python3 scripts/refresh_github_token.py \ |
| 33 | --pem /path/private-key.pem --app-id 4783650 --installation 158026446 |
| 34 | |
| 35 | Outputs |
| 36 | ------- |
| 37 | Prints `coral reloaded, token <prefix> len <n>` and the coral `source add` summary. |
| 38 | """ |
| 39 | import argparse, base64, json, os, subprocess, sys, time |
| 40 | |
| 41 | VERSION = "1.0.0" |
| 42 | |
| 43 | def b64url(data: bytes) -> str: |
| 44 | return base64.urlsafe_b64encode(data).rstrip(b"=").decode() |
| 45 | |
| 46 | def make_app_jwt(app_id: int, pem_path: str) -> str: |
| 47 | """Sign a GitHub App JWT (RS256) valid for ~9 minutes.""" |
| 48 | now = int(time.time()) |
| 49 | header = b64url(json.dumps({"alg": "RS256", "typ": "JWT"}).encode()) |
| 50 | payload = b64url(json.dumps({"iat": now - 30, "exp": now + 540, "iss": str(app_id)}).encode()) |
| 51 | signing_input = f"{header}.{payload}" |
| 52 | sig = subprocess.run( |
| 53 | ["openssl", "dgst", "-sha256", "-sign", pem_path], |
| 54 | input=signing_input.encode(), capture_output=True, |
| 55 | ) |
| 56 | if sig.returncode != 0: |
| 57 | raise RuntimeError(f"openssl signing failed: {sig.stderr.decode()}") |
| 58 | return f"{signing_input}.{b64url(sig.stdout)}" |
| 59 | |
| 60 | def fresh_installation_token(app_id: int, pem_path: str, installation: int) -> str: |
| 61 | jwt = make_app_jwt(app_id, pem_path) |
| 62 | curl = subprocess.run( |
| 63 | ["curl", "-s", "-X", "POST", |
| 64 | "-H", f"Authorization: Bearer {jwt}", |
| 65 | "-H", "Accept: application/vnd.github+json", |
| 66 | f"https://api.github.com/app/installations/{installation}/access_tokens"], |
| 67 | capture_output=True, text=True, |
| 68 | ) |
| 69 | d = json.loads(curl.stdout) |
| 70 | token = d.get("token") |
| 71 | if not token: |
| 72 | raise RuntimeError(f"no installation token in response: {json.dumps(d)[:300]}") |
| 73 | return token |
| 74 | |
| 75 | def reload_coral_token(source: str, token: str) -> subprocess.CompletedProcess: |
| 76 | env = dict(os.environ, GITHUB_TOKEN=token) |
| 77 | subprocess.run(["coral", "source", "remove", source], env=env, capture_output=True) |
| 78 | return subprocess.run(["coral", "source", "add", source], env=env, capture_output=True, text=True) |
| 79 | |
| 80 | def main(): |
| 81 | ap = argparse.ArgumentParser(description="Refresh coral github source installation token") |
| 82 | ap.add_argument("--pem", default=os.environ.get("GITHUB_APP_PEM")) |
| 83 | ap.add_argument("--app-id", type=int, default=int(os.environ.get("GITHUB_APP_ID", "0"))) |
| 84 | ap.add_argument("--installation", type=int, default=int(os.environ.get("GITHUB_APP_INSTALLATION", "0"))) |
| 85 | ap.add_argument("--source", default="github") |
| 86 | ap.add_argument("--print-token", action="store_true", help="also write token to ./_gh_token (gitignored)") |
| 87 | args = ap.parse_args() |
| 88 | if not args.pem or not args.app_id or not args.installation: |
| 89 | print("ERR: --pem, --app-id, --installation are required (or set GITHUB_APP_* env)", file=sys.stderr) |
| 90 | sys.exit(1) |
| 91 | token = fresh_installation_token(args.app_id, args.pem, args.installation) |
| 92 | r = reload_coral_token(args.source, token) |
| 93 | if r.returncode != 0: |
| 94 | print("ERR coral source add:\n" + r.stderr[-1000:], file=sys.stderr) |
| 95 | sys.exit(1) |
| 96 | if args.print_token: |
| 97 | with open("_gh_token", "w") as f: |
| 98 | f.write(token) |
| 99 | print(f"coral reloaded ({args.source}), token {token[:4]} len {len(token)} ({VERSION})") |
| 100 | |
| 101 | if __name__ == "__main__": |
| 102 | main() |
| 103 |
retry_batch.py
Retries previously-failed tables classified as retryable, trying alternate filter values (anomalyco, sst / opencode, linux, Hello-World).
/opt/render/project/src/scripts/retry_batch.py
| 1 | #!/usr/bin/env python3 |
| 2 | """Retry previously-failed probes with rate-limit-aware timing. |
| 3 | |
| 4 | For each failure: |
| 5 | - If category is upstream.not_found (404) on owner=anomalyco, repo=opencode |
| 6 | → try alternates (sst, opencode-ai, octocat) using fast batch. |
| 7 | - If upstream.rate_limit → wait 5s then retry once. |
| 8 | - Otherwise (coral_bug, auth, etc.) → record and skip. |
| 9 | |
| 10 | Usage: |
| 11 | python3 scripts/retry_batch.py --source github [--workers 4] |
| 12 | """ |
| 13 | import argparse, json, os, re, subprocess, sys, time |
| 14 | from concurrent.futures import ThreadPoolExecutor, as_completed |
| 15 | from pathlib import Path |
| 16 | |
| 17 | # Best default filter values that work for many github tables |
| 18 | ALT_OWNER = "anomalyco" |
| 19 | ALT_REPO = "opencode" |
| 20 | ALT_ORG = "opencode" |
| 21 | ALT_USER = "anomalyco" |
| 22 | |
| 23 | DEFAULT_FILTER_VALUES = { |
| 24 | "owner": ALT_OWNER, "repo": ALT_REPO, "org": ALT_ORG, "username": ALT_USER, |
| 25 | "enterprise": "github", "plan_id": "1", "assignment_id": "1", "app_slug": "opencode", |
| 26 | "gist_id": "1", "subject_digest": "sha256:0000", |
| 27 | "branch": "main", "ref": "main", "issue_number": "1", |
| 28 | "pull_number": "1", "alert_number": "1", "attempt_number": "1", |
| 29 | "run_id": "1", "check_run_id": "1", "hook_id": "1", |
| 30 | "runner_group_id": "1", "environment_name": "production", |
| 31 | "commit_sha": "main", "repository_id": "1296269", |
| 32 | "file_sha": "main", "tag_name": "v1", "build_id": "1", |
| 33 | "analysis_id": "1", "sarif_id": "1", "delivery_id": "1", |
| 34 | "deployment_id": "1", "job_id": "1", "secret_name": "1", |
| 35 | "variable_name": "1", "key": "1", "page": "1", |
| 36 | "installation_id": "1", "rule_id": "1", "rule_suite_id": "1", |
| 37 | } |
| 38 | |
| 39 | # Categories we will RETRY |
| 40 | RETRYABLE = { |
| 41 | "upstream.not_found", |
| 42 | "upstream.rate_limit", |
| 43 | "upstream.unprocessable", |
| 44 | "upstream.server_error", |
| 45 | "protocol.decode_failed", |
| 46 | "protocol.empty_response", |
| 47 | "unknown.no_error_msg", |
| 48 | "unknown.unclassified", |
| 49 | "unknown.other", |
| 50 | } |
| 51 | |
| 52 | # Categories we skip entirely |
| 53 | SKIP_CATEGORIES = { |
| 54 | "coral_bug.table_missing", "coral_bug.column_missing", "coral_bug.column_missing_alt", |
| 55 | "coral_bug.required_filter", "coral_bug.named_args", "coral_bug.unsupported_arg", |
| 56 | "coral_bug.invalid_input", |
| 57 | "auth.app_jwt_required", "auth.jwt_decode_failed", |
| 58 | "upstream.gone", "upstream.bad_request", "upstream.forbidden", |
| 59 | } |
| 60 | |
| 61 | def quote(v): return "'" + v.replace("'", "''") + "'" |
| 62 | |
| 63 | def parse_md_table(out): |
| 64 | lines = [l.rstrip() for l in out.splitlines() if l.strip()] |
| 65 | header = None; rows = [] |
| 66 | for ln in lines: |
| 67 | if ln.startswith("+"): continue |
| 68 | if ln.startswith("|") and ln.endswith("|"): |
| 69 | cells = [c.strip() for c in ln.strip("|").split("|")] |
| 70 | if not any(cells): continue |
| 71 | if header is None: header = cells; continue |
| 72 | if len(cells) != len(header) or sum(1 for c in cells if c) == 0: continue |
| 73 | rows.append(dict(zip(header, cells))) |
| 74 | return header, rows |
| 75 | |
| 76 | def run_sql(query, workspace, timeout=60, retry_on_rate_limit=True): |
| 77 | cmd = ["coral", "sql", "--workspace", workspace, query] |
| 78 | for attempt in range(2): |
| 79 | t0 = time.perf_counter() |
| 80 | try: |
| 81 | p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout) |
| 82 | except subprocess.TimeoutExpired: |
| 83 | return -1, True, "timeout", 0, None |
| 84 | wall = int((time.perf_counter() - t0) * 1000) |
| 85 | header, rows = parse_md_table(p.stdout) |
| 86 | is_err = p.returncode != 0 or "Error" in (p.stdout or "")[:200] |
| 87 | err_msg = ((p.stderr or "") + (p.stdout or ""))[:500] if is_err else None |
| 88 | if (not is_err) or (not retry_on_rate_limit): |
| 89 | return wall, is_err, err_msg, len(rows), header |
| 90 | if "rate limit" in (err_msg or "").lower() and attempt == 0: |
| 91 | # Wait then retry once |
| 92 | time.sleep(30) |
| 93 | continue |
| 94 | return wall, is_err, err_msg, len(rows), header |
| 95 | return wall, True, err_msg, 0, header |
| 96 | |
| 97 | def classify(err): |
| 98 | if not err: return "unknown.no_error_msg" |
| 99 | pat = [ |
| 100 | ("upstream.rate_limit", r"Source rate limit exceeded \(429\)"), |
| 101 | ("upstream.not_found", r"Source resource was not found \(404\)"), |
| 102 | ("upstream.unprocessable", r"Source request failed \(422\)"), |
| 103 | ("upstream.server_error", r"Source server error \(5\d\d\)"), |
| 104 | ("auth.app_jwt_required", r"Source authentication failed \(401\)"), |
| 105 | ("auth.jwt_decode_failed", r"A JSON web token could not be decoded"), |
| 106 | ("protocol.decode_failed", r"source (api )?response decode failed"), |
| 107 | ("protocol.empty_response", r"eof while parsing"), |
| 108 | ("upstream.gone", r"Source request failed \(410\)"), |
| 109 | ("upstream.bad_request", r"Source rejected the request \(400\)"), |
| 110 | ("upstream.forbidden", r"Source request failed \(403\)"), |
| 111 | ("coral_bug.table_missing", r"Table `[^`]+` not found"), |
| 112 | ("coral_bug.column_missing", r"No column named `[^`]+`"), |
| 113 | ("coral_bug.required_filter", r"requires a constant equality filter"), |
| 114 | ("coral_bug.named_args", r"requires named arguments"), |
| 115 | ("coral_bug.unsupported_arg", r"Unsupported function argument type"), |
| 116 | ("coral_bug.invalid_input", r"invalid input:"), |
| 117 | ] |
| 118 | for label, p in pat: |
| 119 | if re.search(p, err, re.IGNORECASE): return label |
| 120 | return "unknown.other" |
| 121 | |
| 122 | def retry_table(workspace, table, required, current_err, current_alt_value=None): |
| 123 | """Try alternate filter values for one table. Return (success, new_err, new_fmap, attempts).""" |
| 124 | # Build candidate filter sets |
| 125 | candidates = [] |
| 126 | if required: |
| 127 | # Take current default |
| 128 | base = {rf: DEFAULT_FILTER_VALUES.get(rf, "1") for rf in required} |
| 129 | candidates.append(base.copy()) |
| 130 | # Try alternating owner |
| 131 | for alt_o in [ALT_OWNER, "sst", "opencode-ai", "octocat"]: |
| 132 | for alt_r in ["opencode", "linux", "Hello-World"]: |
| 133 | cand = base.copy() |
| 134 | if "owner" in cand: cand["owner"] = alt_o |
| 135 | if "repo" in cand: cand["repo"] = alt_r |
| 136 | if "username" in cand: cand["username"] = alt_o |
| 137 | if "org" in cand: cand["org"] = alt_o |
| 138 | if cand != base and cand not in candidates: |
| 139 | candidates.append(cand) |
| 140 | # No-filter tables: just retry with different limits |
| 141 | if not required: |
| 142 | return False, current_err, {}, 0 |
| 143 | |
| 144 | for i, fmap in enumerate(candidates[:8]): |
| 145 | where = " AND ".join(f"{k} = {quote(v)}" for k, v in fmap.items()) |
| 146 | sql = f"SELECT * FROM github.{table} WHERE {where} LIMIT 5" |
| 147 | wall, is_err, err_msg, n_rows, _ = run_sql(sql, workspace) |
| 148 | if not is_err and n_rows > 0: |
| 149 | return True, None, fmap, i + 1 |
| 150 | return False, current_err, {}, len(candidates[:8]) |
| 151 | |
| 152 | def main(): |
| 153 | ap = argparse.ArgumentParser() |
| 154 | ap.add_argument("--source", required=True) |
| 155 | ap.add_argument("--workers", type=int, default=4) |
| 156 | ap.add_argument("--only-category", default=None, help="only retry tables in this error category") |
| 157 | ap.add_argument("--workspace", default=os.environ.get("CORAL_WORKSPACE", "default")) |
| 158 | args = ap.parse_args() |
| 159 | |
| 160 | analysis = json.loads(Path(f"reports/{args.source}/analysis.json").read_text()) |
| 161 | catalog = json.loads(Path(f"reports/{args.source}/catalog.json").read_text()) |
| 162 | |
| 163 | retry_targets = [] |
| 164 | for tname, info in analysis["per_table"].items(): |
| 165 | if info["verdict"] is True: continue |
| 166 | first_cat = info["failures"][0]["category"] if info["failures"] else None |
| 167 | if not first_cat: continue |
| 168 | if first_cat in SKIP_CATEGORIES: continue |
| 169 | if args.only_category and first_cat != args.only_category: continue |
| 170 | short = tname.split(".",1)[1] |
| 171 | rec = next((t for t in catalog["tables"] if t["name"] == short), None) |
| 172 | if not rec: continue |
| 173 | if not rec.get("required_filters"): continue # skip no-filter |
| 174 | retry_targets.append((short, rec["required_filters"], info["failures"][0].get("error", ""))) |
| 175 | |
| 176 | print(f"retrying {len(retry_targets)} tables", file=sys.stderr) |
| 177 | |
| 178 | def worker(t): |
| 179 | short, req, err = t |
| 180 | ok, new_err, fmap, attempts = retry_table(args.workspace, short, req, err) |
| 181 | return short, ok, new_err, fmap, attempts |
| 182 | |
| 183 | successes = 0 |
| 184 | failures = [] |
| 185 | with ThreadPoolExecutor(max_workers=args.workers) as ex: |
| 186 | futs = {ex.submit(worker, t): t for t in retry_targets} |
| 187 | for i, fut in enumerate(as_completed(futs)): |
| 188 | short, ok, new_err, fmap, attempts = fut.result() |
| 189 | if ok: |
| 190 | successes += 1 |
| 191 | # Update report |
| 192 | jf = Path(f"reports/{args.source}/tables/{short}.json") |
| 193 | if jf.exists(): |
| 194 | rec = json.loads(jf.read_text()) |
| 195 | rec["verdict"] = {"pass": True, "notes": f"Retry succeeded after {attempts} attempts with {fmap}"} |
| 196 | rec["retried_filters"] = fmap |
| 197 | jf.write_text(json.dumps(rec, indent=2) + "\n") |
| 198 | print(f"✓ [{i+1}/{len(retry_targets)}] {short}: retry worked with {fmap}", file=sys.stderr) |
| 199 | else: |
| 200 | failures.append(short) |
| 201 | print(f"✗ [{i+1}/{len(retry_targets)}] {short}: still failing ({attempts} attempts)", file=sys.stderr) |
| 202 | |
| 203 | print(f"\n{successes} improved, {len(failures)} still failing", file=sys.stderr) |
| 204 | |
| 205 | if __name__ == "__main__": |
| 206 | main() |