"""Hugging Face Script Competition Entrypoint (`huggingface/competitions`) for `ProbioticFarmer/Qwen3.8-27B-BarronPolyStep-RealCoder-v3`. When executed inside a Hugging Face Competition GPU container (`a100-large`): 1. Reads the hidden/private test dataset from `COMPETITION_DATA_DIR` (or `--data-dir`), supporting `test.jsonl` or `test.csv`. 2. Loads the 64-layer `Qwen/Qwen3.8-27B` hybrid backbone (`48 GatedDeltaNet` Triton + `16 FlashAttention-2` layers) and applies the Layer-63 FP32 Barron PolyStep PEFT LoRA adapter (`adapter_model.safetensors`, `lora_scale=2.0`). 3. Scores candidate code completions (or generates free-form solutions) using exact multi-token sequence log-probability under the Barron PolyStep head. 4. Writes `submission.csv` and `submission.jsonl` for automated private-test grading. """ from __future__ import annotations import argparse import csv import json import os import sys import time from typing import Any, Dict, List, Tuple import numpy as np import torch import torch.nn.functional as F SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__)) if SCRIPT_DIR not in sys.path: sys.path.insert(0, SCRIPT_DIR) from cuda_qwen38_gpu import MLXQwen38Engine def find_lora_path() -> str: candidates = [ os.path.join(SCRIPT_DIR, "adapter_model.safetensors"), os.path.join( SCRIPT_DIR, "checkpoints", "Qwen3.8-27B-BarronPolyStep-RealCoder-v3", "adapter_model.safetensors", ), ] for p in candidates: if os.path.exists(p): return p raise FileNotFoundError(f"Could not find adapter_model.safetensors in {candidates}") @torch.inference_mode() def score_problem_candidates( engine: MLXQwen38Engine, prefix: str, candidates: List[str], ) -> Tuple[List[float], str]: """Score candidate completions using 64-layer forward pass + chunked candidate evaluation.""" p_ids = engine.tokenizer.encode(prefix).ids engine.reset_cache() logits_p = engine._prefill_chunk(p_ids, offset=0) engine.cached_tokens = list(p_ids) engine.snapshot_delta_state(len(p_ids), pinned=True) lp0 = F.log_softmax(logits_p[0].float(), dim=-1) scores: List[float] = [] for cand in candidates: c_ids = engine.tokenizer.encode(cand).ids if not c_ids: scores.append(-1e9) continue total_lp = float(lp0[c_ids[0]].item()) if len(c_ids) > 1: engine.restore_to_arbitrary_index(len(p_ids)) cur_off = len(p_ids) for idx in range(len(c_ids) - 1): lg = engine._prefill_chunk([c_ids[idx]], offset=cur_off) cur_off += 1 lpf = F.log_softmax(lg[0].float(), dim=-1) total_lp += float(lpf[c_ids[idx + 1]].item()) scores.append(total_lp / len(c_ids)) best_idx = int(np.argmax(scores)) return scores, candidates[best_idx] def load_blind_test_items(data_dir: str) -> List[Dict[str, Any]]: jsonl_path = os.path.join(data_dir, "test.jsonl") csv_path = os.path.join(data_dir, "test.csv") items: List[Dict[str, Any]] = [] if os.path.exists(jsonl_path): with open(jsonl_path, encoding="utf-8") as f: for line in f: if line.strip(): items.append(json.loads(line)) return items if os.path.exists(csv_path): with open(csv_path, encoding="utf-8") as f: reader = csv.DictReader(f) for row in reader: cands = json.loads(row["candidates"]) if "candidates" in row and row["candidates"] else [] items.append( { "id": row.get("id", ""), "task_id": row.get("task_id", row.get("id", "")), "prefix": row.get("prefix", row.get("prompt", "")), "candidates": cands, "suffix": row.get("suffix", ""), } ) return items raise FileNotFoundError(f"No test.jsonl or test.csv found in {data_dir}") def main() -> None: parser = argparse.ArgumentParser(description="HF Script Competition Runner for PolyStep-RealCoder-v3") parser.add_argument( "--data-dir", type=str, default=os.environ.get("COMPETITION_DATA_DIR", "/tmp/data"), help="Path to mounted private test dataset directory", ) parser.add_argument( "--output", type=str, default="submission.csv", help="Output CSV path expected by Hugging Face Competitions evaluator", ) parser.add_argument( "--model-mode", type=str, choices=["polystep", "base"], default="polystep", help="Run with Barron PolyStep LoRA ('polystep') or unmodified Qwen3.8-27B ('base')", ) args = parser.parse_args() t0 = time.time() lora_path = find_lora_path() if args.model_mode == "polystep" else None engine = MLXQwen38Engine(lora_path=lora_path, lora_scale=2.0 if lora_path else 0.0) if args.model_mode == "base": engine.active_mlp_lora.clear() engine.active_attn_lora.clear() items = load_blind_test_items(args.data_dir) print( f"[HF Competition Runner] Loaded {len(items)} blind test problems from {args.data_dir} " f"(mode={args.model_mode})", flush=True, ) predictions: List[Dict[str, Any]] = [] for idx, item in enumerate(items, 1): prob_id = item.get("id", idx) task_id = item.get("task_id", f"task_{prob_id}") prefix = item["prefix"] suffix = item.get("suffix", "") candidates = item["candidates"] scores, chosen_code = score_problem_candidates(engine, prefix, candidates) pred_idx = int(np.argmax(scores)) sorted_sc = sorted(scores, reverse=True) margin = float(sorted_sc[0] - sorted_sc[1]) if len(sorted_sc) > 1 else 0.0 predictions.append( { "id": prob_id, "task_id": task_id, "prediction_idx": pred_idx, "predicted_code": chosen_code, "confidence_margin_nats": round(margin, 4), "solution": prefix + chosen_code + suffix, } ) print( f" [Blind Test {idx:02d}/{len(items):02d}] id={prob_id} ({task_id}) -> " f"pred_idx={pred_idx} ({chosen_code.strip()!r}, margin={margin:+.4f} nats)", flush=True, ) os.makedirs(os.path.dirname(os.path.abspath(args.output)), exist_ok=True) with open(args.output, "w", newline="", encoding="utf-8") as f: writer = csv.DictWriter( f, fieldnames=["id", "task_id", "prediction_idx", "predicted_code", "confidence_margin_nats"], ) writer.writeheader() for p in predictions: writer.writerow( { "id": p["id"], "task_id": p["task_id"], "prediction_idx": p["prediction_idx"], "predicted_code": p["predicted_code"], "confidence_margin_nats": p["confidence_margin_nats"], } ) jsonl_out = os.path.splitext(args.output)[0] + ".jsonl" with open(jsonl_out, "w", encoding="utf-8") as f: for p in predictions: f.write(json.dumps(p) + "\n") print( f"[HF Competition Runner] Wrote {len(predictions)} predictions to {args.output} " f"and {jsonl_out} in {time.time() - t0:.2f}s.", flush=True, ) if __name__ == "__main__": main()