# /// script # requires-python = ">=3.10" # dependencies = ["duckdb", "huggingface_hub", "requests", "polars", "pyarrow", "pyyaml", "lightgbm"] # /// """Incremental self-refresh for the dataset-lineage graph (scheduled Job). Two phases in one script, chained across two Jobs so the GPU only runs when there is something to judge: cpu (cpu-performance, scheduled monthly) fresh metadata/declared/created_at from the cards dump -> candidates for datasets CREATED SINCE the last cutoff (name-prefix, cross-org collision, card mentions) -> content-verify -> push pending parquet to the scratch repo -> submit the gpu phase via run_uv_job (or finalize directly when nothing new verified). gpu (a100-large, vllm-openai image — same recipe as judge_vllm.py) LLM-judge the pending edges (same model/prompt as the base corpus) -> apply verdicts -> merge into canonical verified_edges.parquet -> rebuild lineage.db -> push to the Space repo (auto-redeploy) -> update refresh/state.json. State lives in refresh/state.json in the private scratch dataset. All intermediate refresh artifacts live under refresh/ so they never collide with the historical verified_edges* slices. Schedule (script self-updates from the public Space repo): hf jobs scheduled uv run "0 6 1 * *" --flavor cpu-performance \ --timeout 4h --secrets HF_TOKEN \ https://huggingface.co/spaces/davanstrien/dataset-lineage-explorer/raw/main/refresh_job.py Smoke test: hf jobs uv run --flavor cpu-performance --timeout 45m --secrets HF_TOKEN \ -d jobs/refresh_job.py --phase cpu --smoke --no-chain """ import argparse import json import os import re import threading import time from collections import defaultdict, deque from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass, field import duckdb import polars as pl import requests import yaml from huggingface_hub import HfApi, hf_hub_download GLOB = "hf://datasets/librarian-bots/dataset_cards_with_metadata/**/*.parquet" MODEL_GLOB = "hf://datasets/librarian-bots/model_cards_with_metadata/data/*.parquet" MODEL_TYPED_RELS = ("finetune", "adapter", "quantized", "merge") MODEL_EDGE_TYPES = {"finetune", "adapter", "quantized", "merge", "trained_on"} IMPACT_MAX_HOPS = 4 # model->model derivation hops beyond the directly-trained models TREND_MAX_HOPS = 4 # upward ancestry hops for trending enablement TREND_TOP_N = 1000 # trending models to fetch from the Hub API (dump has no trending_score) TREND_TOP_ENABLE = 50 # trending models (with >=1 parent) to precompute ancestry for SCRATCH = "davanstrien/dataset-lineage-scratch" SPACE = "davanstrien/dataset-lineage-explorer" DISTILL_REPO = "davanstrien/lineage-judge-distilled" # shadow CPU classifier (model.txt + features.py) SCRIPT_URL = f"https://huggingface.co/spaces/{SPACE}/raw/main/refresh_job.py" MODEL = "Qwen/Qwen3.6-35B-A3B" BASE = "https://datasets-server.huggingface.co" TOKEN = os.environ.get("HF_TOKEN") HEADERS = {"Authorization": f"Bearer {TOKEN}"} if TOKEN else {} STATE_PATH = "refresh/state.json" DEFAULT_CUTOFF = "2026-06-12T00:00:00" # base-corpus snapshot date TRIVIAL = {"unrelated", "related_structure"} # vLLM image pinned by DIGEST (not the mutable :latest tag) so the unattended monthly # gpu phase can't silently break when the tag moves under a new Python. Verified-good # config 2026-07-03: THIS digest + python=/usr/bin/python3 + PYTHONPATH below. Re-pin # deliberately from `docker hub .../vllm-openai/tags/latest` digest when bumping vLLM. GPU_IMAGE = "vllm/vllm-openai@sha256:251eba5cc7c12fed0b75da22a9240e582b1c9e39f6fbc064f86781b963bd814f" GPU_DIST_PACKAGES = "/usr/local/lib/python3.12/dist-packages" INFLIGHT_MAX_AGE_H = 36 # a gpu phase in flight longer than this is presumed dead GPU_POLL_INTERVAL_S = 60 GPU_POLL_MAX_S = 2.5 * 3600 # cpu phase blocks at most this long awaiting the gpu job KEEP_STAMPS = 12 # refresh/ artifact + state["runs"] retention MAX_DUMP_AGE_DAYS = 10 # a cards dump older than this is unactionable -> abort+alert RETRY_MAX_FAILS = 3 # drop a pair from the retry queue after this many failed tries # canonical column set every merge into verified_edges must preserve (schema guard) CORE_COLS = [ "child", "parent", "ok", "primary_type", "tags", "confidence", "schema_jaccard", "size_ratio", "col_containment", "inherited_cols", "changed_cols", "language_shift", "child_rows", "parent_rows", "note", ] PARENT_CAP = 8_000 CHILD_N = 1_000 MIN_PREFIX_LEN = 4 MAX_MENTIONS = 10 MENTION_RE = r"huggingface\.co/datasets/([A-Za-z0-9][A-Za-z0-9_.\-]*/[A-Za-z0-9][A-Za-z0-9_.\-]*)" LOAD_RE = r"""load_dataset\(\s*['"]([A-Za-z0-9][A-Za-z0-9_.\-]*/[A-Za-z0-9][A-Za-z0-9_.\-]*)['"]""" TEMPLATE_NOISE = { "original", "extended", "extended|other", "crowdsourced", "found", "machine-generated", "other", "unknown", } LABEL_ENUM = [ "exact_copy", "filtered_subset", "augmentation", "cleaned", "regenerated_variant", "translation", "reformat", "combined", "unrelated", ] LABEL_SCHEMA = { "type": "object", "properties": { "label": {"type": "string", "enum": LABEL_ENUM}, "reason": {"type": "string"}, }, "required": ["label", "reason"], "additionalProperties": False, } api = HfApi(token=TOKEN) _tls = threading.local() def alert(title, body=""): """Push an alert to Daniel by opening an HF Discussion on the Space repo (the Hub emails watchers by default — that's our notification channel). Best-effort: a failure to post must never mask the real error that triggered the alert.""" try: api.create_discussion(repo_id=SCRATCH, repo_type="dataset", title=f"[refresh-alert] {title}", description=body or title) print(f"ALERT posted: {title}", flush=True) except Exception as e: # noqa: BLE001 print(f"ALERT (discussion post failed: {e}): {title}\n{body}", flush=True) def _iso(s): """Canonical second-precision ISO-8601 string for every cutoff/createdAt compare and persist, so a space- vs 'T'-separated dump (or a native-timestamp column that str()s to a space form) can't skew the boundary. 'YYYY-MM-DDTHH:MM:SS'.""" return str(s)[:19].replace(" ", "T") # -------------------------------------------------------------------------- # shared plumbing (verbatim from verify_job.py) # -------------------------------------------------------------------------- def con(): c = getattr(_tls, "c", None) if c is None: c = duckdb.connect() c.execute("INSTALL httpfs; LOAD httpfs;") c.execute("SET http_timeout=30000; SET http_retries=2;") c.execute("SET threads=1; SET memory_limit='512MB';") if TOKEN: c.execute(f"CREATE SECRET hf (TYPE huggingface, TOKEN '{TOKEN}');") _tls.c = c return c def _get(path, params, retries=3): for i in range(retries): try: r = requests.get(f"{BASE}/{path}", params=params, headers=HEADERS, timeout=30) if r.status_code == 200: return r.json() if r.status_code in (404, 422, 501): return None time.sleep(1.0 * (i + 1)) except requests.RequestException: time.sleep(1.0 * (i + 1)) return None def parquet_urls(dataset): js = _get("parquet", {"dataset": dataset}) if not js or not js.get("parquet_files"): return None files = js["parquet_files"] configs = [f["config"] for f in files] cfg = "default" if "default" in configs else configs[0] sp_files = [f for f in files if f["config"] == cfg] splits = [f["split"] for f in sp_files] sp = "train" if "train" in splits else splits[0] urls = [f["url"] for f in sp_files if f["split"] == sp] return (urls[:1], cfg, sp) if urls else None def get_num_rows(dataset, config, split): js = _get("size", {"dataset": dataset}) if not js: return None for s in js.get("size", {}).get("splits", []): if s.get("config") == config and s.get("split") == split: return s.get("num_rows") return None def _norm(v): return " ".join(v.split()).lower() def _ascii_ratio(s): return 1.0 if not s else sum(c.isascii() for c in s) / len(s) def fetch_string_cols(dataset, max_rows): tgt = parquet_urls(dataset) if not tgt: return None, "no parquet" urls, cfg, sp = tgt cur = con() try: schema = cur.execute(f"DESCRIBE SELECT * FROM read_parquet({urls}) LIMIT 1").fetchall() except Exception as e: # noqa: BLE001 return None, f"describe: {e}" strcols = [r[0] for r in schema if r[1] == "VARCHAR"] listcols = [r[0] for r in schema if r[1] == "VARCHAR[]"] allcols = strcols + listcols nrows = get_num_rows(dataset, cfg, sp) if not allcols: return ([], {}, {}, nrows), None collist = ", ".join(f'"{c}"' for c in allcols) try: rows = cur.execute(f"SELECT {collist} FROM read_parquet({urls}) LIMIT {max_rows}").fetchall() except Exception as e: # noqa: BLE001 return None, f"select: {e}" listset = set(listcols) strcols = allcols vals = {c: [] for c in strcols} ascii_acc = {c: [] for c in strcols} for row in rows: for c, v in zip(strcols, row): if c in listset: for el in (v or []): if isinstance(el, str) and el.strip(): vals[c].append(_norm(el)) if len(ascii_acc[c]) < 200: ascii_acc[c].append(_ascii_ratio(el)) continue if isinstance(v, str) and v.strip(): vals[c].append(_norm(v)) if len(ascii_acc[c]) < 200: ascii_acc[c].append(_ascii_ratio(v)) ascii_r = {c: (sum(a) / len(a) if a else 1.0) for c, a in ascii_acc.items()} return (strcols, vals, ascii_r, nrows), None @dataclass class ParentProfile: dataset: str ok: bool cols: list = field(default_factory=list) sets: dict = field(default_factory=dict) ascii: dict = field(default_factory=dict) num_rows: int | None = None note: str = "" def build_parent_profile(parent): res, err = fetch_string_cols(parent, PARENT_CAP) if not res: return ParentProfile(parent, ok=False, note=err) cols, vals, ascii_r, nrows = res return ParentProfile(parent, ok=True, cols=cols, sets={c: set(v) for c, v in vals.items() if v}, ascii=ascii_r, num_rows=nrows) @dataclass class Result: child: str parent: str ok: bool schema_jaccard: float = 0.0 size_ratio: float | None = None col_containment: dict = field(default_factory=dict) inherited_cols: list = field(default_factory=list) changed_cols: list = field(default_factory=list) language_shift: bool = False child_rows: int | None = None parent_rows: int | None = None primary_type: str = "unknown" tags: list = field(default_factory=list) confidence: float = 0.0 note: str = "" def hint_tags(hint): h = (hint or "").lower() t = set() rules = [ ("clean", "cleaned"), ("dedup", "dedup"), ("filter", "filtered"), ("sample", "subset"), ("subset", "subset"), ("mini", "subset"), ("small", "subset"), ("tiny", "subset"), ("translat", "translated"), ("format", "reformatted"), ("chatml", "reformatted"), ("sharegpt", "reformatted"), ("messages", "reformatted"), ("conversation", "reformatted"), ("dpo", "preference"), ("orpo", "preference"), ("kto", "preference"), ("preference", "preference"), ("pairs", "preference"), ("sft", "sft"), ("instruct", "sft"), ("tokeniz", "tokenized"), ("embed", "embeddings"), ("augment", "augmented"), ("expand", "augmented"), ("synthetic", "augmented"), ("distil", "augmented"), ("gpt4", "regenerated"), ("gpt-4", "regenerated"), ("rewrit", "regenerated"), ("merge", "combined"), ("combined", "combined"), ("mix", "combined"), ("balanc", "rebalanced"), ] for needle, tag in rules: if needle in h: t.add(tag) if re.search(r"(^|[-_.])v?\d+(\.\d+)?($|[-_.])", h) or re.search(r"\d+k\b", h): t.add("versioned") return t def classify(r, hint): conts = list(r.col_containment.values()) sr = r.size_ratio htags = hint_tags(hint) tags = set(htags) if not conts: if r.schema_jaccard >= 0.6 and (sr is None or 0.5 <= sr <= 2.0): tags.add("related_structure") return next(iter(htags), "related_structure"), sorted(tags), 0.45 return "unrelated", sorted(tags | {"unrelated"}), 0.6 maxc, minc = max(conts), min(conts) if maxc >= 0.85: if minc >= 0.85: tags.add("inherited_all") if sr is None or 0.97 <= sr <= 1.03: return "exact_copy", sorted(tags | {"copy"}), 0.92 if sr < 0.97: return "filtered_subset", sorted(tags | {"subset"}), 0.85 return "augmentation", sorted(tags | {"superset"}), 0.7 tags.add("inputs_inherited") changed_overlap = min(conts) if sr is None or 0.9 <= sr <= 1.1: if "cleaned" in htags or 0.5 <= changed_overlap < 0.85: return "cleaned", sorted(tags | {"outputs_edited"}), 0.75 return "regenerated_variant", sorted(tags | {"outputs_changed"}), 0.8 if sr < 0.9: return "subset_modified", sorted(tags | {"subset"}), 0.7 return "augmentation_modified", sorted(tags | {"superset"}), 0.6 if 0.4 <= maxc < 0.85: tags.add("partial_overlap") if r.schema_jaccard < 0.8: return "reformat", sorted(tags | {"reformatted"}), 0.6 return "partial_overlap", sorted(tags), 0.55 if r.schema_jaccard >= 0.6 and sr is not None and 0.6 <= sr <= 1.4: if r.language_shift or "translated" in htags: return "translation", sorted(tags | {"translated"}), 0.7 return "modified_variant", sorted(tags | {"modified"}), 0.5 if sr is not None and sr < 0.3 and r.schema_jaccard >= 0.6: return "subset_reformatted", sorted(tags | {"subset", "reformatted"}), 0.5 return "unrelated", sorted(tags | {"unrelated"}), 0.55 def verify_child(child, profile, hint=""): if not profile.ok: return Result(child, profile.dataset, ok=False, note=f"parent: {profile.note}") res, err = fetch_string_cols(child, CHILD_N) if not res: return Result(child, profile.dataset, ok=False, note=f"child: {err}") c_cols, c_vals, c_ascii, c_n = res schema_j = ( len(set(c_cols) & set(profile.cols)) / len(set(c_cols) | set(profile.cols)) if (c_cols or profile.cols) else 0.0 ) col_cont = {} for c in c_cols: if c in profile.sets and c_vals.get(c): cv = c_vals[c] col_cont[c] = round(sum(1 for x in cv if x in profile.sets[c]) / len(cv), 3) lang_shift = False shared = [c for c in c_cols if c in profile.cols] if shared: key = max(shared, key=lambda c: sum(len(x) for x in c_vals.get(c, [])) / max(len(c_vals.get(c, [])), 1)) lang_shift = profile.ascii.get(key, 1.0) > 0.8 and c_ascii.get(key, 1.0) < 0.5 sr = (c_n / profile.num_rows) if (c_n and profile.num_rows) else None r = Result( child=child, parent=profile.dataset, ok=True, schema_jaccard=round(schema_j, 3), size_ratio=round(sr, 4) if sr else None, col_containment=col_cont, inherited_cols=[c for c, v in col_cont.items() if v >= 0.85], changed_cols=[c for c, v in col_cont.items() if v < 0.4], language_shift=lang_shift, child_rows=c_n, parent_rows=profile.num_rows, ) r.primary_type, r.tags, r.confidence = classify(r, hint) return r def verify_one_parent(parent, children): profile = build_parent_profile(parent) out = [] for child, hint in children: try: out.append(verify_child(child, profile, hint)) except Exception as e: # noqa: BLE001 out.append(Result(child, parent, ok=False, note=f"error: {e}")) return out def run_grouped(pairs, workers, label="", checkpoint=None, checkpoint_every=300, chunk_size=200): """Verify pairs grouped by parent. Parents with more than chunk_size children are split into chunks so one mega-family can't serialize the whole tail (the parent profile is re-fetched per chunk — cheap vs the hours a 5k-child parent costs on a single thread).""" by_parent = defaultdict(list) for child, parent, hint in pairs: by_parent[parent].append((child, hint)) tasks = [] for p, ch in by_parent.items(): for i in range(0, len(ch), chunk_size): tasks.append((p, ch[i:i + chunk_size])) print(f"{label}verifying {len(pairs)} edges / {len(by_parent)} parents / " f"{len(tasks)} chunks / {workers} workers", flush=True) results, done, t0 = [], 0, time.time() with ThreadPoolExecutor(max_workers=workers) as ex: futs = {ex.submit(verify_one_parent, p, ch): p for p, ch in tasks} for fut in as_completed(futs): results.extend(fut.result()) done += 1 if done % 100 == 0 or done == len(tasks): ok = sum(1 for r in results if r.ok) print(f" {label}{done}/{len(tasks)} edges={len(results)} ok={ok} ({time.time() - t0:.0f}s)", flush=True) if checkpoint and done % checkpoint_every == 0: try: checkpoint(results) except Exception as e: # noqa: BLE001 print(f" checkpoint failed: {e}", flush=True) return results # explicit schema so an EMPTY result set still yields a well-typed frame with the # expected columns (a 0x0 frame would raise ColumnNotFoundError downstream) _RESULT_SCHEMA = { "child": pl.Utf8, "parent": pl.Utf8, "ok": pl.Boolean, "primary_type": pl.Utf8, "tags": pl.List(pl.Utf8), "confidence": pl.Float64, "schema_jaccard": pl.Float64, "size_ratio": pl.Float64, "col_containment": pl.Utf8, "inherited_cols": pl.List(pl.Utf8), "changed_cols": pl.List(pl.Utf8), "language_shift": pl.Boolean, "child_rows": pl.Int64, "parent_rows": pl.Int64, "note": pl.Utf8, "hint": pl.Utf8, } def results_to_df(results, hints): rows = [] for r in results: rows.append({ "child": r.child, "parent": r.parent, "ok": r.ok, "primary_type": r.primary_type, "tags": list(r.tags), "confidence": r.confidence, "schema_jaccard": r.schema_jaccard, "size_ratio": r.size_ratio, "col_containment": json.dumps(r.col_containment), "inherited_cols": r.inherited_cols, "changed_cols": r.changed_cols, "language_shift": r.language_shift, "child_rows": r.child_rows, "parent_rows": r.parent_rows, "note": r.note, "hint": hints.get((r.child, r.parent), ""), }) return pl.DataFrame(rows, schema=_RESULT_SCHEMA) # -------------------------------------------------------------------------- # scratch-repo I/O # -------------------------------------------------------------------------- def load_scratch(name): return pl.read_parquet(hf_hub_download(SCRATCH, name, repo_type="dataset", token=TOKEN)) def push_df(df, name, msg=""): path = f"/tmp/{os.path.basename(name)}" df.write_parquet(path) api.upload_file(path_or_fileobj=path, path_in_repo=name, repo_id=SCRATCH, repo_type="dataset", commit_message=msg or f"refresh: {name}") print(f"pushed {name}: {len(df)} rows", flush=True) def load_state(): try: p = hf_hub_download(SCRATCH, STATE_PATH, repo_type="dataset", token=TOKEN) return json.load(open(p)) except Exception: # noqa: BLE001 return {"cutoff": DEFAULT_CUTOFF, "runs": []} def save_state(state): with open("/tmp/state.json", "w") as f: json.dump(state, f, indent=2) api.upload_file(path_or_fileobj="/tmp/state.json", path_in_repo=STATE_PATH, repo_id=SCRATCH, repo_type="dataset", commit_message=f"refresh: state cutoff={state['cutoff']}") # -------------------------------------------------------------------------- # chaining / lease / retry-queue / housekeeping helpers # -------------------------------------------------------------------------- def _inflight_path(stamp): return f"refresh/inflight_{stamp}.json" def write_inflight(stamp, job_id): """Write the inflight marker (our lease) so a concurrent cpu phase can detect the still-running gpu job, and a watchdog can spot a gpu phase that died.""" payload = {"gpu_job_id": job_id, "stamp": stamp, "submitted_at": time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime())} with open("/tmp/inflight.json", "w") as f: json.dump(payload, f) api.upload_file(path_or_fileobj="/tmp/inflight.json", path_in_repo=_inflight_path(stamp), repo_id=SCRATCH, repo_type="dataset", commit_message=f"refresh: inflight {stamp}") def read_any_inflight(): """(path, payload) of the most recent refresh/inflight_*.json in scratch, or (None, None). Best-effort; a scan failure returns (None, None) so it never blocks.""" try: files = [f for f in api.list_repo_files(SCRATCH, repo_type="dataset") if f.startswith("refresh/inflight_") and f.endswith(".json")] except Exception as e: # noqa: BLE001 print(f" (inflight scan failed: {e})", flush=True) return None, None if not files: return None, None path = sorted(files)[-1] try: p = hf_hub_download(SCRATCH, path, repo_type="dataset", token=TOKEN) return path, json.load(open(p)) except Exception: # noqa: BLE001 return path, None def delete_inflight(stamp): """Clear the lease. Quiet no-op when there is no marker (cpu-direct finalize).""" try: api.delete_file(path_in_repo=_inflight_path(stamp), repo_id=SCRATCH, repo_type="dataset", commit_message=f"refresh: clear inflight {stamp}") except Exception: # noqa: BLE001 — usually just "no such file" on the no-gpu path pass def _age_hours(iso_str): try: t = time.mktime(time.strptime(_iso(iso_str), "%Y-%m-%dT%H:%M:%S")) return (time.time() - t) / 3600 except Exception: # noqa: BLE001 return 1e9 # unparseable -> treat as ancient so we don't block forever def _inflight_interlock(stamp): """Refuse to start a fresh cpu phase while a prior gpu phase is still in flight (double-run guard). A marker older than INFLIGHT_MAX_AGE_H means that gpu phase died without finalizing: clear it and proceed.""" path, payload = read_any_inflight() if not path: return age = _age_hours((payload or {}).get("submitted_at", "")) if age < INFLIGHT_MAX_AGE_H: alert("prior refresh still in flight", f"{path} age={age:.1f}h (< {INFLIGHT_MAX_AGE_H}h). Skipping this cpu phase " "to avoid a duplicate gpu run. If that job is actually dead, delete the " "marker in the scratch repo.") raise SystemExit("prior refresh in flight — skipping") alert("stale inflight marker cleared", f"{path} age={age:.1f}h (> {INFLIGHT_MAX_AGE_H}h). The prior gpu phase likely " "died without finalizing; clearing the marker and proceeding.") try: api.delete_file(path_in_repo=path, repo_id=SCRATCH, repo_type="dataset", commit_message="refresh: clear stale inflight") except Exception as e: # noqa: BLE001 print(f" (stale inflight delete failed: {e})", flush=True) def poll_gpu(job_id, stamp): """Block until the chained gpu job reaches a terminal state, so a gpu failure is NOT silent (the cpu job would otherwise report success while nothing updated). On ERROR/CANCELED/DELETED or a poll timeout -> alert + non-zero exit.""" t0 = time.time() while time.time() - t0 < GPU_POLL_MAX_S: time.sleep(GPU_POLL_INTERVAL_S) try: info = api.inspect_job(job_id=job_id) stage = getattr(getattr(info, "status", None), "stage", None) except Exception as e: # noqa: BLE001 print(f" gpu poll: inspect failed ({e}); retrying", flush=True) continue stage_s = getattr(stage, "value", stage) print(f" gpu {job_id}: {stage_s} ({time.time() - t0:.0f}s)", flush=True) if stage == "COMPLETED": print("gpu phase completed", flush=True) return if stage in ("ERROR", "CANCELED", "DELETED"): alert("gpu phase FAILED", f"gpu job {job_id} (stamp {stamp}) ended in {stage_s}. The graph was " "NOT updated this run; verified_edges/state are untouched. Check the " f"job logs at https://huggingface.co/jobs/{job_id}.") raise SystemExit(f"gpu job {job_id} {stage_s}") alert("gpu phase poll timed out", f"gpu job {job_id} (stamp {stamp}) did not reach a terminal state within " f"{GPU_POLL_MAX_S / 3600:.1f}h. It may still finish; the inflight marker is " "left so the watchdog/next run can reconcile.") raise SystemExit(f"gpu job {job_id} poll timeout") def load_retry_queue(): """(pairs, prior_fail) from refresh/retry_pairs.parquet — the transient-failure pairs carried from earlier runs, plus their fail counts. ([], {}) when absent.""" try: rq = load_scratch("refresh/retry_pairs.parquet") except Exception: # noqa: BLE001 return [], {} pairs, fails = [], {} for r in rq.iter_rows(named=True): pairs.append((r["child"], r["parent"], r.get("hint") or "")) fails[(r["child"], r["parent"])] = int(r.get("fail_count") or 0) return pairs, fails def save_retry_queue(df, prior_fail): """Persist pairs that FAILED verification this run (fail_count+1, dropped at RETRY_MAX_FAILS). Pairs that succeeded simply don't reappear here, which clears them; writing the (possibly empty) file each run keeps the queue current.""" rows = [] for r in df.filter(~pl.col("ok")).iter_rows(named=True): key = (r["child"], r["parent"]) fc = prior_fail.get(key, 0) + 1 if fc >= RETRY_MAX_FAILS: continue rows.append({"child": r["child"], "parent": r["parent"], "hint": r.get("hint") or "", "fail_count": fc}) rq = pl.DataFrame(rows, schema={"child": pl.Utf8, "parent": pl.Utf8, "hint": pl.Utf8, "fail_count": pl.Int64}) push_df(rq, "refresh/retry_pairs.parquet", "refresh: retry queue") def prune_scratch(): """Keep only the newest KEEP_STAMPS artifacts per refresh/ family; best-effort.""" try: files = api.list_repo_files(SCRATCH, repo_type="dataset") except Exception as e: # noqa: BLE001 print(f" (prune skipped: {e})", flush=True) return prefixes = ["refresh/pending_partial_", "refresh/pending_", "refresh/judged_", "refresh/added_", "refresh/verified_backup_", "refresh/inflight_"] for pre in prefixes: # a stamp always begins with a digit (the year) — this keeps 'pending_' from # also matching 'pending_partial_' (handled by its own, earlier prefix) fam = sorted(f for f in files if f.startswith(pre) and f[len(pre):len(pre) + 1].isdigit()) for f in fam[:-KEEP_STAMPS]: try: api.delete_file(path_in_repo=f, repo_id=SCRATCH, repo_type="dataset", commit_message=f"refresh: prune {f}") except Exception as e: # noqa: BLE001 print(f" (prune {f} failed: {e})", flush=True) def _shrink_guard(n_edges, stamp): """Refuse to deploy (SystemExit) if the rebuilt graph shrank >2% vs the live Space, OR if live /stats can't be read (fail CLOSED). Called BEFORE any canonical write, so an abort here leaves the source of truth untouched.""" try: live = requests.get(f"https://{SPACE.replace('/', '-')}.hf.space/stats", timeout=30).json() except (requests.RequestException, ValueError) as e: alert("no-shrink guard unavailable", f"stamp={stamp}: could not read live /stats ({e}). Failing closed — not " "deploying (nothing was overwritten). Re-run once the Space is reachable.") raise SystemExit(f"no-shrink guard unavailable: {e}") live_edges = _total_edges(live) # robust to old {edges:N} and ecosystem stats shapes if live_edges and n_edges < 0.98 * live_edges: alert("rebuilt graph shrank — deploy refused", f"stamp={stamp}: rebuilt {n_edges:,} edges vs live {live_edges:,} (>2% " "shrink). An incremental refresh cannot shrink the graph. Check " "verified_edges.parquet, ecosystem/model_*.parquet, cards-dump freshness.") raise SystemExit( f"REFUSING deploy: rebuilt {n_edges:,} edges < 0.98 * live {live_edges:,}") print(f"no-shrink guard OK: rebuilt {n_edges:,} vs live {live_edges:,}", flush=True) # -------------------------------------------------------------------------- # CPU phase: snapshot -> candidates -> verify -> hand off # -------------------------------------------------------------------------- def extract_yaml_frontmatter(card): if not card or not isinstance(card, str): return None m = re.match(r"^---\s*\n(.*?)\n---", card, re.DOTALL) if not m: return None try: d = yaml.safe_load(m.group(1)) return d if isinstance(d, dict) else None except Exception: # noqa: BLE001 return None def declared_from_cards(c, limit=""): """Re-extract declared source_datasets edges from the cards dump.""" rows = c.execute( f"SELECT datasetId, downloads, card FROM read_parquet('{GLOB}') " f"WHERE card LIKE '%source_datasets%' {limit}" ).fetchall() out = [] for did, dl, card in rows: fm = extract_yaml_frontmatter(card) if not fm: continue src = fm.get("source_datasets") if isinstance(src, str): src = [src] if not isinstance(src, list): continue for s in src: if not isinstance(s, str) or not s.strip(): continue s = s.strip() if s.lower() in TEMPLATE_NOISE: continue kind = "org_name" if re.fullmatch(r"[\w.\-]+/[\w.\-]+", s) else "other" out.append({"child": did, "parent": s, "parent_kind": kind, "downloads": dl or 0}) return pl.DataFrame(out) if out else pl.DataFrame( schema={"child": pl.Utf8, "parent": pl.Utf8, "parent_kind": pl.Utf8, "downloads": pl.Int64}) def name_part(did): return did.split("/", 1)[1].lower() if "/" in did else did.lower() def org_part(did): return did.split("/", 1)[0].lower() if "/" in did else None def phase_cpu(args): # seconds in the stamp so two runs in the same minute can't collide on artifact # names / the chained --stamp handoff stamp = args.stamp or time.strftime("%Y%m%d%H%M%S", time.gmtime()) smoke = args.smoke limit = "LIMIT 30000" if smoke else "" state = load_state() cutoff = _iso(args.cutoff or state["cutoff"]) print(f"== cpu phase: stamp={stamp} cutoff={cutoff} smoke={smoke} ==", flush=True) # double-run interlock: don't start while a prior gpu phase is still in flight # (skips smoke, which never writes canonical state or an inflight marker) if not smoke: _inflight_interlock(stamp) # carry-over queue of pairs that failed verification transiently in earlier runs retry_pairs, prior_fail = ([], {}) if not smoke: retry_pairs, prior_fail = load_retry_queue() if retry_pairs: print(f" retry queue: {len(retry_pairs)} pairs carried over", flush=True) c = duckdb.connect() c.execute("INSTALL httpfs; LOAD httpfs;") if TOKEN: c.execute(f"CREATE SECRET hf (TYPE huggingface, TOKEN '{TOKEN}');") print("pulling metadata (full corpus)...", flush=True) meta = c.execute( f"SELECT datasetId, author, downloads, likes, trending_score, createdAt " f"FROM read_parquet('{GLOB}') {limit}" ).pl() # canonicalize createdAt to second-precision ISO strings so every cutoff compare # is format-stable whether the dump stores createdAt as a string or a timestamp if "createdAt" in meta.columns: meta = meta.with_columns( pl.col("createdAt").cast(pl.Utf8).str.slice(0, 19) .str.replace(" ", "T", literal=True).alias("createdAt") ) print(f" {len(meta):,} datasets", flush=True) newest = meta["createdAt"].max() if "createdAt" in meta.columns else None new = meta.filter(pl.col("createdAt") > cutoff) # cutoff must never move backwards (a stale dump has an old max createdAt) max_created = _iso(max(str(newest or cutoff), str(cutoff))) print(f" {len(new):,} created since cutoff (max createdAt {max_created})", flush=True) # freshness is a HARD gate: a stale/empty dump is unactionable, and a failed Job # (red on the dashboard) + a Discussion IS the alert — no silent stale refresh. if newest is None: alert("empty cards dump", f"stamp={stamp}: the dump has no usable createdAt — read likely failed.") raise SystemExit("empty/unreadable cards dump — aborting") dump_age_days = (time.time() - time.mktime(time.strptime(_iso(newest)[:10], "%Y-%m-%d"))) / 86400 if dump_age_days > MAX_DUMP_AGE_DAYS: alert("stale cards dump", f"stamp={stamp}: newest dataset is {dump_age_days:.0f} days old " f"(> {MAX_DUMP_AGE_DAYS}). The librarian-bots card-pipeline compile job is " "likely stuck; there is nothing fresh to refresh. Aborting.") raise SystemExit(f"stale cards dump ({dump_age_days:.0f}d) — aborting") print("re-extracting declared edges...", flush=True) declared = declared_from_cards(c, limit) print(f" {len(declared):,} declared edges", flush=True) # refresh the ecosystem model layer (its own dump scan, ~23s) and replace the # canonical ecosystem/model_*.parquet in scratch. skipped in smoke. if not smoke: print("harvesting model layer (ecosystem)...", flush=True) harvest_model_layer() print("scanning new cards for mentions...", flush=True) mentions_df = c.execute( f""" SELECT datasetId, regexp_extract_all(card, $rx${MENTION_RE}$rx$, 1) AS links, regexp_extract_all(card, $rx${LOAD_RE}$rx$, 1) AS loads FROM read_parquet('{GLOB}') WHERE createdAt > ? AND (card LIKE '%huggingface.co/datasets/%' OR card LIKE '%load_dataset(%') {limit} """, [cutoff], ).pl() print(f" {len(mentions_df):,} new cards with mentions", flush=True) # ---- candidates for new children ------------------------------------- ids = meta["datasetId"].to_list() dls = dict(zip(ids, meta["downloads"].to_list())) lks = dict(zip(ids, meta["likes"].to_list())) id_set = set(ids) lower_to_id = {} name_best, name_to_ids = {}, {} for did in ids: lower_to_id.setdefault(did.lower(), did) nm = name_part(did) name_to_ids.setdefault(nm, []).append(did) if nm not in name_best or (dls.get(did) or 0) > (dls.get(name_best[nm]) or 0): name_best[nm] = did declared_pairs = set(zip(declared["child"].to_list(), declared["parent"].to_list())) cands = {} def add(child, parent, signal, hint): if not parent or parent == child: return key = (child, parent) if key in cands: cands[key]["signal"] += "+" + signal return cands[key] = { "child": child, "parent": parent, "child_dl": dls.get(child) or 0, "parent_dl": dls.get(parent) or 0, "signal": signal, "hint": hint, "declared": key in declared_pairs, } for did in new["datasetId"].to_list(): nm = name_part(did) best_prefix = None for m in re.finditer(r"[-_.]", nm): prefix = nm[: m.start()] if len(prefix) >= MIN_PREFIX_LEN and prefix in name_best and name_best[prefix] != did: best_prefix = (prefix, m.start()) if best_prefix: prefix, pos = best_prefix add(did, name_best[prefix], "prefix", nm[pos + 1:]) twins = [t for t in name_to_ids.get(nm, []) if t != did and org_part(t) != org_part(did)] if twins: parent = max(twins, key=lambda t: dls.get(t) or 0) if (dls.get(parent) or 0) > (dls.get(did) or 0): add(did, parent, "collision", "") listing_pages = 0 for child, links, loads in mentions_df.iter_rows(): raw = [(m, "mention_link") for m in (links or [])] + [(m, "mention_load") for m in (loads or [])] resolved = {} for m, sig in raw: target = m if m in id_set else lower_to_id.get(m.lower()) if target and target != child: resolved.setdefault(target, sig) if len(resolved) > MAX_MENTIONS: listing_pages += 1 continue for target, sig in resolved.items(): add(child, target, sig, "") print(f"candidates: {len(cands):,} ({listing_pages:,} listing pages skipped)", flush=True) if not cands and not retry_pairs: print("no new candidates and empty retry queue — finalizing without judge", flush=True) finalize(args, stamp, meta, declared, added=None, new_cutoff=max_created) return # load the canonical seen-set ONCE: reused for the fresh-candidate anti-join AND # for de-duping the retry queue against pairs verified since they last failed try: seen_df = load_scratch("verified_edges.parquet").select(["child", "parent"]) except Exception as e: # noqa: BLE001 print(f" (no existing verified_edges: {e})", flush=True) seen_df = None keep_pairs = [] if cands: cand = pl.DataFrame(list(cands.values())) novel = cand.filter(~pl.col("declared")) same_org = pl.col("child").str.split("/").list.first() == pl.col("parent").str.split("/").list.first() keep = novel.filter(same_org | (pl.col("parent_dl") >= args.min_parent_dl)) if seen_df is not None: keep = keep.join(seen_df, on=["child", "parent"], how="anti") keep = keep.with_columns( pl.col("parent").replace_strict(lks, default=0).alias("parent_likes") ).with_columns((pl.col("parent_dl") + 10 * pl.col("parent_likes")).alias("parent_score")) keep = keep.sort(["parent_score", "child_dl"], descending=[True, True]).head(args.max_edges) keep_pairs = [(r["child"], r["parent"], r["hint"]) for r in keep.iter_rows(named=True)] # fold in the carried retry queue (transient-failure pairs) REGARDLESS of cutoff, # so a child whose verify failed once isn't lost when the cutoff moves past it; # the anti-join vs verified still applies (anything since-verified is dropped) seen_set = set(map(tuple, seen_df.iter_rows())) if seen_df is not None else set() have = {(c_, p_) for c_, p_, _ in keep_pairs} retry_extra = [(c_, p_, h) for (c_, p_, h) in retry_pairs if (c_, p_) not in have and (c_, p_) not in seen_set] pairs = keep_pairs + retry_extra print(f"verifying {len(pairs):,} edges ({len(keep_pairs)} new + {len(retry_extra)} retried; " f"cap {args.max_edges})", flush=True) if not pairs: print("nothing to verify after dedup — finalizing", flush=True) finalize(args, stamp, meta, declared, added=None, new_cutoff=max_created) return con() # init duckdb in main thread hints = {(c_, p_): h for c_, p_, h in pairs} tag_ = "_smoke" if smoke else "" res = run_grouped( pairs, args.workers, label="[refresh] ", checkpoint=lambda r: push_df(results_to_df(r, hints), f"refresh/pending_partial_{stamp}{tag_}.parquet"), ) df = results_to_df(res, hints) # persist the retry queue: pairs that failed verification this run (fail_count+1, # dropped at RETRY_MAX_FAILS); succeeded pairs don't reappear, which clears them if not smoke: save_retry_queue(df, prior_fail) nt = df.filter(pl.col("ok") & ~pl.col("primary_type").is_in(list(TRIVIAL))) print(f"verified: {len(df)} edges, {len(nt)} non-trivial", flush=True) tag = "_smoke" if smoke else "" pending_name = f"refresh/pending_{stamp}{tag}.parquet" push_df(df, pending_name) if not smoke: push_df(meta.drop("createdAt"), "datasets_meta.parquet", "refresh: metadata") push_df(meta.select(["datasetId", "createdAt"]), "created_at.parquet", "refresh: created_at") push_df(declared, "declared_edges.parquet", "refresh: declared") if len(nt) == 0: print("nothing non-trivial to judge — finalizing", flush=True) finalize(args, stamp, meta, declared, added=None, new_cutoff=max_created) return if args.no_chain: print(f"--no-chain: gpu phase NOT submitted. pending={pending_name} new_cutoff={max_created}", flush=True) return print("submitting gpu phase...", flush=True) job = api.run_uv_job( script=args.script_url, script_args=["--phase", "gpu", "--pending", pending_name, "--stamp", stamp, "--new-cutoff", str(max_created)] + (["--smoke"] if smoke else []), image=GPU_IMAGE, python="/usr/bin/python3", env={"PYTHONPATH": GPU_DIST_PACKAGES, "VLLM_USE_DEEP_GEMM": "0"}, secrets={"HF_TOKEN": TOKEN}, flavor="a100-large", timeout="2h", ) print(f"gpu job: {job.id} ({job.url})", flush=True) # record the lease + block until the chained gpu job finishes, so a gpu failure # surfaces (red cpu Job + alert) instead of silently leaving the graph un-updated write_inflight(stamp, job.id) poll_gpu(job.id, stamp) # -------------------------------------------------------------------------- # GPU phase: judge -> merge -> rebuild db -> deploy -> state # -------------------------------------------------------------------------- def fetch_card(repo_id, max_chars=600): from huggingface_hub import DatasetCard try: content = DatasetCard.load(repo_id).content or "" content = re.sub(r"^---\s*\n.*?\n---\s*\n", "", content, flags=re.DOTALL).strip() return content[:max_chars] except Exception as e: # noqa: BLE001 return f"(no card available: {type(e).__name__})" def build_prompt(row, child_card, parent_card, hint): cc = json.loads(row.get("col_containment") or "{}") cols_str = ", ".join(f"{k}={v}" for k, v in list(cc.items())[:8]) or "(no shared text cols)" return f"""Two HuggingFace datasets — judge their derivation relationship. CHILD: {row["child"]} {child_card} PARENT: {row["parent"]} {parent_card} Evidence: - per-column value containment child→parent: {cols_str} - size_ratio (child rows / parent rows): {row.get("size_ratio")} - schema_jaccard: {row.get("schema_jaccard")} - language shift (ASCII-script heuristic): {row.get("language_shift")} - child-name suffix hint (after parent name): {hint or "(none)"} - heuristic verdict: {row.get("primary_type")} Pick ONE label from: {" | ".join(LABEL_ENUM)} Respond as compact JSON only, nothing else: {{"label": "", "reason": ""}}""" def parse_json_out(s): s = (s or "").strip() depth, start = 0, None for i, ch in enumerate(s): if ch == "{": if depth == 0: start = i depth += 1 elif ch == "}": depth -= 1 if depth == 0 and start is not None: try: d = json.loads(s[start:i + 1]) return d.get("label"), d.get("reason", "") except json.JSONDecodeError: start = None return None, s[:200] def judge(df, args): """LLM-judge ok+non-trivial rows; returns df with llm_label/llm_reason.""" rows = df.filter(pl.col("ok") & ~pl.col("primary_type").is_in(list(TRIVIAL))).to_dicts() print(f"judging {len(rows)} edges", flush=True) unique_ids = sorted({r["child"] for r in rows} | {r["parent"] for r in rows}) cards = {} with ThreadPoolExecutor(max_workers=32) as ex: for did, card in zip(unique_ids, ex.map(fetch_card, unique_ids)): cards[did] = card prompts = [build_prompt(r, cards[r["child"]], cards[r["parent"]], r.get("hint") or "") for r in rows] from vllm import LLM, SamplingParams from vllm.sampling_params import StructuredOutputsParams llm = LLM(model=MODEL, tensor_parallel_size=1, max_model_len=4096, max_num_seqs=64, dtype="bfloat16", trust_remote_code=True) tokenizer = llm.get_tokenizer() chats = [ tokenizer.apply_chat_template( [{"role": "system", "content": "You are a careful dataset-relationship annotator. Respond with ONLY compact JSON."}, {"role": "user", "content": p}], tokenize=False, add_generation_prompt=True, enable_thinking=False, ) for p in prompts ] sp = SamplingParams(temperature=0.1, top_p=0.95, top_k=20, max_tokens=200, structured_outputs=StructuredOutputsParams(json=LABEL_SCHEMA)) t0 = time.time() outputs = llm.generate(chats, sp) print(f"generation done in {time.time() - t0:.0f}s", flush=True) verdicts = [] for r, o in zip(rows, outputs): label, reason = parse_json_out(o.outputs[0].text) verdicts.append({"child": r["child"], "parent": r["parent"], "llm_label": label, "llm_reason": reason, "llm_model": MODEL, "child_card_excerpt": cards[r["child"]], "parent_card_excerpt": cards[r["parent"]]}) return pl.DataFrame(verdicts, infer_schema_length=None) def apply_verdicts(df, judged): """v3/apply_mentions logic: LLM label wins, drop LLM 'unrelated', 0.85 conf on override.""" j = df.join(judged.select(["child", "parent", "llm_label", "llm_reason"]), on=["child", "parent"], how="left") before = len(j) j = j.filter(pl.col("llm_label").ne_missing("unrelated") | ~pl.col("ok")) print(f"dropped {before - len(j)} rows judged 'unrelated'", flush=True) j = j.with_columns( pl.when(pl.col("llm_label").is_not_null()) .then(pl.col("llm_label")).otherwise(pl.col("primary_type")).alias("final_label"), pl.when(pl.col("llm_label").is_null()).then(pl.lit("heuristic")) .when(pl.col("llm_label") == pl.col("primary_type")).then(pl.lit("consensus")) .otherwise(pl.lit("llm_override")).alias("label_source"), ).with_columns( pl.col("primary_type").alias("primary_type_heuristic"), pl.col("final_label").alias("primary_type"), ).with_columns( pl.when(pl.col("label_source") == "llm_override").then(0.85) .otherwise(pl.col("confidence")).alias("confidence"), ) return j.drop(["final_label"]) def harvest_model_layer(min_dl=1): """Refresh the model layer: scan the model-cards dump for Hub-declared base_model:/dataset: tags and push ecosystem/model_edges.parquet + ecosystem/model_meta.parquet to scratch, replacing the canonical files. Inlined from jobs/ecosystem_harvest.py — refresh_job.py is fetched standalone from a URL at run time, so it can NOT import from src/. Runs in its own duckdb connection (closed on exit to free the ~642k-row models table). ~23s scan; call non-smoke only. model->model edges are kept when the child OR the parent has downloads >= min_dl (parent-cascade); ALL model->dataset edges are kept. These are declared facts (no inference); confidence is assigned at db build.""" rels = ", ".join(f"'{r}'" for r in MODEL_TYPED_RELS) hc = duckdb.connect() hc.execute("INSTALL httpfs; LOAD httpfs;") if TOKEN: hc.execute(f"CREATE SECRET hf (TYPE huggingface, TOKEN '{TOKEN}');") try: hc.execute(f""" CREATE OR REPLACE TABLE models AS SELECT modelId, downloads, likes, author, createdAt, tags FROM read_parquet('{MODEL_GLOB}') """) n_models = hc.execute("SELECT count(*) FROM models").fetchone()[0] # explode base_model tags -> typed model->model edges (4 real relations) hc.execute(f""" CREATE OR REPLACE TABLE m2m AS SELECT DISTINCT child, parent, rel FROM ( SELECT modelId AS child, CASE WHEN regexp_matches(t, '^base_model:[^:]+:') THEN regexp_extract(t, '^base_model:[^:]+:(.+)$', 1) ELSE regexp_extract(t, '^base_model:(.+)$', 1) END AS parent, CASE WHEN regexp_matches(t, '^base_model:[^:]+:') THEN regexp_extract(t, '^base_model:([^:]+):', 1) ELSE 'untyped' END AS rel FROM models, UNNEST(tags) AS u(t) WHERE t LIKE 'base_model:%' ) WHERE rel IN ({rels}) AND parent <> '' AND child <> '' AND child <> parent """) # parent-cascade threshold: keep an edge if child OR parent is a kept node hc.execute(f""" CREATE OR REPLACE TABLE m2m_kept AS SELECT e.child, e.parent, e.rel FROM m2m e LEFT JOIN models cc ON cc.modelId = e.child LEFT JOIN models pp ON pp.modelId = e.parent WHERE COALESCE(cc.downloads, 0) >= {min_dl} OR COALESCE(pp.downloads, 0) >= {min_dl} """) # dataset tags -> model->dataset edges (org/name form); keep ALL. a model # id can equal its dataset id (LeRobot repos) — valid cross-kind edge. hc.execute(""" CREATE OR REPLACE TABLE m2d AS SELECT DISTINCT modelId AS child, regexp_extract(t, '^dataset:(.+)$', 1) AS parent FROM models, UNNEST(tags) AS u(t) WHERE t LIKE 'dataset:%' AND regexp_extract(t, '^dataset:(.+)$', 1) LIKE '%/%' """) hc.execute(""" CREATE OR REPLACE TABLE model_edges AS SELECT child, 'model' AS child_kind, parent, 'model' AS parent_kind, rel AS primary_type, 'declared_tag' AS source FROM m2m_kept UNION ALL SELECT child, 'model' AS child_kind, parent, 'dataset' AS parent_kind, 'trained_on' AS primary_type, 'declared_tag' AS source FROM m2d """) # meta for every referenced MODEL node (dataset parents get their meta from # the dataset layer at db-build time, so they're excluded here) hc.execute(""" CREATE OR REPLACE TABLE model_meta AS WITH nodeset AS ( SELECT child AS id FROM m2m_kept UNION SELECT parent FROM m2m_kept UNION SELECT child FROM m2d ) SELECT m.modelId, m.downloads, m.likes, m.author, m.createdAt FROM models m JOIN nodeset n ON n.id = m.modelId """) n_m2m = hc.execute("SELECT count(*) FROM model_edges WHERE parent_kind='model'").fetchone()[0] n_m2d = hc.execute("SELECT count(*) FROM model_edges WHERE parent_kind='dataset'").fetchone()[0] n_meta = hc.execute("SELECT count(*) FROM model_meta").fetchone()[0] # trending_score isn't in the dump; fetch the top-N from the Hub API and # LEFT JOIN into model_meta (models outside the top-N get 0.0). A transient # API failure must NOT fail the monthly refresh — degrade to all-zero # trending (empty trending_enablement table), same as the harvest interlock. try: trending = [(m.id, float(getattr(m, "trending_score", 0) or 0)) for m in api.list_models(sort="trendingScore", limit=TREND_TOP_N, expand=["trendingScore"])] except Exception as e: # noqa: BLE001 — trending is optional enrichment print(f"WARNING: trending fetch failed ({e}); trending_score=0 this run", flush=True) trending = [] hc.execute("CREATE OR REPLACE TABLE trending(modelId VARCHAR, trending_score DOUBLE)") if trending: hc.executemany("INSERT INTO trending VALUES (?, ?)", trending) hc.execute("COPY (SELECT child, child_kind, parent, parent_kind, primary_type, source " "FROM model_edges) TO '/tmp/model_edges.parquet' (FORMAT parquet)") hc.execute("""COPY ( SELECT mm.modelId, mm.downloads, mm.likes, mm.author, mm.createdAt, COALESCE(t.trending_score, 0.0) AS trending_score FROM model_meta mm LEFT JOIN trending t ON t.modelId = mm.modelId ) TO '/tmp/model_meta.parquet' (FORMAT parquet)""") finally: hc.close() api.upload_file(path_or_fileobj="/tmp/model_edges.parquet", path_in_repo="ecosystem/model_edges.parquet", repo_id=SCRATCH, repo_type="dataset", commit_message="refresh: model_edges") api.upload_file(path_or_fileobj="/tmp/model_meta.parquet", path_in_repo="ecosystem/model_meta.parquet", repo_id=SCRATCH, repo_type="dataset", commit_message="refresh: model_meta") print(f"model layer: {n_models:,} models -> {n_m2m:,} model->model + {n_m2d:,} " f"model->dataset edges, {n_meta:,} model nodes (pushed to scratch)", flush=True) def compute_trending_enablement(model_edges, model_meta, top_n=TREND_TOP_ENABLE, max_hops=TREND_MAX_HOPS): """For the top-N trending models with >=1 parent edge, the upward ancestry walk (<=max_hops): model ancestors toward the root base + datasets reached via trained_on. trending_score comes from model_meta (Hub API top-N, carried by the harvest). Upward fan-in is small, so this is cheap. Returns list of (model_id, trending_score, ancestry_json); ancestry = JSON array of {id, kind, rel, hop} ordered by hop (shortest path wins).""" if model_meta is None or "trending_score" not in model_meta.columns: return [] up = defaultdict(list) # child -> [(parent, parent_kind, rel)] for child, _ck, parent, pk, pt, _src in model_edges.iter_rows(): up[child].append((parent, pk, pt)) def ancestry(mid): out, seen = [], {mid} q = deque([(mid, 0)]) while q: node, hop = q.popleft() if hop >= max_hops: continue for parent, pk, rel in up.get(node, []): if parent in seen: continue seen.add(parent) out.append({"id": parent, "kind": pk, "rel": rel, "hop": hop + 1}) if pk == "model": # datasets are terminal in the upward model walk q.append((parent, hop + 1)) return out trending = sorted( ((r["modelId"], float(r["trending_score"] or 0)) for r in model_meta.iter_rows(named=True) if (r["trending_score"] or 0) > 0), key=lambda x: x[1], reverse=True, ) rows = [] for mid, score in trending: if mid in up: rows.append((mid, score, json.dumps(ancestry(mid)))) if len(rows) >= top_n: break return rows def compute_dataset_impact(model_edges, model_meta, max_hops=IMPACT_MAX_HOPS): """Per-dataset downstream impact via the model-lineage closure. For each dataset with >=1 trained_on child: direct_models (trained directly on it), descendant_models (transitive closure = direct models + their model->model derivatives via finetune/adapter/quantized/merge, each once), descendant_downloads (summed downloads over that closure, each model once). Closure capped at max_hops model->model hops; a visited set dedups merge diamonds (no double-count) and guards any cycle. Whole-graph per-hop DuckDB joins, not per-dataset BFS (~0.2s). Returns list of (dataset_id, direct_models, descendant_models, descendant_downloads).""" if model_edges is None or model_meta is None or len(model_edges) == 0: return [] dc = duckdb.connect() dc.register("me", model_edges) dc.register("mm", model_meta) # child is DERIVED FROM parent; a model's derivatives are the children of m2m # edges where it is the parent. direct = the trained_on (model->dataset) edges. dc.execute("CREATE TABLE m2m AS SELECT DISTINCT child, parent FROM me WHERE parent_kind='model'") dc.execute("CREATE TABLE direct AS SELECT DISTINCT parent AS dataset_id, child AS model " "FROM me WHERE parent_kind='dataset'") dc.execute("CREATE TABLE visited AS SELECT dataset_id, model FROM direct") dc.execute("CREATE TABLE frontier AS SELECT dataset_id, model FROM direct") for _ in range(max_hops): dc.execute(""" CREATE OR REPLACE TABLE nxt AS (SELECT DISTINCT f.dataset_id, e.child AS model FROM frontier f JOIN m2m e ON e.parent = f.model) EXCEPT SELECT dataset_id, model FROM visited """) if dc.execute("SELECT count(*) FROM nxt").fetchone()[0] == 0: break dc.execute("INSERT INTO visited SELECT * FROM nxt") dc.execute("DROP TABLE frontier") dc.execute("ALTER TABLE nxt RENAME TO frontier") rows = dc.execute(""" SELECT d.dataset_id, d.direct_models, v.descendant_models, v.descendant_downloads FROM (SELECT dataset_id, count(*) AS direct_models FROM direct GROUP BY dataset_id) d JOIN (SELECT vv.dataset_id, count(*) AS descendant_models, COALESCE(sum(m.downloads), 0) AS descendant_downloads FROM visited vv LEFT JOIN mm m ON m.modelId = vv.model GROUP BY vv.dataset_id) v USING (dataset_id) """).fetchall() dc.close() return [(r[0], int(r[1]), int(r[2]), int(r[3])) for r in rows] def _total_edges(stats: dict) -> int: """Total edge count from a /stats payload, robust to the schema shape. Both the old dataset-only app and the current ecosystem app expose a top-level int `edges` = grand total (the per-kind fields are NODE breakdowns; declared_edges is a subset, not an addend). Future-proof: if `edges` is ever a per-kind dict, sum it. Returns 0 when absent so the caller skips the guard rather than crash.""" e = stats.get("edges") if isinstance(e, dict): return int(sum(e.values())) if isinstance(e, (int, float)): return int(e) return 0 def build_db(verified, declared, meta, created, out_path, model_edges=None, model_meta=None, min_conf=0.5, cutoff=None): import sqlite3 dl = dict(zip(meta["datasetId"].to_list(), meta["downloads"].to_list())) lk = dict(zip(meta["datasetId"].to_list(), meta["likes"].to_list())) au = dict(zip(meta["datasetId"].to_list(), meta["author"].to_list())) ds_ts = (dict(zip(meta["datasetId"].to_list(), meta["trending_score"].to_list())) if "trending_score" in meta.columns else {}) edges = [] ver = verified.filter( pl.col("ok") & ~pl.col("primary_type").is_in(list(TRIVIAL)) & (pl.col("confidence") >= min_conf) ) for r in ver.iter_rows(named=True): tags = r["tags"] tags = list(tags) if tags is not None else [] edges.append((r["child"], r["parent"], r["primary_type"], float(r["confidence"]), json.dumps(tags), r["size_ratio"], r["col_containment"], "inferred")) decl = declared.filter(pl.col("parent_kind") == "org_name").unique(subset=["child", "parent"]) inferred_pairs = {(c, p) for c, p, *_ in edges} for r in decl.iter_rows(named=True): if (r["child"], r["parent"]) in inferred_pairs: continue edges.append((r["child"], r["parent"], "declared", 1.0, "[]", None, None, "declared")) n_self = sum(1 for e in edges if e[0] == e[1]) edges = [e for e in edges if e[0] != e[1]] edge_map = {(e[0], e[1]): e for e in edges} n_flip = 0 for (c_, p_), e in list(edge_map.items()): if e[2] != "exact_copy" or (p_, c_) in edge_map: continue cc, pc = created.get(c_), created.get(p_) if cc and pc and pc > cc: del edge_map[(c_, p_)] edge_map[(p_, c_)] = (p_, c_, *e[2:]) n_flip += 1 drop = set() for (a, b) in list(edge_map): if (b, a) in edge_map and (a, b) not in drop and (b, a) not in drop: e_ab, e_ba = edge_map[(a, b)], edge_map[(b, a)] ca, cb = created.get(a), created.get(b) if ca and cb and ca != cb: keep_ab = cb < ca else: keep_ab = ((dl.get(b) or 0), e_ab[3]) >= ((dl.get(a) or 0), e_ba[3]) drop.add((b, a) if keep_ab else (a, b)) edges = [e for k, e in edge_map.items() if k not in drop] print(f"db: dropped {n_self} self-loops, flipped {n_flip}, dropped {len(drop)} mutual", flush=True) # --- assemble v2 edge list: dataset->dataset rows carry kind='dataset' both ends all_edges = [(c, "dataset", p, "dataset", pt, conf, tg, sr, ev, src) for (c, p, pt, conf, tg, sr, ev, src) in edges] n_ds_edges = len(all_edges) # --- model layer (declared-tag edges): confidence 1.0, source='declared_tag'; # acyclic by construction, so only a defensive self-loop drop is applied --- md_dl = md_lk = md_au = md_ts = md_created = {} n_self_model = 0 if model_edges is not None and model_meta is not None and len(model_edges): md_dl = dict(zip(model_meta["modelId"].to_list(), model_meta["downloads"].to_list())) md_lk = dict(zip(model_meta["modelId"].to_list(), model_meta["likes"].to_list())) md_au = dict(zip(model_meta["modelId"].to_list(), model_meta["author"].to_list())) md_ts = (dict(zip(model_meta["modelId"].to_list(), model_meta["trending_score"].to_list())) if "trending_score" in model_meta.columns else {}) md_created = (dict(zip(model_meta["modelId"].to_list(), model_meta["createdAt"].to_list())) if "createdAt" in model_meta.columns else {}) for r in model_edges.iter_rows(named=True): if r["child"] == r["parent"] and r["child_kind"] == r["parent_kind"]: n_self_model += 1 continue if r["primary_type"] not in MODEL_EDGE_TYPES: continue all_edges.append((r["child"], r["child_kind"], r["parent"], r["parent_kind"], r["primary_type"], 1.0, "[]", None, None, "declared_tag")) print(f"db: model layer +{len(all_edges) - n_ds_edges:,} edges " f"({n_self_model} self-loops dropped)", flush=True) else: print("db: WARNING — no model edges provided; building DATASET-ONLY graph " "on the v2 schema (model layer MISSING until next cpu-phase harvest)", flush=True) # --- nodes: (id, kind) from every edge endpoint; meta looked up per kind --- node_keys = set() for c_, ck, p_, pk, *_ in all_edges: node_keys.add((c_, ck)) node_keys.add((p_, pk)) def _author(nid, kind): if kind == "dataset": return au.get(nid, "") # model author: real value, else namespace fallback for parents absent from meta return md_au.get(nid) or (nid.split("/", 1)[0] if "/" in nid else "") node_rows = [] for nid, kind in node_keys: if kind == "dataset": d_, l_, ts_, ca_ = (int(dl.get(nid) or 0), int(lk.get(nid) or 0), ds_ts.get(nid), created.get(nid)) else: d_, l_, ts_, ca_ = (int(md_dl.get(nid) or 0), int(md_lk.get(nid) or 0), md_ts.get(nid), md_created.get(nid)) node_rows.append((nid, d_, l_, _author(nid, kind), kind, float(ts_) if ts_ is not None else None, _iso(ca_) if ca_ else None)) if os.path.exists(out_path): os.unlink(out_path) db = sqlite3.connect(out_path) # composite (dataset_id, kind) PK: Hub model ids can collide with dataset ids, # so kind is part of node identity (the `dataset_id` column name is kept for # schema-compat with the dataset-only app even though it now holds model ids too) db.executescript(""" CREATE TABLE nodes ( dataset_id TEXT, downloads INTEGER, likes INTEGER, author TEXT, kind TEXT CHECK(kind IN ('dataset','model')), trending_score REAL, created_at TEXT, PRIMARY KEY (dataset_id, kind) ); CREATE TABLE edges ( child TEXT, child_kind TEXT, parent TEXT, parent_kind TEXT, primary_type TEXT, confidence REAL, tags TEXT, size_ratio REAL, evidence TEXT, source TEXT ); """) db.executemany("INSERT OR IGNORE INTO nodes VALUES (?,?,?,?,?,?,?)", node_rows) db.executemany("INSERT INTO edges VALUES (?,?,?,?,?,?,?,?,?,?)", all_edges) db.executescript(""" CREATE INDEX idx_edges_child ON edges(child, child_kind); CREATE INDEX idx_edges_parent ON edges(parent, parent_kind); CREATE INDEX idx_nodes_downloads ON nodes(downloads); """) # dataset downstream-impact precompute (model-lineage closure). Always present # (empty in the dataset-only fallback); the UI reads it by dataset_id. impact = compute_dataset_impact(model_edges, model_meta) db.executescript(""" CREATE TABLE dataset_impact ( dataset_id TEXT PRIMARY KEY, direct_models INTEGER, descendant_models INTEGER, descendant_downloads INTEGER ); """) db.executemany("INSERT OR IGNORE INTO dataset_impact VALUES (?,?,?,?)", impact) db.execute("CREATE INDEX idx_impact_downloads ON dataset_impact(descendant_downloads)") # trending enablement: top-N trending models + upward ancestry enablement = compute_trending_enablement(model_edges, model_meta) db.executescript(""" CREATE TABLE trending_enablement ( model_id TEXT PRIMARY KEY, trending_score REAL, ancestry TEXT ); """) db.executemany("INSERT OR IGNORE INTO trending_enablement VALUES (?,?,?)", enablement) # timeline: precomputed per-quarter counts for the Overview time charts (avoids # client-side scans of ~320k nodes). period = ISO quarter-START date ('YYYY-MM-01', # sorts and parses as a real date). series='node' (category=node kind) or 'edge' # (category=primary_type). Edge creation time = the CHILD's created_at. Nodes/edges # with an unknown child created_at are omitted. Per-period counts (not cumulative). db.executescript(""" CREATE TABLE timeline ( period TEXT, series TEXT, category TEXT, count INTEGER, PRIMARY KEY (period, series, category) ); """) db.execute(""" INSERT INTO timeline SELECT substr(created_at,1,4)||'-'||printf('%02d',((CAST(substr(created_at,6,2) AS INTEGER)-1)/3)*3+1)||'-01', 'node', kind, count(*) FROM nodes WHERE created_at IS NOT NULL AND length(created_at)>=7 GROUP BY 1,3 """) db.execute(""" INSERT INTO timeline SELECT substr(n.created_at,1,4)||'-'||printf('%02d',((CAST(substr(n.created_at,6,2) AS INTEGER)-1)/3)*3+1)||'-01', 'edge', e.primary_type, count(*) FROM edges e JOIN nodes n ON n.dataset_id=e.child AND n.kind=e.child_kind WHERE n.created_at IS NOT NULL AND length(n.created_at)>=7 GROUP BY 1,3 """) # one-row build provenance so the Space can surface "data as of " + cutoff db.execute("CREATE TABLE meta (key TEXT PRIMARY KEY, value TEXT)") db.executemany("INSERT OR REPLACE INTO meta VALUES (?,?)", [ ("built_at", time.strftime("%Y-%m-%dT%H:%M:%S", time.gmtime())), ("cutoff", _iso(cutoff) if cutoff is not None else ""), ]) db.commit() db.close() print(f"db: {len(node_keys):,} nodes / {len(all_edges):,} edges / " f"{len(impact):,} dataset_impact / {len(enablement):,} trending -> {out_path}", flush=True) return len(node_keys), len(all_edges) def finalize(args, stamp, meta, declared, added, new_cutoff): """Merge IN MEMORY -> rebuild lineage.db -> no-shrink guard -> ONLY THEN overwrite canonical verified_edges + deploy + update state. Nothing canonical is written until the rebuilt graph passes the guard, so a bad/partial input aborts the run instead of poisoning the source of truth (the 2026-07-03 failure mode).""" smoke = args.smoke verified = load_scratch("verified_edges.parquet") merged = None n_added = 0 if added is not None and len(added): # schema guard: every CORE column must be present on BOTH sides, else the old # select(common) would have silently amputated a column from canonical missing_v = [c for c in CORE_COLS if c not in verified.columns] missing_a = [c for c in CORE_COLS if c not in added.columns] if missing_v or missing_a: alert("verified_edges schema drift", f"stamp={stamp}: CORE columns missing (verified={missing_v}, " f"added={missing_a}). Refusing to merge — would narrow/corrupt the " "canonical schema.") raise SystemExit("verified_edges schema drift — aborting") # diagonal_relaxed keeps EVERY column from both sides (audit cols persist as # nulls) instead of dropping to the intersection merged = pl.concat([verified, added], how="diagonal_relaxed").sort( "confidence", descending=True, nulls_last=True ).unique(subset=["child", "parent"], keep="first") n_added = len(merged) - len(verified) print(f"merged in memory: {len(verified):,} + {len(added):,} new -> " f"{len(merged):,} (net +{n_added})", flush=True) if "createdAt" in meta.columns: created_df = meta.select(["datasetId", "createdAt"]) else: try: created_df = load_scratch("created_at.parquet") except Exception as e: # noqa: BLE001 — created dates are an optional orientation signal print(f"created_at.parquet unavailable ({e}); proceeding without", flush=True) created_df = pl.DataFrame(schema={"datasetId": pl.Utf8, "createdAt": pl.Utf8}) created = {k: v for k, v in zip(created_df["datasetId"].to_list(), created_df["createdAt"].to_list()) if v} # model layer: the cpu phase refreshed these; if absent, build DATASET-ONLY on # the v2 schema and warn loudly (the no-shrink guard below then refuses to # deploy a graph that dropped the whole model layer). try: model_edges = load_scratch("ecosystem/model_edges.parquet") model_meta = load_scratch("ecosystem/model_meta.parquet") print(f"model layer loaded: {len(model_edges):,} edges / {len(model_meta):,} nodes", flush=True) except Exception as e: # noqa: BLE001 — missing model layer must not abort the dataset refresh print(f"WARNING: ecosystem model parquets unavailable ({e}); building " "DATASET-ONLY graph on the v2 schema — model layer will reappear once " "the next cpu phase repopulates ecosystem/model_*.parquet", flush=True) model_edges = model_meta = None build_source = merged if merged is not None else verified n_nodes, n_edges = build_db(build_source, declared, meta, created, "/tmp/lineage.db", model_edges=model_edges, model_meta=model_meta, cutoff=new_cutoff) if smoke: print("SMOKE: not deploying db / not updating state", flush=True) return # no-shrink guard runs BEFORE any canonical write and fails CLOSED (abort, don't # deploy) if it can't run — so an abort here leaves the source of truth untouched _shrink_guard(n_edges, stamp) # guard passed — NOW it's safe to overwrite canonical + deploy if merged is not None: push_df(verified, f"refresh/verified_backup_{stamp}.parquet", f"refresh {stamp}: pre-merge backup") push_df(merged, "verified_edges.parquet", f"refresh {stamp}: +{n_added} edges") push_df(added, f"refresh/added_{stamp}.parquet") api.upload_file(path_or_fileobj="/tmp/lineage.db", path_in_repo="lineage.db", repo_id=SPACE, repo_type="space", commit_message=f"refresh {stamp}: {n_nodes:,} nodes / {n_edges:,} edges") print(f"deployed lineage.db to {SPACE}", flush=True) state = load_state() state["cutoff"] = _iso(new_cutoff) runs = state.get("runs", []) run_rec = {"stamp": stamp, "new_cutoff": _iso(new_cutoff), "edges_added": n_added, "db_nodes": n_nodes, "db_edges": n_edges} shadow = getattr(args, "shadow_summary", None) if shadow: run_rec["shadow"] = shadow # distilled-classifier agreement audit (no effect on graph) runs.append(run_rec) state["runs"] = runs[-KEEP_STAMPS:] # bound the history save_state(state) print(f"state updated: cutoff={_iso(new_cutoff)}", flush=True) delete_inflight(stamp) # clear the lease (gpu phase reached a clean finish) prune_scratch() # keep only the newest KEEP_STAMPS refresh/ artifacts def shadow_predict(pending, judged): """SHADOW MODE: score the judged edges with the distilled CPU classifier and attach `distilled_label` + `distilled_conf` to `judged`. Pure audit — it never touches verdicts or the canonical graph. The caller wraps this in try/except so a shadow failure can never break a refresh. Feature evidence lives in the pending frame under refresh_job's names; the classifier's features.py expects the training-corpus names, so we join the two and rename. Card excerpts come from `judged` (fetched during judging). Returns (judged_with_cols, summary_dict).""" import importlib.util import lightgbm as lgb if judged.is_empty(): return (judged.with_columns(pl.lit(None, dtype=pl.Utf8).alias("distilled_label"), pl.lit(None, dtype=pl.Float64).alias("distilled_conf")), {"n": 0}) model_path = hf_hub_download(DISTILL_REPO, "model.txt", repo_type="model", token=TOKEN) feats_path = hf_hub_download(DISTILL_REPO, "features.py", repo_type="model", token=TOKEN) spec = importlib.util.spec_from_file_location("distill_features", feats_path) feat_mod = importlib.util.module_from_spec(spec) spec.loader.exec_module(feat_mod) evidence = pending.select([ "child", "parent", "primary_type", "confidence", "schema_jaccard", "size_ratio", "col_containment", "inherited_cols", "changed_cols", "language_shift", "child_rows", "parent_rows", "hint", ]) m = judged.join(evidence, on=["child", "parent"], how="left").rename({ "primary_type": "heuristic_v2_label", "confidence": "heuristic_v2_confidence", "hint": "name_suffix_hint", }) feat_cols = feat_mod.feature_columns() X = feat_mod.numeric_frame(m).select(feat_cols).to_numpy() booster = lgb.Booster(model_file=model_path) if booster.num_feature() != len(feat_cols): raise ValueError(f"model expects {booster.num_feature()} features, built {len(feat_cols)}") proba = booster.predict(X) out = judged.with_columns( pl.Series("distilled_label", [feat_mod.LABELS[i] for i in proba.argmax(axis=1)]), pl.Series("distilled_conf", proba.max(axis=1)), ) ev = out.filter(pl.col("llm_label").is_not_null()) n = len(ev) agree = float((ev["distilled_label"] == ev["llm_label"]).mean()) if n else 0.0 unrel = ev.filter(pl.col("llm_label") == "unrelated") unrel_rec = float((unrel["distilled_label"] == "unrelated").mean()) if len(unrel) else None summary = {"n": n, "agreement": round(agree, 4), "unrelated_n": len(unrel), "unrelated_recall": round(unrel_rec, 4) if unrel_rec is not None else None} print(f"[shadow] distilled classifier vs LLM: n={n} agreement={agree:.4f} " f"unrelated_recall={summary['unrelated_recall']} (n_unrelated={len(unrel)})", flush=True) return out, summary def phase_gpu(args): stamp = args.stamp or "manual" print(f"== gpu phase: pending={args.pending} stamp={stamp} smoke={args.smoke} ==", flush=True) # preflight: fail fast + loud if the pinned image can't import vllm (e.g. a re-pin # drifted from the /usr/bin/python3 + python3.12 dist-packages assumption) instead # of dying deep inside judge(); the cpu poller also surfaces this as a gpu failure try: import vllm # noqa: F401 print(f"vllm preflight OK ({vllm.__version__})", flush=True) except Exception as e: # noqa: BLE001 alert("gpu image broken (vllm import failed)", f"stamp={stamp}: `import vllm` failed in the pinned image ({GPU_IMAGE}): " f"{type(e).__name__}: {e}. Re-pin GPU_IMAGE / fix PYTHONPATH. The inflight " "marker is left so the cpu poller / watchdog reconciles.") raise SystemExit(f"vllm import failed: {e}") df = load_scratch(args.pending) judged = judge(df, args) # SHADOW MODE: attach distilled-classifier predictions for audit. Never affects # verdicts; a shadow failure must not break the refresh, so swallow everything. try: judged, args.shadow_summary = shadow_predict(df, judged) except Exception as e: # noqa: BLE001 args.shadow_summary = {"error": f"{type(e).__name__}: {e}"} print(f"[shadow] FAILED (ignored, no effect on verdicts): {type(e).__name__}: {e}", flush=True) if not args.smoke: push_df(judged, f"refresh/judged_{stamp}.parquet") applied = apply_verdicts(df, judged) # only ok + non-trivial rows enter the canonical edge set; everything else # stays in the pending/judged audit files added = applied.filter(pl.col("ok")) meta = load_scratch("datasets_meta.parquet") declared = load_scratch("declared_edges.parquet") finalize(args, stamp, meta, declared, added, new_cutoff=args.new_cutoff) def main(): ap = argparse.ArgumentParser() ap.add_argument("--phase", choices=["cpu", "gpu"], default="cpu") ap.add_argument("--smoke", action="store_true", help="limited pull, tagged outputs, no deploy/state") ap.add_argument("--no-chain", action="store_true", help="cpu phase: don't submit the gpu job") ap.add_argument("--cutoff", default="", help="override state cutoff (createdAt ISO)") ap.add_argument("--stamp", default="", help="run stamp (gpu phase inherits cpu's)") ap.add_argument("--pending", default="", help="gpu phase: pending parquet path in scratch") ap.add_argument("--new-cutoff", default="", help="gpu phase: cutoff to persist on success") ap.add_argument("--max-edges", type=int, default=10000) ap.add_argument("--min-parent-dl", type=int, default=50) ap.add_argument("--workers", type=int, default=32) ap.add_argument("--script-url", default=SCRIPT_URL) args = ap.parse_args() if args.phase == "gpu": if not args.pending: raise SystemExit("--phase gpu requires --pending") phase_gpu(args) else: phase_cpu(args) if __name__ == "__main__": main()