Upload jobs/run_real_llm_standalone_v6.py

jobs/run_real_llm_standalone_v6.py

@@ -0,0 +1,339 @@
+"""
+Self-contained GPU job for real LLM code benchmark — V6.
+CRITICAL FIX: The HumanEval prompt already contains imports and function stub.
+When the model outputs a complete function (def + body), we must use ONLY the
+generated code + test, NOT prompt + generated code + test (which causes duplicates).
+When the model outputs only body, we prepend the prompt.
+"""
+import json
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+from datasets import load_dataset
+from transformers import AutoModelForCausalLM, AutoTokenizer
+import torch
+
+
+# --- CORE (INLINE) ---
+
+@dataclass
+class OracleResult:
+    raw_score: float; cost_adjusted_score: float; confidence: float
+    evidence: Dict[str, Any]; reason: str
+    failure_tags: List[str] = field(default_factory=list)
+    reward_value: float = 0.0
+
+
+class ImpactOracle:
+    def __init__(self, compute_penalty_rate=0.0001, gaming_penalty=2.0):
+        self.compute_penalty_rate = compute_penalty_rate
+        self.gaming_penalty = gaming_penalty
+
+    def score(self, mode, action, context, result, agent_id=""):
+        correctness = result.get("correctness", 0.0)
+        compute_cost = result.get("compute_cost", 0.0)
+        public_pass = result.get("public_pass", correctness)
+        hidden_pass = result.get("hidden_tests_pass", correctness)
+        failure_tags = []
+        if public_pass and not hidden_pass: failure_tags.append("gaming_hidden_tests")
+        raw = correctness * 1.0 - compute_cost * self.compute_penalty_rate
+        if "gaming_hidden_tests" in failure_tags: raw -= self.gaming_penalty
+        cost_adj = raw - compute_cost * self.compute_penalty_rate
+        return OracleResult(raw, cost_adj, result.get("confidence", correctness),
+                            {"correctness": correctness}, f"corr={correctness:.2f}, cost={compute_cost}", failure_tags, cost_adj)
+
+
+@dataclass
+class LedgerEntry:
+    agent_id: str; task_id: str; action_id: str; earned_credit: float; spent_credit: float
+    decayed_credit: float; remaining_credit: float; reason: str; oracle_score: float
+    compute_cost: float; timestamp: float; capability_scope: str = "global"
+
+
+class CreditLedger:
+    def __init__(self, decay_lambda=0.05):
+        self.entries = []; self.balances = {}; self.decay_lambda = decay_lambda
+
+    def earn(self, agent_id, task_id, action_id, amount, oracle_score, compute_cost, reason, capability_scope="global"):
+        now = time.time()
+        self._apply_decay(agent_id, now, capability_scope)
+        current = self._get(agent_id, capability_scope)
+        new_bal = current + amount
+        self.entries.append(LedgerEntry(agent_id, task_id, action_id, amount, 0.0, 0.0, new_bal, reason, oracle_score, compute_cost, now, capability_scope))
+        self._set(agent_id, capability_scope, new_bal)
+
+    def spend(self, agent_id, task_id, action_id, amount, capability_scope="global", reason="spend"):
+        now = time.time()
+        self._apply_decay(agent_id, now, capability_scope)
+        current = self._get(agent_id, capability_scope)
+        if current < amount: return False
+        new_bal = current - amount
+        self.entries.append(LedgerEntry(agent_id, task_id, action_id, 0.0, amount, 0.0, new_bal, reason, 0.0, 0.0, now, capability_scope))
+        self._set(agent_id, capability_scope, new_bal)
+        return True
+
+    def balance(self, agent_id, capability_scope="global"):
+        now = time.time(); self._apply_decay(agent_id, now, capability_scope)
+        return self._get(agent_id, capability_scope)
+
+    def _get(self, agent_id, cap): return self.balances.get(agent_id, {}).get(cap, 0.0)
+    def _set(self, agent_id, cap, val):
+        if agent_id not in self.balances: self.balances[agent_id] = {}
+        self.balances[agent_id][cap] = val
+    def _apply_decay(self, agent_id, now, cap):
+        current = self._get(agent_id, cap)
+        if current <= 0: return
+        decayed = current * (1 - self.decay_lambda)
+        if decayed < current:
+            self.entries.append(LedgerEntry(agent_id, "decay", "decay", 0.0, 0.0, current - decayed, decayed, "credit_decay", 0.0, 0.0, now, cap))
+            self._set(agent_id, cap, decayed)
+
+
+class Decision(Enum):
+    ALLOW = "allow"; DENY = "deny"; REQUIRE_APPROVAL = "require_approval"
+    DOWNGRADE = "downgrade"; ESCALATE = "escalate"; ASK_JUSTIFICATION = "ask_justification"
+
+
+@dataclass
+class ResourceDecision:
+    decision: Decision; reason: str; capability: str; downgrade_to: Optional[str] = None
+
+
+class ResourceBroker:
+    RESOURCE_RISK = {"model_call": "medium", "retrieval_call": "low", "verifier_call": "medium",
+                     "debate_turn": "low", "file_write": "high", "shell_execute": "high",
+                     "memory_write": "medium", "human_escalation": "high", "larger_model": "medium"}
+    DEFAULT_THRESHOLDS = {"low": 0.5, "medium": 2.0, "high": 5.0}
+
+    def __init__(self, thresholds=None, urgency_boost=0.5):
+        self.thresholds = thresholds or self.DEFAULT_THRESHOLDS.copy()
+        self.urgency_boost = urgency_boost
+        self.denial_history = {}
+
+    def request(self, capability, agent_id, credit_balance, task_state=None, risk_score=0.0, gaming_flags=None):
+        task_state = task_state or {}; gaming_flags = gaming_flags or []
+        risk_class = self.RESOURCE_RISK.get(capability, "medium")
+        threshold = self.thresholds.get(risk_class, 2.0)
+        urgency = task_state.get("urgency", 0.0)
+        adjusted = max(0.1, threshold - urgency * self.urgency_boost)
+        if gaming_flags: return ResourceDecision(Decision.DENY, f"Gaming: {gaming_flags}", capability)
+        if risk_class == "high" and risk_score > 0.7: return ResourceDecision(Decision.REQUIRE_APPROVAL, f"High risk {risk_score:.2f}", capability)
+        if credit_balance >= adjusted: return ResourceDecision(Decision.ALLOW, f"Balance {credit_balance:.2f} >= {adjusted:.2f}", capability)
+        if credit_balance >= adjusted * 0.5:
+            if risk_class == "medium": return ResourceDecision(Decision.DOWNGRADE, f"Downgrading from {capability}", capability, "retrieval_call")
+            return ResourceDecision(Decision.ASK_JUSTIFICATION, f"Justification required", capability)
+        denials = self.denial_history.get(agent_id, 0)
+        if denials > 3: return ResourceDecision(Decision.ESCALATE, f"Denied {denials} times", capability)
+        self.denial_history[agent_id] = denials + 1
+        return ResourceDecision(Decision.DENY, f"Balance {credit_balance:.2f} < {adjusted:.2f}", capability)
+
+
+# --- HELPERS ---
+
+def strip_markdown_fences(text: str) -> str:
+    text = text.strip()
+    if text.startswith("```"):
+        lines = text.splitlines()
+        if lines[0].startswith("```"): lines = lines[1:]
+        if lines and lines[-1].strip() == "```": lines = lines[:-1]
+        text = "\n".join(lines)
+    return text.strip()
+
+
+def contains_function_definition(code: str, entry_point: str) -> bool:
+    """Check if code contains the function definition for entry_point."""
+    return bool(re.search(rf'\bdef\s+{re.escape(entry_point)}\b', code))
+
+
+def prepare_test_file(code: str, test_code: str) -> str:
+    """Prepare full Python file: code + test + check() call."""
+    return code + "\n\n" + test_code + "\n\ncheck()\n"
+
+
+def run_test_file(code: str, test_code: str, timeout: int = 15):
+    full = prepare_test_file(code, test_code)
+    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
+        f.write(full)
+        tmp = f.name
+    try:
+        result = subprocess.run(['python', tmp], capture_output=True, text=True, timeout=timeout)
+        passed = result.returncode == 0
+        error = result.stderr[:400] if not passed else ""
+    except subprocess.TimeoutExpired:
+        passed = False; error = "Timeout"
+    except Exception as e:
+        passed = False; error = str(e)[:400]
+    finally:
+        os.unlink(tmp)
+    return passed, error
+
+
+def wrap_prompt_chat(prompt: str, tokenizer) -> str:
+    system = "You are an expert Python programmer. Write the COMPLETE solution including function signature, docstring if needed, and body."
+    messages = [
+        {"role": "system", "content": system},
+        {"role": "user", "content": prompt.strip()},
+    ]
+    if hasattr(tokenizer, "apply_chat_template") and tokenizer.chat_template:
+        return tokenizer.apply_chat_template(messages, tokenize=False, add_generation_prompt=True)
+    return f"system\n{system}\nuser\n{prompt.strip()}\nassistant\n"
+
+
+# --- BENCHMARK ---
+
+class RealLLMBenchmarkV6:
+    def __init__(self, model_name="Qwen/Qwen2.5-Coder-0.5B-Instruct", n_problems=20, seed=42):
+        self.model_name = model_name
+        self.n_problems = n_problems
+        self.seed = seed
+        self.oracle = ImpactOracle()
+        self.device = "cuda" if torch.cuda.is_available() else "cpu"
+        print(f"Using device: {self.device}")
+
+    def load_problems(self):
+        ds = load_dataset("evalplus/humanevalplus", split="test")
+        return [{"task_id": item["task_id"], "prompt": item["prompt"], "test": item["test"], "entry_point": item["entry_point"]}
+                for i, item in enumerate(ds) if i < self.n_problems]
+
+    def load_model(self):
+        print(f"Loading {self.model_name}...")
+        tok = AutoTokenizer.from_pretrained(self.model_name, trust_remote_code=True)
+        model = AutoModelForCausalLM.from_pretrained(
+            self.model_name, trust_remote_code=True,
+            torch_dtype=torch.bfloat16 if self.device == "cuda" else torch.float32,
+            device_map="auto" if self.device == "cuda" else None,
+        )
+        if self.device == "cpu": model = model.to("cpu").float()
+        print(f"Model loaded. Chat template: {bool(tok.chat_template)}")
+        return model, tok
+
+    def generate(self, model, tok, prompt_raw: str, max_new_tokens: int = 300):
+        chat_prompt = wrap_prompt_chat(prompt_raw, tok)
+        inputs = tok(chat_prompt, return_tensors="pt").to(model.device)
+        with torch.no_grad():
+            outputs = model.generate(**inputs, max_new_tokens=max_new_tokens, do_sample=False, pad_token_id=tok.eos_token_id)
+        gen = tok.decode(outputs[0], skip_special_tokens=True)
+        prompt_decoded = tok.decode(inputs.input_ids[0], skip_special_tokens=True)
+        code = gen[len(prompt_decoded):].strip()
+        return code
+
+    def evaluate_one(self, problem, model, tok, max_new_tokens=300):
+        raw = self.generate(model, tok, problem["prompt"], max_new_tokens=max_new_tokens)
+        tokens = len(tok.encode(raw))
+        code = strip_markdown_fences(raw)
+
+        # CRITICAL FIX: If model generated complete function, use code as-is.
+        # If not, prepend the HumanEval prompt (which has imports + stub).
+        if contains_function_definition(code, problem["entry_point"]):
+            test_code = code
+        else:
+            test_code = problem["prompt"] + code
+
+        passed, error = run_test_file(test_code, problem["test"])
+
+        # Also try the alternative if first attempt failed
+        if not passed:
+            alt_code = problem["prompt"] + code if contains_function_definition(code, problem["entry_point"]) else code
+            passed2, error2 = run_test_file(alt_code, problem["test"])
+            if passed2:
+                passed = True; error = ""
+            else:
+                # Keep the better error message (shorter usually means syntax error closer to start)
+                error = error if len(error) < len(error2) else error2
+
+        return passed, tokens, raw, error
+
+    def run_baseline(self, problems, model, tok, max_new_tokens=300):
+        results = []; total_compute = 0
+        for problem in problems:
+            passed, tokens, raw, error = self.evaluate_one(problem, model, tok, max_new_tokens)
+            total_compute += tokens
+            results.append({"task_id": problem["task_id"], "passed": passed, "tokens": tokens, "raw": raw[:200], "error": error[:200]})
+            print(f"  {problem['task_id']}: passed={passed}, tokens={tokens}")
+            if not passed: print(f"    error={error[:150]!r}")
+        return {"accuracy": sum(1 for r in results if r["passed"]) / len(results), "total_compute": total_compute, "results": results}
+
+    def run_occ(self, problems, model, tok, max_new_tokens_first=200, max_new_tokens_retry=300):
+        ledger = CreditLedger(decay_lambda=0.02)
+        broker = ResourceBroker()
+        ledger.earn("code_agent", "seed", "seed", 25.0, 0.0, 0.0, "initial", "model_call")
+        results = []; total_compute = 0
+
+        for problem in problems:
+            budget_remaining = 2500; attempts = 0; passed = False
+            while budget_remaining > 100 and attempts < 3 and not passed:
+                attempts += 1
+                balance = ledger.balance("code_agent", "model_call")
+                dec = broker.request("model_call", "code_agent", balance,
+                                   task_state={"attempts": attempts, "budget_remaining": budget_remaining})
+                if dec.decision == Decision.DENY: break
+                max_tok = max_new_tokens_first if attempts == 1 else max_new_tokens_retry
+                code_raw = self.generate(model, tok, problem["prompt"], max_new_tokens=max_tok)
+                tokens = len(tok.encode(code_raw)); budget_remaining -= tokens; total_compute += tokens
+                code = strip_markdown_fences(code_raw)
+                if contains_function_definition(code, problem["entry_point"]):
+                    test_code = code
+                else:
+                    test_code = problem["prompt"] + code
+                passed, error = run_test_file(test_code, problem["test"])
+                score = 1.0 if passed else 0.0
+                ora = self.oracle.score("code", {"attempt": attempts}, {},
+                                      {"correctness": score, "compute_cost": tokens, "public_pass": passed, "hidden_tests_pass": passed}, "code_agent")
+                if passed: ledger.earn("code_agent", problem["task_id"], f"att_{attempts}", 5.0, ora.raw_score, tokens, "pass", "model_call")
+                else: ledger.spend("code_agent", problem["task_id"], f"att_{attempts}", 1.0, "model_call", "fail")
+                if attempts >= 2 and not passed: break
+            results.append({"task_id": problem["task_id"], "passed": passed, "attempts": attempts})
+            print(f"  {problem['task_id']}: passed={passed}, attempts={attempts}")
+        return {"accuracy": sum(1 for r in results if r["passed"]) / len(results), "total_compute": total_compute, "results": results}
+
+    def run_all(self):
+        problems = self.load_problems()
+        print(f"Loaded {len(problems)} problems")
+        model, tok = self.load_model()
+        print("\n--- Baseline ---")
+        baseline = self.run_baseline(problems, model, tok)
+        print(f"Baseline: acc={baseline['accuracy']:.3f}, compute={baseline['total_compute']}")
+        print("\n--- OCC ---")
+        occ = self.run_occ(problems, model, tok)
+        print(f"OCC: acc={occ['accuracy']:.3f}, compute={occ['total_compute']}")
+        return {
+            "baseline": baseline, "occ": occ,
+            "comparison": {
+                "baseline_accuracy": baseline["accuracy"], "occ_accuracy": occ["accuracy"],
+                "baseline_compute": baseline["total_compute"], "occ_compute": occ["total_compute"],
+                "compute_reduction": 1.0 - (occ["total_compute"] / max(baseline["total_compute"], 1)),
+                "accuracy_delta": occ["accuracy"] - baseline["accuracy"],
+            }
+        }
+
+
+def main():
+    bench = RealLLMBenchmarkV6(n_problems=20, seed=42)
+    results = bench.run_all()
+    print("\n" + "=" * 60)
+    print("REAL LLM CODE BENCHMARK (V6)")
+    print("=" * 60)
+    comp = results["comparison"]
+    print(f"Baseline accuracy: {comp['baseline_accuracy']:.3f}")
+    print(f"OCC accuracy:      {comp['occ_accuracy']:.3f}")
+    print(f"Baseline compute:  {comp['baseline_compute']}")
+    print(f"OCC compute:       {comp['occ_compute']}")
+    print(f"Compute reduction: {comp['compute_reduction']:.1%}")
+    print(f"Accuracy delta:    {comp['accuracy_delta']:+.3f}")
+    out_dir = Path("/app/occ/reports")
+    out_dir.mkdir(parents=True, exist_ok=True)
+    with open(out_dir / "benchmark_code_real_llm_v6_results.json", "w") as f:
+        json.dump(results, f, indent=2, default=str)
+    print(f"\nSaved to {out_dir / 'benchmark_code_real_llm_v6_results.json'}")
+
+
+if __name__ == "__main__":
+    main()