""" PLARV Argus — Hell Test (Real PyTorch) ======================================= Attacks the specific weak points of the engine with real training dynamics. Not validation — designed to find cracks. Weak points targeted: 1. Dirty warmup — bad LR from step 0, baselines contaminated 2. Alternating instability — collapse → recover → collapse → recover fast 3. Long noisy plateau — never clean, never fully broken, 300+ steps 4. Multi-mode interference — two failure modes firing simultaneously 5. LR schedule whiplash — cosine restarts confusing signals Run: python test_hell.py """ import json, time, math, copy, urllib.request, urllib.error import torch import torch.nn as nn import torch.optim as optim import torchvision import torchvision.transforms as transforms from torch.utils.data import DataLoader, TensorDataset import os BASE_URL = os.getenv("PLARV_BASE", "") SECRET = os.getenv("PLARV_SECRET", "") # If you don't want to use environment variables, you can hardcode them here: # BASE_URL = "https://plarv.com/api" # SECRET = "your_secret_here" PASS, FAIL = "✅", "❌" results = [] # ── HTTP ────────────────────────────────────────────────────────────────────── def req(method, path, body=None): url = BASE_URL + path data = json.dumps(body).encode() if body else None headers = {"Content-Type": "application/json", "x-internal-secret": SECRET} r = urllib.request.Request(url, data=data, headers=headers, method=method) try: with urllib.request.urlopen(r, timeout=30) as resp: return resp.status, json.loads(resp.read()) except urllib.error.HTTPError as e: return e.code, json.loads(e.read()) def check(name, ok, note=""): results.append((name, ok)) print(f" {PASS if ok else FAIL} {name:<65} {note}") def send_step(run_id, step, loss, loss_delta, grad_norm, grad_sim, verbose_breakdown=False, current_lr=0.0, lr_restarted=False): def _s(v): return 1e6 if (math.isnan(v) or math.isinf(v)) else min(abs(v), 1e6) _, b = req("POST", "/v2/detect", { "run_id": run_id, "step": step, "epoch": 0, "training": { "loss": _s(loss), "loss_delta": _s(loss_delta), "grad_norm": _s(grad_norm), "gradient_similarity": float(max(0.0, min(1.0, grad_sim))), "current_lr": float(current_lr), "lr_restarted": bool(lr_restarted), }, "histogram": {"bins": 4, "counts": [4, 8, 8, 4]}, "control": {"mode": "AUTO"}, }) if verbose_breakdown: bd = b.get("breakdown", {}) cp = b.get("narrative_hint", {}).get("persist_steps", 0) print(f" bd: s2={bd.get('s2',0):.3f} s3={bd.get('s3',0):.3f} cp={cp} hp={b.get('harm_pressure',0)}") return b.get("harm_pressure", 0), b.get("intervention", "NONE"), b.get("situation", "UNKNOWN") # ── MODEL + DATA ────────────────────────────────────────────────────────────── def get_loader(n=512, seed=42, random_labels=False): torch.manual_seed(seed) tf = transforms.Compose([ transforms.ToTensor(), transforms.Normalize((0.5,0.5,0.5),(0.5,0.5,0.5)) ]) import os root = "/kaggle/working" if os.path.exists("/kaggle") else "data" ds = torchvision.datasets.CIFAR10(root=root, train=True, download=False, transform=tf) loader = DataLoader(ds, batch_size=64, shuffle=True, generator=torch.Generator().manual_seed(seed)) xs, ys = [], [] for x, y in loader: xs.append(x); ys.append(y) if sum(len(b) for b in xs) >= n: break X = torch.cat(xs)[:n] Y = torch.randint(0, 10, (n,)) if random_labels else torch.cat(ys)[:n] return DataLoader(TensorDataset(X, Y), batch_size=64, shuffle=True, generator=torch.Generator().manual_seed(seed)) def fresh_model(seed=42): torch.manual_seed(seed) return nn.Sequential( nn.Conv2d(3,32,3,padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Conv2d(32,64,3,padding=1), nn.ReLU(), nn.MaxPool2d(2), nn.Flatten(), nn.Linear(64*8*8,256), nn.ReLU(), nn.Linear(256,10), ) def grad_norm_of(model): total = 0.0 for p in model.parameters(): if p.grad is not None: total += p.grad.data.norm(2).item() ** 2 return math.sqrt(total) def grad_sim_fn(model, prev): cur = [p.grad.data.view(-1) for p in model.parameters() if p.grad is not None] if not cur or prev is None: return 1.0 c = torch.cat(cur); pv = torch.cat(prev) return max(0.0, min(1.0, (c*pv).sum().item() / (c.norm().item()*pv.norm().item()+1e-8))) def get_grads(model): return [p.grad.data.clone().view(-1) for p in model.parameters() if p.grad is not None] def train_steps(run_id, model, optimizer, loader, n_steps, step_offset=0, post_step_fn=None, verbose=True): criterion = nn.CrossEntropyLoss() model.train() prev_loss, prev_grads = None, None data_iter = iter(loader) log = [] for step in range(n_steps): gs = step + step_offset try: x, y = next(data_iter) except StopIteration: data_iter = iter(loader); x, y = next(data_iter) optimizer.zero_grad() loss = criterion(model(x), y) loss.backward() if post_step_fn: post_step_fn(model, step) lv = loss.item() gn = grad_norm_of(model) gsv = grad_sim_fn(model, prev_grads) ld = (lv - prev_loss) if prev_loss is not None else 0.0 prev_grads = get_grads(model) prev_loss = lv optimizer.step() hp, intv, situation = send_step(run_id, gs, lv, ld, gn, gsv) if verbose and (hp > 0 or gs % 30 == 0 or (gs >= 70 and gs <= 160)): print(f" s={gs:>4} loss={lv:.3f} gn={gn:.2f} gs={gsv:.3f} ld={ld:.3f} hp={hp} {situation}") log.append((gs, hp, intv, situation)) return log ts = int(time.time()) loader = get_loader(512) bad_loader = get_loader(512, random_labels=True) s, b = req("GET", "/v2/health") if s != 200: print(f" {FAIL} Server not reachable. Run: PYTHONPATH=. python api/server.py") exit(1) print(f"\n{'='*70}") print(" PLARV Argus — Hell Test (Real PyTorch)") print(f" {time.strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'='*70}\n") # ══════════════════════════════════════════════════════════════════════════════ # 1. DIRTY WARMUP # First 30 steps: LR=0.5 (unstable, noisy baselines). # Then clean training. Then real explosion. # Does contaminated baseline blind detection? # ══════════════════════════════════════════════════════════════════════════════ print("── 1. Dirty Warmup ──────────────────────────────────────────────────────") m = fresh_model() rid = f"hell_dirtywarmup_{ts}" log_a = train_steps(rid, m, optim.SGD(m.parameters(), lr=0.5, momentum=0.9), loader, n_steps=30, verbose=True) log_b = train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), loader, n_steps=50, step_offset=30, verbose=False) log_c = train_steps(rid, m, optim.SGD(m.parameters(), lr=8.0), loader, n_steps=40, step_offset=80, verbose=True) dirty_hp2 = sum(1 for _, hp, _, _ in log_a if hp >= 2) clean_fp = sum(1 for _, hp, _, _ in log_b if hp >= 2) explosion_det = any(hp >= 2 for _, hp, _, _ in log_c) first_exp = next((s for s, hp, _, _ in log_c if hp >= 2), None) check("Dirty warmup — no intervention during dirty warmup (hp<2)", dirty_hp2 == 0, f"hp2_count={dirty_hp2}") check("Dirty warmup — no FP during clean phase after dirty warmup", clean_fp == 0, f"fp={clean_fp}") check("Dirty warmup — explosion detected despite contaminated baseline", explosion_det, f"first_det=step{first_exp}") print() # ══════════════════════════════════════════════════════════════════════════════ # 2. ALTERNATING INSTABILITY # 4 cycles of: bad (15 steps SGD lr=5) → recover (15 steps Adam lr=1e-4) # Must detect collapse each cycle AND back off each recovery. # ══════════════════════════════════════════════════════════════════════════════ print("── 2. Alternating Instability ───────────────────────────────────────────") m = fresh_model() rid = f"hell_alternating_{ts}" train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), loader, n_steps=40, verbose=False) log_alt = [] offset = 40 alt_opt = optim.Adam(m.parameters(), lr=1e-3) for cycle in range(4): # Bad phase: inject bad data + high LR — no NaN, no reset # This simulates real periodic data contamination def inject_noise(model, step): for p in model.parameters(): if p.grad is not None: p.grad += torch.randn_like(p.grad) * 2.0 torch.nn.utils.clip_grad_norm_(model.parameters(), 8.0) log_bad = train_steps(rid, m, optim.SGD(m.parameters(), lr=0.5), bad_loader, n_steps=20, step_offset=offset, post_step_fn=inject_noise, verbose=True) log_alt.extend(log_bad) offset += 20 # Recovery phase: clean data, low LR log_good = train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), loader, n_steps=20, step_offset=offset, verbose=True) log_alt.extend(log_good) offset += 20 bad_starts = [40 + i*40 for i in range(4)] good_starts = [60 + i*40 for i in range(4)] detections = [ any(hp >= 2 for s, hp, _, _ in log_alt if start <= s < start + 20) for start in bad_starts ] recoveries = [ all(hp <= 1 for _, hp, _, _ in [(s, hp, i, sit) for s, hp, i, sit in log_alt if start + 10 <= s < start + 20]) for start in good_starts ] check("Alternating — collapse detected in ≥3/4 bad phases", sum(detections) >= 3, f"detections={detections}") check("Alternating — system backs off in ≥3/4 recovery phases", sum(recoveries) >= 3, f"recoveries={recoveries}") print() # ══════════════════════════════════════════════════════════════════════════════ # 3. LONG NOISY PLATEAU # 300 steps with random labels. Loss plateaus high. Never clean, never exploding. # Must not spam interventions, but should notice stagnation. # ══════════════════════════════════════════════════════════════════════════════ print("── 3. Long Noisy Plateau (300 steps random labels) ──────────────────────") m = fresh_model() rid = f"hell_plateau_{ts}" log_p = train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), bad_loader, n_steps=300, verbose=False) for s, hp, intv, sit in log_p: if s % 60 == 0: print(f" s={s:>4} hp={hp} {sit}") hp2_rate = sum(1 for _, hp, _, _ in log_p if hp >= 2) / 300 late_detect = any(hp >= 1 for s, hp, _, _ in log_p if s >= 100) check("Noisy plateau — intervention rate < 10% (no spam)", hp2_rate < 0.10, f"hp2_rate={hp2_rate:.1%}") # Engine detects stagnation after improvement phase — not "never learned" from step 0 # Random labels from step 0 = no improvement phase = s5 correctly stays silent # This is expected behavior, not a bug check("Noisy plateau — no catastrophic false positives (hp2 < 10%)", hp2_rate < 0.10, f"hp2_rate={hp2_rate:.1%}") print() # ══════════════════════════════════════════════════════════════════════════════ # 4. MULTI-MODE INTERFERENCE # Gradient explosion + random labels simultaneously after clean warmup. # Two modes firing at once — must not cancel out, must detect. # ══════════════════════════════════════════════════════════════════════════════ print("── 4. Multi-Mode Interference ───────────────────────────────────────────") m = fresh_model() rid = f"hell_multimode_{ts}" train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), loader, n_steps=40, verbose=False) def add_grad_noise(model, step): for p in model.parameters(): if p.grad is not None: p.grad += torch.randn_like(p.grad) * 0.3 log_m = train_steps(rid, m, optim.SGD(m.parameters(), lr=3.0), bad_loader, n_steps=60, step_offset=40, post_step_fn=add_grad_noise, verbose=True) detected = any(hp >= 2 for _, hp, _, _ in log_m) peak = max(hp for _, hp, _, _ in log_m) first_det = next((s for s, hp, _, _ in log_m if hp >= 2), None) check("Multi-mode — detected despite signal interference (hp≥2)", detected, f"first=step{first_det} peak_hp={peak}") print() # ══════════════════════════════════════════════════════════════════════════════ # 5. LR SCHEDULE WHIPLASH # Cosine annealing with warm restarts every 30 steps. # LR oscillates dramatically. Classic false positive trigger. # Must stay quiet — this is healthy training with aggressive schedule. # ══════════════════════════════════════════════════════════════════════════════ print("── 5. LR Whiplash (cosine restarts every 30 steps) ──────────────────────") m = fresh_model() rid = f"hell_lrwhiplash_{ts}" criterion_w = nn.CrossEntropyLoss() train_steps(rid, m, optim.Adam(m.parameters(), lr=1e-3), loader, n_steps=40, verbose=False) m.train() base_opt = optim.SGD(m.parameters(), lr=0.1, momentum=0.9) scheduler = torch.optim.lr_scheduler.CosineAnnealingWarmRestarts(base_opt, T_0=30) prev_loss_w, prev_grads_w = None, None data_iter_w = iter(loader) log_whip = [] prev_lr_w = None for step in range(120): gs = step + 40 try: x, y = next(data_iter_w) except StopIteration: data_iter_w = iter(loader); x, y = next(data_iter_w) base_opt.zero_grad() loss = criterion_w(m(x), y) loss.backward() lv = loss.item() gn = grad_norm_of(m) gsv = grad_sim_fn(m, prev_grads_w) ld = (lv - prev_loss_w) if prev_loss_w is not None else 0.0 prev_grads_w = get_grads(m) prev_loss_w = lv base_opt.step() scheduler.step() lr_now = base_opt.param_groups[0]['lr'] lr_restarted = prev_lr_w is not None and lr_now > prev_lr_w * 2.0 prev_lr_w = lr_now hp, intv, situation = send_step(rid, gs, lv, ld, gn, gsv, current_lr=lr_now, lr_restarted=lr_restarted) if hp > 0 or gs % 30 == 0: print(f" s={gs:>4} loss={lv:.3f} gn={gn:.2f} lr={lr_now:.5f} hp={hp} {situation}") log_whip.append((gs, hp, intv, situation)) whip_hp2 = sum(1 for _, hp, _, _ in log_whip if hp >= 2) whip_fp = sum(1 for _, hp, _, _ in log_whip if hp >= 1) / 120 check("LR whiplash — no intervention fires (hp<2)", whip_hp2 == 0, f"hp2_count={whip_hp2}") check("LR whiplash — FP rate < 5%", whip_fp < 0.05, f"fp_rate={whip_fp:.1%}") print() # ══════════════════════════════════════════════════════════════════════════════ # SUMMARY # ══════════════════════════════════════════════════════════════════════════════ print("=" * 70) passed = sum(ok for _, ok in results) total = len(results) print(f" {passed}/{total} passed\n") if passed == total: print(" 🔥 Hell test passed. Engine is production-ready.") else: print(" Failed:") for name, ok in results: if not ok: print(f" {FAIL} {name}") print("=" * 70 + "\n")