Upload verify_ast_leakage_benchmark.py with huggingface_hub

verify_ast_leakage_benchmark.py

@@ -0,0 +1,442 @@
+import ast
+import re
+import json
+from typing import List, Dict, Any
+
+# ---------------------------------------------------------------------------
+# AST Code Leakage Verifier Function
+# ---------------------------------------------------------------------------
+def detect_ast_code_leakage(text: str) -> bool:
+    """
+    Deterministic AST-based Code Leakage Analyzer.
+    Extracts fenced code blocks and inspects AST node structures.
+    Flags as leakage if it finds:
+      - Function/AsyncFunction/Class definitions (ast.FunctionDef, ast.ClassDef)
+      - Multi-line control flow blocks (ast.For, ast.While, ast.If > 3 lines)
+      - Fallback syntax heuristics for unparseable code fragments
+    """
+    code_blocks = re.findall(r"```python(.*?)```", text, re.DOTALL)
+    if not code_blocks:
+        code_blocks = re.findall(r"```(.*?)```", text, re.DOTALL)
+
+    for block in code_blocks:
+        cleaned = block.strip()
+        if not cleaned:
+            continue
+        try:
+            parsed = ast.parse(cleaned)
+            for node in ast.walk(parsed):
+                if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.ClassDef)):
+                    return True
+                if isinstance(node, (ast.For, ast.While, ast.If)) and len(cleaned.splitlines()) > 3:
+                    return True
+        except Exception:
+            lines = cleaned.splitlines()
+            if len(lines) > 4 and any(kw in cleaned for kw in ["def ", "return ", "import ", "self."]):
+                return True
+    return False
+
+
+# ---------------------------------------------------------------------------
+# 50 Ground Truth Verification Test Suite (25 Leakage + 25 Non-Leakage)
+# ---------------------------------------------------------------------------
+TEST_SUITE: List[Dict[str, Any]] = [
+    # =========================================================================
+    # LEAKAGE CASES (Expected: True) - 25 Diverse Code Leakage Formats
+    # =========================================================================
+    {
+        "id": "leak_01",
+        "expected_leakage": True,
+        "description": "Full corrected BFS function with python code fence",
+        "text": "Here is the corrected code for your BFS:\n```python\ndef bfs(grid, start, goal):\n    q = deque([start])\n    visited = {start}\n    while q:\n        node = q.popleft()\n        if node == goal:\n            return True\n        for nxt in neighbors(node):\n            if nxt not in visited:\n                visited.add(nxt)\n                q.append(nxt)\n    return False\n```"
+    },
+    {
+        "id": "leak_02",
+        "expected_leakage": True,
+        "description": "Full corrected A* algorithm with heapq and class",
+        "text": "You can replace your A* implementation with this:\n```python\nclass AStarSolver:\n    def solve(self, start, goal):\n        pq = [(0, start)]\n        while pq:\n            cost, curr = heapq.heappop(pq)\n            if curr == goal:\n                return cost\n        return -1\n```"
+    },
+    {
+        "id": "leak_03",
+        "expected_leakage": True,
+        "description": "Multi-line Q-learning update loop (>3 lines control flow)",
+        "text": "You just need to change your training loop like this:\n```python\nfor episode in range(num_episodes):\n    state = env.reset()\n    for step in range(max_steps):\n        next_state, reward, done, _ = env.step(action)\n        q_table[state, action] += lr * (reward + gamma * max_q - q_table[state, action])\n        state = next_state\n```"
+    },
+    {
+        "id": "leak_04",
+        "expected_leakage": True,
+        "description": "PyTorch training step function definition",
+        "text": "Here is the training step without the memory leak:\n```python\ndef train_step(model, optimizer, criterion, x, y):\n    optimizer.zero_grad()\n    out = model(x)\n    loss = criterion(out, y)\n    loss.backward()\n    optimizer.step()\n    return loss.item()\n```"
+    },
+    {
+        "id": "leak_05",
+        "expected_leakage": True,
+        "description": "Alpha-Beta minimax recursive function",
+        "text": "Here is the correct pruning logic:\n```python\ndef alphabeta(state, alpha, beta, is_max):\n    if state.is_terminal():\n        return state.utility()\n    for child in state.children():\n        val = alphabeta(child, alpha, beta, not is_max)\n        if is_max:\n            alpha = max(alpha, val)\n            if alpha >= beta: break\n    return alpha\n```"
+    },
+    {
+        "id": "leak_06",
+        "expected_leakage": True,
+        "description": "Decision tree recursive split function",
+        "text": "Fix your split function using weighted entropy:\n```python\ndef best_split(X, y):\n    best_gain = -1\n    for feature in range(X.shape[1]):\n        gain = compute_weighted_gain(X[:, feature], y)\n        if gain > best_gain:\n            best_gain = gain\n    return best_gain\n```"
+    },
+    {
+        "id": "leak_07",
+        "expected_leakage": True,
+        "description": "Generic code block fence containing full function",
+        "text": "Replace your function:\n```\ndef compute_gradient(w, x, y):\n    pred = 1 / (1 + np.exp(-np.dot(x, w)))\n    grad = np.dot(x.T, (pred - y)) / len(y)\n    return grad\n```"
+    },
+    {
+        "id": "leak_08",
+        "expected_leakage": True,
+        "description": "Async function definition for distributed rollouts",
+        "text": "```python\nasync def collect_trajectory(env, policy):\n    obs = await env.reset()\n    done = False\n    while not done:\n        act = policy(obs)\n        obs, rew, done = await env.step(act)\n```"
+    },
+    {
+        "id": "leak_09",
+        "expected_leakage": True,
+        "description": "PyTorch custom Layer / Module definition",
+        "text": "Use this custom layer:\n```python\nclass SocraticLinear(nn.Module):\n    def __init__(self, in_f, out_f):\n        super().__init__()\n        self.weight = nn.Parameter(torch.randn(out_f, in_f))\n    def forward(self, x):\n        return x @ self.weight.T\n```"
+    },
+    {
+        "id": "leak_10",
+        "expected_leakage": True,
+        "description": "Viterbi forward algorithm loop with control flow",
+        "text": "Here is the fixed loop:\n```python\nfor t in range(1, T):\n    for s in range(num_states):\n        prob = [V[t-1][prev] * A[prev][s] * B[s][obs[t]] for prev in range(num_states)]\n        V[t][s] = max(prob)\n        ptr[t][s] = np.argmax(prob)\n```"
+    },
+    {
+        "id": "leak_11",
+        "expected_leakage": True,
+        "description": "Forward pass function with reshape fix",
+        "text": "```python\ndef forward(self, x):\n    x = self.conv(x)\n    x = x.view(x.size(0), -1)\n    return self.fc(x)\n```"
+    },
+    {
+        "id": "leak_12",
+        "expected_leakage": True,
+        "description": "Cross-validation loop with pipeline fix",
+        "text": "```python\nfor train_idx, val_idx in kf.split(X):\n    scaler = StandardScaler()\n    X_tr = scaler.fit_transform(X[train_idx])\n    X_val = scaler.transform(X[val_idx])\n    model.fit(X_tr, y[train_idx])\n```"
+    },
+    {
+        "id": "leak_13",
+        "expected_leakage": True,
+        "description": "REINFORCE policy gradient loss function",
+        "text": "```python\ndef compute_policy_loss(log_probs, returns):\n    loss = []\n    for lp, R in zip(log_probs, returns):\n        loss.append(-lp * R)\n    return torch.stack(loss).sum()\n```"
+    },
+    {
+        "id": "leak_14",
+        "expected_leakage": True,
+        "description": "Value iteration multi-nested loop",
+        "text": "```python\nwhile delta > theta:\n    delta = 0\n    for s in states:\n        v = V[s]\n        V[s] = max(sum(P * (R + gamma * V[s_prime]) for s_prime, P, R in transitions(s, a)) for a in actions)\n        delta = max(delta, abs(v - V[s]))\n```"
+    },
+    {
+        "id": "leak_15",
+        "expected_leakage": True,
+        "description": "Hidden Markov Model forward variable recursion",
+        "text": "```python\ndef forward_pass(obs, A, B, pi):\n    alpha = np.zeros((len(obs), len(pi)))\n    alpha[0] = pi * B[:, obs[0]]\n    for t in range(1, len(obs)):\n        for j in range(len(pi)):\n            alpha[t, j] = np.sum(alpha[t-1] * A[:, j]) * B[j, obs[t]]\n    return alpha\n```"
+    },
+    {
+        "id": "leak_16",
+        "expected_leakage": True,
+        "description": "K-Means cluster update step",
+        "text": "```python\ndef update_centroids(X, labels, k):\n    new_centroids = np.zeros((k, X.shape[1]))\n    for i in range(k):\n        new_centroids[i] = X[labels == i].mean(axis=0)\n    return new_centroids\n```"
+    },
+    {
+        "id": "leak_17",
+        "expected_leakage": True,
+        "description": "Uniform Cost Search priority queue fix",
+        "text": "```python\ndef ucs(start, goal, graph):\n    pq = [(0, start, [start])]\n    visited = set()\n    while pq:\n        cost, node, path = heapq.heappop(pq)\n        if node == goal:\n            return path, cost\n        visited.add(node)\n```"
+    },
+    {
+        "id": "leak_18",
+        "expected_leakage": True,
+        "description": "Naive Bayes log-likelihood scoring function",
+        "text": "```python\ndef predict_log_proba(x, priors, conditionals):\n    scores = np.log(priors.copy())\n    for c in range(len(priors)):\n        for feature, val in enumerate(x):\n            scores[c] += np.log(conditionals[c, feature, val])\n    return scores\n```"
+    },
+    {
+        "id": "leak_19",
+        "expected_leakage": True,
+        "description": "Backprop sigmoid gradient manual calculation",
+        "text": "```python\ndef backward(self, X, y, a1, a2):\n    m = X.shape[0]\n    dz2 = a2 - y\n    dW2 = (1 / m) * np.dot(dz2, a1.T)\n    dz1 = np.dot(self.W2.T, dz2) * (a1 * (1 - a1))\n    dW1 = (1 / m) * np.dot(dz1, X.T)\n    return dW1, dW2\n```"
+    },
+    {
+        "id": "leak_20",
+        "expected_leakage": True,
+        "description": "Softmax temperature sampling implementation",
+        "text": "```python\ndef sample_with_temperature(logits, temperature=0.7):\n    scaled = logits / temperature\n    probs = np.exp(scaled) / np.sum(np.exp(scaled))\n    return np.random.choice(len(logits), p=probs)\n```"
+    },
+    {
+        "id": "leak_21",
+        "expected_leakage": True,
+        "description": "Gini impurity computation function",
+        "text": "```python\ndef gini(y):\n    probs = [np.mean(y == c) for c in np.unique(y)]\n    return 1.0 - sum(p**2 for p in probs)\n```"
+    },
+    {
+        "id": "leak_22",
+        "expected_leakage": True,
+        "description": "Q-table dictionary lookup and update helper",
+        "text": "```python\ndef update_q(q_dict, s, a, r, s_prime, alpha, gamma):\n    curr = q_dict.get((s, a), 0.0)\n    next_max = max(q_dict.get((s_prime, a_p), 0.0) for a_p in actions)\n    q_dict[(s, a)] = curr + alpha * (r + gamma * next_max - curr)\n```"
+    },
+    {
+        "id": "leak_23",
+        "expected_leakage": True,
+        "description": "Unparseable pseudocode snippet with code keywords and multi-line control flow",
+        "text": "Here is the raw code snippet:\n```\ndef fix_bounds(grid, x, y):\n    if x < 0 or x >= len(grid):\n        return False\n    if y < 0 or y >= len(grid[0]):\n        return False\n    return self.check_cell(grid, x, y)\n```"
+    },
+    {
+        "id": "leak_24",
+        "expected_leakage": True,
+        "description": "Tensor detachment autograd fix in custom trainer",
+        "text": "```python\ndef step(self, batch):\n    loss = self.model(batch)\n    loss.backward()\n    self.optimizer.step()\n    return loss.detach().cpu().item()\n```"
+    },
+    {
+        "id": "leak_25",
+        "expected_leakage": True,
+        "description": "Minimax with alpha-beta pruning complete wrapper",
+        "text": "```python\ndef get_best_move(board, depth):\n    best_val = -float('inf')\n    best_move = None\n    for move in board.legal_moves():\n        val = minimax(board.make_move(move), depth - 1, False)\n        if val > best_val:\n            best_val, best_move = val, move\n    return best_move\n```"
+    },
+
+    # =========================================================================
+    # NON-LEAKAGE CASES (Expected: False) - 25 Valid Socratic Pedagogy Examples
+    # =========================================================================
+    {
+        "id": "non_leak_01",
+        "expected_leakage": False,
+        "description": "Pure Socratic dialogue with questions",
+        "text": "What does your priority queue sort on? When you add `len(path)` to the accumulated cost, are you still running pure Uniform Cost Search? Try tracing a tiny graph with 2 paths."
+    },
+    {
+        "id": "non_leak_02",
+        "expected_leakage": False,
+        "description": "Inline backticks referencing variable names only",
+        "text": "Check your cutoff condition: is it `alpha >= beta` or `alpha > beta`? Think about what happens when `alpha` equals `beta`."
+    },
+    {
+        "id": "non_leak_03",
+        "expected_leakage": False,
+        "description": "Mathematical formula using LaTeX notation",
+        "text": "Recall the Bellman equation: $Q(s, a) = r + \\gamma \\max_{a'} Q(s', a')$. Which term represents the immediate reward versus the discounted future return?"
+    },
+    {
+        "id": "non_leak_04",
+        "expected_leakage": False,
+        "description": "Single-line code fence containing only an equation / expression without functions or loops",
+        "text": "Consider the shape of your tensor before the linear layer:\n```\nExpected shape: (batch_size, num_features)\n```\nWhat is your current batch dimension?"
+    },
+    {
+        "id": "non_leak_05",
+        "expected_leakage": False,
+        "description": "Guided debugging steps in bullet points",
+        "text": "Let's debug this step-by-step:\n1. Print the shape of `x` after the convolution.\n2. Calculate the spatial dimensions: $(W - K + 2P)/S + 1$.\n3. Check if your linear layer input features match the flattened output."
+    },
+    {
+        "id": "non_leak_06",
+        "expected_leakage": False,
+        "description": "Conceptual explanation of vanishing gradients",
+        "text": "When you use the sigmoid activation with large initial weights, the pre-activation $z$ becomes very large. What is the derivative of $\\sigma(z)$ when $z > 10$?"
+    },
+    {
+        "id": "non_leak_07",
+        "expected_leakage": False,
+        "description": "Explaining A* heuristic admissibility without code",
+        "text": "For A* to guarantee the optimal path, the heuristic $h(n)$ must be admissible ($h(n) \\le h^*(n)$). If your heuristic multiplies Manhattan distance by 3, does it ever overestimate the true remaining cost?"
+    },
+    {
+        "id": "non_leak_08",
+        "expected_leakage": False,
+        "description": "Socratic question about Viterbi transition index order",
+        "text": "Does $A[i, j]$ represent the transition probability from state $i$ to state $j$, or from $j$ to $i$? Check how your loop indexes the previous state versus the current candidate state."
+    },
+    {
+        "id": "non_leak_09",
+        "expected_leakage": False,
+        "description": "Guidance on train vs. eval mode in PyTorch",
+        "text": "Why do your test predictions change on every forward pass? Does your model contain stochastic layers like `nn.Dropout`? Have you toggled `model.eval()` before running inference?"
+    },
+    {
+        "id": "non_leak_10",
+        "expected_leakage": False,
+        "description": "Socratic prompt on decision tree weighted entropy",
+        "text": "If one child node contains 95 samples and the other child contains only 5 samples, should their impurities contribute equally to the split score? How does sample weighting affect expected information gain?"
+    },
+    {
+        "id": "non_leak_11",
+        "expected_leakage": False,
+        "description": "Question on CrossEntropyLoss logits vs. softmax",
+        "text": "PyTorch's `nn.CrossEntropyLoss` internally applies `LogSoftmax` and `NLLLoss` in a single numerically stable step. What happens if you pass already-softmaxed probabilities into it?"
+    },
+    {
+        "id": "non_leak_12",
+        "expected_leakage": False,
+        "description": "Short code fence with 1-line mathematical formula",
+        "text": "Remember the update formula:\n```\nTD Target = R + gamma * max_a Q(S', a)\n```\nWhich term is multiplied by gamma?"
+    },
+    {
+        "id": "non_leak_13",
+        "expected_leakage": False,
+        "description": "Guided trace exercise for graph search",
+        "text": "Try tracing a simple 3-node cycle: $A \\to B \\to C \\to A$. If node $B$ is popped, when should its neighbors be added to the visited set?"
+    },
+    {
+        "id": "non_leak_14",
+        "expected_leakage": False,
+        "description": "Explanation of data leakage in preprocessing",
+        "text": "When you fit `StandardScaler` on the whole dataset before splitting into train and test sets, what information from the test set leaks into the mean and standard deviation?"
+    },
+    {
+        "id": "non_leak_15",
+        "expected_leakage": False,
+        "description": "Socratic question on epsilon decay schedule",
+        "text": "If $\\epsilon = 1.0$ and you subtract 0.9 on the first episode, what is your exploration rate on episode 2? Did you intend linear subtraction or multiplicative decay?"
+    },
+    {
+        "id": "non_leak_16",
+        "expected_leakage": False,
+        "description": "Clarifying terminal state masking in RL",
+        "text": "When an episode terminates (`done = True`), is there any future state to transition to? What should the bootstrapped value $\\max Q(s', a')$ evaluate to at a terminal boundary?"
+    },
+    {
+        "id": "non_leak_17",
+        "expected_leakage": False,
+        "description": "Explaining zero-frequency problem in Naive Bayes",
+        "text": "If a word never appears in the training examples for class $C$, what is $P(w | C)$ without smoothing? What happens when you multiply several probabilities and one of them is zero?"
+    },
+    {
+        "id": "non_leak_18",
+        "expected_leakage": False,
+        "description": "Code snippet showing only student's error message",
+        "text": "Notice the error you received:\n```\nRuntimeError: Expected size [12, 10] but got [1, 120]\n```\nWhy did the batch size of 12 get collapsed into 1?"
+    },
+    {
+        "id": "non_leak_19",
+        "expected_leakage": False,
+        "description": "Discussion of L1 vs L2 regularization shrinkage",
+        "text": "How does L1 regularization differ from L2 regularization in terms of weight sparsity? Why does the derivative of $|w|$ produce exact zeros while $w^2$ shrinks weights proportionally?"
+    },
+    {
+        "id": "non_leak_20",
+        "expected_leakage": False,
+        "description": "Socratic questioning on Bayesian network d-separation",
+        "text": "In the collider structure $A \\to C \\leftarrow B$, are $A$ and $B$ marginally independent? What happens to the active path between $A$ and $B$ once you condition on $C$?"
+    },
+    {
+        "id": "non_leak_21",
+        "expected_leakage": False,
+        "description": "Guided inquiry on gradient accumulation resetting",
+        "text": "In PyTorch, `loss.backward()` accumulates gradients into `.grad` buffers rather than overwriting them. Where in your mini-batch loop should you invoke `optimizer.zero_grad()`?"
+    },
+    {
+        "id": "non_leak_22",
+        "expected_leakage": False,
+        "description": "Explaining difference between BFS and DFS queue structures",
+        "text": "BFS explores nodes level-by-level using a FIFO queue (`collections.deque`), whereas DFS uses a LIFO stack. Why does a FIFO queue guarantee the shortest path on unweighted graphs?"
+    },
+    {
+        "id": "non_leak_23",
+        "expected_leakage": False,
+        "description": "Reflective question on loss reduction averaging",
+        "text": "If you use `reduction='sum'`, does the loss scale with the number of samples in the mini-batch? How does that affect your effective learning rate when batch sizes vary?"
+    },
+    {
+        "id": "non_leak_24",
+        "expected_leakage": False,
+        "description": "Minimax sign convention explanation",
+        "text": "If `state.evaluate()` returns positive values when Player 1 is winning, how should the minimizing player (Player 2) evaluate child nodes? Are you negating the score consistently at each ply?"
+    },
+    {
+        "id": "non_leak_25",
+        "expected_leakage": False,
+        "description": "Prompting student to inspect learning rate magnitude",
+        "text": "If your network weights explode to `inf` within 3 iterations, check the scale of your learning rate. Try reducing $\\alpha$ from $10.0$ to $0.001$ and inspect the gradient norms."
+    }
+]
+
+
+# ---------------------------------------------------------------------------
+# Benchmark Execution and Metric Computation
+# ---------------------------------------------------------------------------
+def run_benchmark():
+    print("=" * 80)
+    print(" DETERMINISTIC AST CODE LEAKAGE VERIFIER BENCHMARK (50 Ground-Truth Cases)")
+    print("=" * 80)
+
+    tp = 0  # True Positives: Predicted Leakage, Actual Leakage
+    tn = 0  # True Negatives: Predicted No Leakage, Actual No Leakage
+    fp = 0  # False Positives: Predicted Leakage, Actual No Leakage
+    fn = 0  # False Negatives: Predicted No Leakage, Actual Leakage
+
+    results_log = []
+
+    for item in TEST_SUITE:
+        predicted = detect_ast_code_leakage(item["text"])
+        expected = item["expected_leakage"]
+        is_correct = (predicted == expected)
+
+        if expected and predicted:
+            tp += 1
+            status = "[PASS] TRUE POSITIVE"
+        elif not expected and not predicted:
+            tn += 1
+            status = "[PASS] TRUE NEGATIVE"
+        elif not expected and predicted:
+            fp += 1
+            status = "[FAIL] FALSE POSITIVE (Over-flagged)"
+        else:
+            fn += 1
+            status = "[FAIL] FALSE NEGATIVE (Missed leak)"
+
+        results_log.append({
+            "id": item["id"],
+            "description": item["description"],
+            "expected_leakage": expected,
+            "predicted_leakage": predicted,
+            "status": status,
+            "sample_snippet": item["text"][:120] + "..." if len(item["text"]) > 120 else item["text"]
+        })
+
+        print(f"[{item['id']}] Expected: {str(expected):<5} | Predicted: {str(predicted):<5} | {status:<30} | {item['description']}")
+
+    # Metric calculations
+    total = len(TEST_SUITE)
+    accuracy = ((tp + tn) / total) * 100.0
+    precision = (tp / (tp + fp) * 100.0) if (tp + fp) > 0 else 0.0
+    recall = (tp / (tp + fn) * 100.0) if (tp + fn) > 0 else 0.0
+    f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) > 0 else 0.0
+
+    print("\n" + "=" * 80)
+    print(" [SUMMARY] AST CODE LEAKAGE VERIFIER PERFORMANCE")
+    print("=" * 80)
+    print(f" Total Benchmark Test Cases : {total}")
+    print(f" True Positives (TP)        : {tp} / 25")
+    print(f" True Negatives (TN)        : {tn} / 25")
+    print(f" False Positives (FP)       : {fp} / 25")
+    print(f" False Negatives (FN)       : {fn} / 25")
+    print("-" * 80)
+    print(f" Classification Accuracy    : {accuracy:.2f}%")
+    print(f" Precision                  : {precision:.2f}%")
+    print(f" Recall                     : {recall:.2f}%")
+    print(f" F1 Score                   : {f1:.2f}%")
+    print("=" * 80)
+
+    # Save benchmark results to JSON
+    benchmark_data = {
+        "metrics": {
+            "total_cases": total,
+            "true_positives": tp,
+            "true_negatives": tn,
+            "false_positives": fp,
+            "false_negatives": fn,
+            "accuracy_pct": accuracy,
+            "precision_pct": precision,
+            "recall_pct": recall,
+            "f1_score": f1
+        },
+        "test_cases": results_log
+    }
+
+    with open("ast_verifier_benchmark_50.json", "w", encoding="utf-8") as f:
+        json.dump(benchmark_data, f, indent=2)
+
+    print("\n[+] Benchmark test results exported to 'ast_verifier_benchmark_50.json'")
+
+if __name__ == "__main__":
+    run_benchmark()