"""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}")