| """Oracle-window ceiling for Direction C (Review #22). CPU only; TRAIN fit, VALID read once. |
| |
| Gives the frozen character classifier only the lines around the diff-changed region of each |
| function (an ORACLE locator that uses the vulnerable/patched diff, so it cannot be deployed). |
| It bounds how much a perfect evidence locator could help long functions. Controls separate |
| that effect from window-length shortcuts: a random-location window with the same line count, |
| and a length-only score. TEST is never read. |
| """ |
| from __future__ import annotations |
|
|
| import difflib |
| import hashlib |
| import json |
| from pathlib import Path |
|
|
| import numpy as np |
| import pandas as pd |
| from sklearn.feature_extraction.text import TfidfVectorizer |
| from sklearn.linear_model import SGDClassifier |
| from sklearn.metrics import roc_auc_score |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CORPUS = ROOT / "data/processed/corpus.parquet" |
| SPLITS = {name: ROOT / f"data/splits/chronological/{name}.json" for name in ("train", "valid")} |
| OUTPUT = ROOT / "results/phase6/review22/oracle_slice_ceiling.json" |
| MARGINS = (0, 3, 10) |
| BANDS = {"<=30": (0, 30), "31-100": (31, 100), ">100": (101, 10**9)} |
| SEEDS = (0, 1, 2) |
|
|
|
|
| def windows(vuln: str, patch: str, margin: int) -> tuple[str, str]: |
| v, p = vuln.splitlines(), patch.splitlines() |
| ops = [op for op in difflib.SequenceMatcher(None, v, p, autojunk=False).get_opcodes() if op[0] != "equal"] |
| if not ops: |
| return vuln, patch |
|
|
| def pick(lines, spans): |
| keep = sorted({i for lo, hi in spans for i in range(max(0, lo - margin), min(len(lines), hi + margin))}) |
| return "\n".join(lines[i] for i in keep) or "\n".join(lines) |
| return (pick(v, [(o[1], max(o[2], o[1] + 1)) for o in ops]), |
| pick(p, [(o[3], max(o[4], o[3] + 1)) for o in ops])) |
|
|
|
|
| def random_windows(vuln: str, patch: str, margin: int, rng: np.random.Generator) -> tuple[str, str]: |
| """Contiguous random-location windows with the same line counts as the oracle windows.""" |
| ov, op = windows(vuln, patch, margin) |
| out = [] |
| for text, target in ((vuln, len(ov.splitlines())), (patch, len(op.splitlines()))): |
| lines = text.splitlines() |
| n = min(max(target, 1), len(lines)) |
| start = int(rng.integers(0, max(1, len(lines) - n + 1))) |
| out.append("\n".join(lines[start:start + n])) |
| return out[0], out[1] |
|
|
|
|
| def build(frame: pd.DataFrame, ids: set[str], mode: str, margin: int, rng, hit_rate: float = 1.0): |
| texts, labels, pair_idx, full_lines = [], [], [], [] |
| for k, (pid, group) in enumerate(frame[frame["pair_id"].astype(str).isin(ids)].groupby("pair_id", sort=True)): |
| vrow = group[group["target"].astype(int) == 1].iloc[0] |
| vuln, patch = str(vrow["func_vuln"]), str(vrow["func_patch"]) |
| if mode == "full": |
| tv, tp = vuln, patch |
| elif mode == "oracle" or (mode == "mixture" and rng.random() < hit_rate): |
| tv, tp = windows(vuln, patch, margin) |
| else: |
| tv, tp = random_windows(vuln, patch, margin, rng) |
| for text, label, full in ((tp, 0, patch), (tv, 1, vuln)): |
| texts.append(text) |
| labels.append(label) |
| pair_idx.append(k) |
| full_lines.append(len(full.splitlines())) |
| return texts, np.asarray(labels), np.asarray(pair_idx), np.asarray(full_lines) |
|
|
|
|
| def fit_eval(train, valid, seeds=SEEDS): |
| vec = TfidfVectorizer(analyzer="char", ngram_range=(3, 5), min_df=2, max_features=300_000, |
| sublinear_tf=True, dtype=np.float32) |
| x_train, x_valid = vec.fit_transform(train[0]), vec.transform(valid[0]) |
| scores = [] |
| for seed in seeds: |
| model = SGDClassifier(loss="log_loss", alpha=1e-5, max_iter=200, tol=1e-3, random_state=seed, average=True) |
| scores.append(model.fit(x_train, train[1]).decision_function(x_valid)) |
| return np.mean(scores, axis=0) |
|
|
|
|
| def summarize(scores, valid, rng, reps=300): |
| labels, pair_idx, lines = valid[1], valid[2], valid[3] |
|
|
| def auc_ci(mask): |
| auc = float(roc_auc_score(labels[mask], scores[mask])) |
| pairs = np.unique(pair_idx[mask]) |
| by_pair = {p: np.where((pair_idx == p) & mask)[0] for p in pairs} |
| draws = [] |
| for _ in range(reps): |
| idx = np.concatenate([by_pair[p] for p in rng.choice(pairs, size=len(pairs), replace=True)]) |
| if len(set(labels[idx])) == 2: |
| draws.append(roc_auc_score(labels[idx], scores[idx])) |
| return {"row_roc_auc": auc, "ci95": [float(x) for x in np.quantile(draws, [0.025, 0.975])], |
| "rows": int(mask.sum()), |
| "balanced_accuracy_at_zero": float(0.5 * ((scores[mask & (labels == 1)] >= 0).mean() |
| + (scores[mask & (labels == 0)] < 0).mean()))} |
| out = {"all": auc_ci(np.ones(len(labels), dtype=bool))} |
| for name, (lo, hi) in BANDS.items(): |
| out[f"original_function_lines_{name}"] = auc_ci((lines >= lo) & (lines <= hi)) |
| return out |
|
|
|
|
| def main() -> dict: |
| frame = pd.read_parquet(CORPUS) |
| ids = {name: set(map(str, json.loads(path.read_text()))) for name, path in SPLITS.items()} |
| rng = np.random.default_rng(42) |
| report = {"analysis": "oracle diff-window ceiling for evidence-guided long-function detection", |
| "caveat": ("ORACLE: windows are located with the vulnerable/patched diff, which is unavailable at " |
| "inference. Window length differs by class (fixes usually add lines), so the length-only " |
| "control must be read alongside every AUC. Exploratory; VALID read once; no TEST."), |
| "results": {}} |
| full = (build(frame, ids["train"], "full", 0, rng), build(frame, ids["valid"], "full", 0, rng)) |
| report["results"]["full_function"] = summarize(fit_eval(*full), full[1], rng) |
| for margin in MARGINS: |
| oracle = (build(frame, ids["train"], "oracle", margin, rng), build(frame, ids["valid"], "oracle", margin, rng)) |
| scores = fit_eval(*oracle) |
| res = summarize(scores, oracle[1], rng) |
| length_scores = np.asarray([len(t) for t in oracle[1][0]], dtype=float) |
| res["window_length_only_row_roc_auc"] = float(roc_auc_score(oracle[1][1], length_scores)) |
| res["mean_window_lines_SAFE_vs_VULNERABLE"] = [ |
| float(np.mean([len(t.splitlines()) for t, y in zip(oracle[1][0], oracle[1][1]) if y == 0])), |
| float(np.mean([len(t.splitlines()) for t, y in zip(oracle[1][0], oracle[1][1]) if y == 1]))] |
| report["results"][f"oracle_window_margin_{margin}"] = res |
| rand = (build(frame, ids["train"], "random", margin, rng), build(frame, ids["valid"], "random", margin, rng)) |
| rres = summarize(fit_eval(*rand), rand[1], rng) |
| rres["window_length_only_row_roc_auc"] = float(roc_auc_score( |
| rand[1][1], np.asarray([len(t) for t in rand[1][0]], dtype=float))) |
| report["results"][f"random_window_same_line_count_margin_{margin}"] = rres |
| report["locator_recall_mixture"] = { |
| "description": ("per-function: oracle window with probability h, otherwise a random same-size window; " |
| "train and VALID built with the same h, so the classifier learns under locator noise"), |
| "results": {}} |
| for margin in (0, 3): |
| for h in (0.25, 0.5, 0.75): |
| mix = (build(frame, ids["train"], "mixture", margin, rng, h), build(frame, ids["valid"], "mixture", margin, rng, h)) |
| report["locator_recall_mixture"]["results"][f"margin_{margin}_h_{h}"] = summarize(fit_eval(*mix), mix[1], rng) |
| report["provenance"] = {"corpus_sha256": hashlib.sha256(CORPUS.read_bytes()).hexdigest(), |
| **{f"{k}_split_sha256": hashlib.sha256(p.read_bytes()).hexdigest() for k, p in SPLITS.items()}, |
| "test_read": False} |
| OUTPUT.write_text(json.dumps(report, indent=2) + "\n") |
| return report |
|
|
|
|
| if __name__ == "__main__": |
| result = main() |
| for name, res in result["results"].items(): |
| bands = {k.replace("original_function_lines_", ""): round(v["row_roc_auc"], 3) for k, v in res.items() |
| if isinstance(v, dict) and "row_roc_auc" in v} |
| print(f"{name:48s} {bands} len_only={round(res.get('window_length_only_row_roc_auc', float('nan')), 3)}") |
| for name, res in result["locator_recall_mixture"]["results"].items(): |
| bands = {k.replace("original_function_lines_", ""): round(v["row_roc_auc"], 3) for k, v in res.items() |
| if isinstance(v, dict) and "row_roc_auc" in v} |
| print(f"mixture {name:40s} {bands}") |
|
|
