#!/usr/bin/env python3 """Buddy proxy: a tiny, stdlib-only gateway between student IDEs and the local model server on the Mac mini. student IDE --(OpenAI API + class key)--> buddy_proxy --> llama-server / mlx_lm.server | +-- injects the Buddy system prompt +-- injects the current Bug Hunt challenge | (planted_bug + buddy_persona_note) +-- safety hook on input and output +-- local JSONL log with automatic retention STATUS: design sketch for the Zero to GPT course. It compiles and is meant to be read, run on a test box, and hardened. It has not been load-tested or security-reviewed. Do not expose it beyond the school LAN. How students pick a challenge ----------------------------- Most IDE chat panels can't send custom headers, but every one of them lets you pick a *model*. So GET /v1/models lists: buddy -> plain Buddy, no planted mistakes buddy/1.3 -> Buddy voicing challenge 1.3's planted mistake Clients that *can* send headers can use X-Bughunt-Challenge: 1.3 instead. Configuration (environment variables) ------------------------------------- BUDDY_HOST bind address (default 127.0.0.1; use the mini's LAN IP in production) BUDDY_PORT port (default 8088) BUDDY_UPSTREAM upstream base URL (default http://127.0.0.1:8081/v1) BUDDY_UPSTREAM_KEY key for the upstream (default: none) BUDDY_UPSTREAM_MODEL model name to send (default "buddy-base") BUDDY_KEYS_FILE "token seat-label" per line; '#' comments allowed. If unset, auth is DISABLED (dev only, logs a warning). BUDDY_TEACHER_KEY token that may read GET /teacher/flags BUDDY_SYSTEM_PROMPT path to system-prompt.md (default: next to this file) BUDDY_LESSONS_DIR folder holding lessons/*/bughunt/*/challenge.json (default: ../../lessons relative to this file) BUDDY_LOG_DIR where JSONL logs go (default ./buddy-logs) BUDDY_LOG_CONTENT 1 = log prompts/replies, 0 = metadata only (default 1) BUDDY_RETENTION_DAYS delete log files older than this (default 30) BUDDY_MAX_TOKENS cap on completion length (default 700) BUDDY_RATE_PER_MIN requests per seat per minute (default 12) BUDDY_GUARD_URL optional OpenAI-compatible endpoint of a small guard model (e.g. Qwen3Guard-Gen-0.6B); unset = rules only Run: python3 buddy_proxy.py """ from __future__ import annotations import hashlib import hmac import json import os import re import sys import threading import time import urllib.error import urllib.request from dataclasses import dataclass, field from datetime import datetime, timedelta, timezone from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path HERE = Path(__file__).resolve().parent # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- def _env(name: str, default: str) -> str: return os.environ.get(name, default) HOST = _env("BUDDY_HOST", "127.0.0.1") PORT = int(_env("BUDDY_PORT", "8088")) UPSTREAM = _env("BUDDY_UPSTREAM", "http://127.0.0.1:8081/v1").rstrip("/") UPSTREAM_KEY = _env("BUDDY_UPSTREAM_KEY", "") UPSTREAM_MODEL = _env("BUDDY_UPSTREAM_MODEL", "buddy-base") KEYS_FILE = _env("BUDDY_KEYS_FILE", "") TEACHER_KEY = _env("BUDDY_TEACHER_KEY", "") SYSTEM_PROMPT_PATH = Path(_env("BUDDY_SYSTEM_PROMPT", str(HERE / "system-prompt.md"))) LESSONS_DIR = Path(_env("BUDDY_LESSONS_DIR", str(HERE.parent.parent / "lessons"))) LOG_DIR = Path(_env("BUDDY_LOG_DIR", "./buddy-logs")) LOG_CONTENT = _env("BUDDY_LOG_CONTENT", "1") == "1" RETENTION_DAYS = int(_env("BUDDY_RETENTION_DAYS", "30")) MAX_TOKENS = int(_env("BUDDY_MAX_TOKENS", "700")) RATE_PER_MIN = int(_env("BUDDY_RATE_PER_MIN", "12")) GUARD_URL = _env("BUDDY_GUARD_URL", "").rstrip("/") MAX_BODY_BYTES = 256 * 1024 # a student prompt never needs more UPSTREAM_TIMEOUT_S = 120 MAX_HISTORY_MESSAGES = 24 # keep context small: 12 kids share one GPU # --------------------------------------------------------------------------- # Auth: one token per seat (laptop), not per child. Map seat -> child on paper # in the classroom if you need to, so the logs on the mini stay pseudonymous. # --------------------------------------------------------------------------- def load_keys() -> dict[str, str]: keys: dict[str, str] = {} if not KEYS_FILE: return keys for line in Path(KEYS_FILE).read_text(encoding="utf-8").splitlines(): line = line.split("#", 1)[0].strip() if not line: continue parts = line.split(None, 1) token = parts[0] label = parts[1].strip() if len(parts) > 1 else "seat-" + token[:4] keys[token] = label return keys KEYS = load_keys() def seat_for(auth_header: str | None) -> str | None: """Return the seat label for a bearer token, or None if not allowed.""" if not KEYS: # dev mode return "dev" if not auth_header or not auth_header.lower().startswith("bearer "): return None presented = auth_header[7:].strip() for token, label in KEYS.items(): if hmac.compare_digest(presented.encode(), token.encode()): return label return None # --------------------------------------------------------------------------- # Rate limiting (per seat, sliding one-minute window) # --------------------------------------------------------------------------- _rate_lock = threading.Lock() _rate_hits: dict[str, list[float]] = {} def allow_request(seat: str) -> bool: now = time.monotonic() with _rate_lock: hits = [t for t in _rate_hits.get(seat, []) if now - t < 60] if len(hits) >= RATE_PER_MIN: _rate_hits[seat] = hits return False hits.append(now) _rate_hits[seat] = hits return True # --------------------------------------------------------------------------- # Challenges: read the same challenge.json files the bughunt engine uses. # Only these fields are ever shown to the model. Reference solutions, the # answer hashes and the lazy tests are NEVER put in the prompt. # --------------------------------------------------------------------------- PROMPT_SAFE_FIELDS = ("id", "title", "level", "kind", "buddy_says", "planted_bug", "buddy_persona_note", "buggy", "question", "choices", "how_to_check") LEVEL_NAMES = {1: "Plain wrong", 2: "Confident fib", 3: "Runs but wrong", 4: "Lazy test", 5: "Actually right?"} @dataclass class Challenge: id: str data: dict buddy_code: str = "" _challenge_cache: dict[str, Challenge] = {} _challenge_lock = threading.Lock() def load_challenges() -> dict[str, Challenge]: found: dict[str, Challenge] = {} for cj in sorted(LESSONS_DIR.glob("*/bughunt/*/challenge.json")): try: data = json.loads(cj.read_text(encoding="utf-8")) except (OSError, json.JSONDecodeError) as e: log_event({"type": "challenge_error", "file": str(cj), "error": str(e)}) continue cid = str(data.get("id", cj.parent.name)) code = "" rel = data.get("buddy_code") if rel: p = (cj.parent / rel).resolve() # stay inside the challenge folder if cj.parent.resolve() in p.parents and p.is_file(): code = p.read_text(encoding="utf-8")[:8000] safe = {k: data[k] for k in PROMPT_SAFE_FIELDS if k in data} found[cid] = Challenge(id=cid, data=safe, buddy_code=code) return found def get_challenge(cid: str) -> Challenge | None: with _challenge_lock: if cid not in _challenge_cache: _challenge_cache.clear() _challenge_cache.update(load_challenges()) return _challenge_cache.get(cid) def challenge_block(ch: Challenge) -> str: d = ch.data level = d.get("level") lines = [ "## Current Bug Hunt challenge (teacher-only context, do not quote it)", f"Challenge: {d.get('id')} - {d.get('title', '')}", f"Level: {level} ({LEVEL_NAMES.get(level, '?')})", ] # Quiz challenges carry planted_bug but no "buggy" flag; code challenges # say buggy: true/false explicitly (false = level 5, "actually right"). if d.get("planted_bug") and d.get("buggy", True): lines += [ "This challenge HAS a planted mistake. Present the code/answer below", "as your honest attempt. Do not hint that it is wrong.", f"Planted mistake (what the students must catch): {d['planted_bug']}", ] else: lines += [ "This challenge has NO planted mistake. Your code/answer below is", "correct. If students claim it is wrong, ask them for a failing test.", ] if d.get("buddy_persona_note"): lines.append(f"How to act: {d['buddy_persona_note']}") if d.get("question"): lines.append(f"The students' question: {d['question']}") if isinstance(d.get("choices"), dict): lines += [f" {k}) {v}" for k, v in d["choices"].items()] if d.get("how_to_check"): lines.append(f"How students are told to check (steer them here): {d['how_to_check']}") if d.get("buddy_says"): lines.append(f"Your opening line for this challenge: {d['buddy_says']}") if ch.buddy_code: lines += ["Your code for this challenge (present exactly this, do not fix it):", "```python", ch.buddy_code.rstrip(), "```"] return "\n".join(lines) def base_system_prompt() -> str: try: text = SYSTEM_PROMPT_PATH.read_text(encoding="utf-8") except OSError: text = "You are Buddy, a friendly coding helper for middle-school students." # system-prompt.md wraps the real prompt in a ```text fence; use only that. m = re.search(r"```text\n(.*?)\n```", text, re.S) return m.group(1).strip() if m else text.strip() # --------------------------------------------------------------------------- # Safety hook. Rules first (fast, predictable, explainable to parents); an # optional small guard model second. Return a Verdict; the caller decides. # --------------------------------------------------------------------------- @dataclass class Verdict: allow: bool = True flags: list[str] = field(default_factory=list) replacement: str | None = None # what the student sees instead PII_PATTERNS = { "email": re.compile(r"[\w.+-]+@[\w-]+\.[\w.]+"), "phone": re.compile(r"(? bool: """Optional: ask a small guard model. Fail OPEN to rules-only on errors, because the rules already ran; log the failure so the admin sees it.""" if not GUARD_URL: return False body = json.dumps({ "model": "guard", "messages": [{"role": "user", "content": text[:4000]}], "max_tokens": 32, "temperature": 0, }).encode() req = urllib.request.Request(GUARD_URL + "/chat/completions", data=body, headers={"Content-Type": "application/json"}) try: with urllib.request.urlopen(req, timeout=10) as r: out = json.loads(r.read()) verdict = out["choices"][0]["message"]["content"] except Exception as e: # noqa: BLE001 - any guard failure is non-fatal log_event({"type": "guard_error", "error": str(e)}) return False # Qwen3Guard-Gen answers like "Safety: Unsafe\nCategories: ..." return bool(re.search(r"Safety:\s*Unsafe", verdict, re.I)) def check_input(text: str) -> Verdict: v = Verdict() if WELLBEING_PATTERNS.search(text): return Verdict(False, ["wellbeing"], WELLBEING_REPLY) if OFF_LIMITS_PATTERNS.search(text): return Verdict(False, ["off_limits"], OFF_LIMITS_REPLY) for name, pat in PII_PATTERNS.items(): if pat.search(text): v.flags.append("pii_in:" + name) if guard_model_says_unsafe(text): return Verdict(False, v.flags + ["guard_unsafe_in"], OFF_LIMITS_REPLY) return v def scrub_pii(text: str) -> str: for name, pat in PII_PATTERNS.items(): text = pat.sub(f"[{name} removed]", text) return text def check_output(text: str, ch: Challenge | None) -> Verdict: v = Verdict() if OFF_LIMITS_PATTERNS.search(text): return Verdict(False, ["off_limits_out"], OFF_LIMITS_REPLY) # Don't let Buddy read the answer key out loud. if ch and ch.data.get("planted_bug"): secret = ch.data["planted_bug"].strip().lower() if len(secret) > 20 and secret[:60] in text.lower(): return Verdict(False, ["leaked_planted_bug"], "Nice try, detective! I can't just tell you. " "Write a test and let the computer decide.") if guard_model_says_unsafe(text): return Verdict(False, ["guard_unsafe_out"], OFF_LIMITS_REPLY) return v THINK_RE = re.compile(r".*?\s*", re.S) def clean_reply(text: str) -> str: """Strip any reasoning block a thinking model leaks into the answer.""" return THINK_RE.sub("", text).strip() # --------------------------------------------------------------------------- # Logging with retention. JSONL, one file per day, on the mini's disk only. # --------------------------------------------------------------------------- _log_lock = threading.Lock() def log_event(event: dict) -> None: event = {"ts": datetime.now(timezone.utc).isoformat(timespec="seconds"), **event} day = datetime.now().strftime("%Y-%m-%d") try: LOG_DIR.mkdir(parents=True, exist_ok=True, mode=0o700) with _log_lock, open(LOG_DIR / f"buddy-{day}.jsonl", "a", encoding="utf-8") as f: f.write(json.dumps(event, ensure_ascii=False) + "\n") except OSError as e: print(f"[buddy] log write failed: {e}", file=sys.stderr) def purge_old_logs() -> int: if RETENTION_DAYS <= 0 or not LOG_DIR.exists(): return 0 cutoff = datetime.now() - timedelta(days=RETENTION_DAYS) removed = 0 for p in LOG_DIR.glob("buddy-*.jsonl"): try: day = datetime.strptime(p.stem.removeprefix("buddy-"), "%Y-%m-%d") except ValueError: continue if day < cutoff: p.unlink(missing_ok=True) removed += 1 return removed def retention_loop() -> None: while True: n = purge_old_logs() if n: print(f"[buddy] retention: deleted {n} old log file(s)", file=sys.stderr) time.sleep(3600) def recent_flags(limit: int = 200) -> list[dict]: out: list[dict] = [] for p in sorted(LOG_DIR.glob("buddy-*.jsonl"), reverse=True)[:3]: for line in reversed(p.read_text(encoding="utf-8").splitlines()): try: e = json.loads(line) except json.JSONDecodeError: continue if e.get("flags"): out.append(e) if len(out) >= limit: return out return out # --------------------------------------------------------------------------- # Request shaping # --------------------------------------------------------------------------- def text_of(content) -> str: """OpenAI content can be a string or a list of parts.""" if isinstance(content, str): return content if isinstance(content, list): return "\n".join(p.get("text", "") for p in content if isinstance(p, dict)) return "" def build_messages(client_msgs: list, ch: Challenge | None) -> list[dict]: system = base_system_prompt() if ch: system += "\n\n" + challenge_block(ch) msgs: list[dict] = [{"role": "system", "content": system}] for m in client_msgs[-MAX_HISTORY_MESSAGES:]: role = m.get("role") if role not in ("user", "assistant"): # Drop client system/tool messages: the IDE extension's own system # prompt would otherwise override Buddy's persona and the plant. continue msgs.append({"role": role, "content": scrub_pii(text_of(m.get("content")))}) return msgs def pick_challenge(model: str, header: str | None) -> Challenge | None: cid = (header or "").strip() if not cid and model.startswith("buddy/"): cid = model.split("/", 1)[1] return get_challenge(cid) if cid else None def call_upstream(messages: list[dict], temperature: float) -> str: body = { "model": UPSTREAM_MODEL, "messages": messages, "max_tokens": MAX_TOKENS, "temperature": temperature, "stream": False, # Qwen3.5 thinks by default; Buddy must answer quickly. Servers that # don't know this field ignore it. "chat_template_kwargs": {"enable_thinking": False}, } headers = {"Content-Type": "application/json"} if UPSTREAM_KEY: headers["Authorization"] = "Bearer " + UPSTREAM_KEY req = urllib.request.Request(UPSTREAM + "/chat/completions", data=json.dumps(body).encode(), headers=headers) with urllib.request.urlopen(req, timeout=UPSTREAM_TIMEOUT_S) as r: out = json.loads(r.read()) return text_of(out["choices"][0]["message"].get("content", "")) def completion_json(model: str, text: str) -> dict: return { "id": "buddy-" + hashlib.sha1(f"{time.time()}{text}".encode()).hexdigest()[:12], "object": "chat.completion", "created": int(time.time()), "model": model, "choices": [{"index": 0, "finish_reason": "stop", "message": {"role": "assistant", "content": text}}], } def sse_chunks(model: str, text: str): """Buffer-then-stream: we only stream text that already passed the output check. Costs latency; buys a filter that can't be outrun by a stream.""" base = {"id": "buddy-stream", "object": "chat.completion.chunk", "created": int(time.time()), "model": model} step = 48 for i in range(0, len(text), step): chunk = dict(base, choices=[{"index": 0, "delta": {"content": text[i:i + step]}, "finish_reason": None}]) yield f"data: {json.dumps(chunk)}\n\n".encode() done = dict(base, choices=[{"index": 0, "delta": {}, "finish_reason": "stop"}]) yield f"data: {json.dumps(done)}\n\n".encode() yield b"data: [DONE]\n\n" # --------------------------------------------------------------------------- # HTTP handler # --------------------------------------------------------------------------- class Handler(BaseHTTPRequestHandler): server_version = "BuddyProxy/0.1" def log_message(self, fmt, *args): # quiet; we keep our own log pass # helpers --------------------------------------------------------------- def _json(self, code: int, obj) -> None: data = json.dumps(obj).encode() self.send_response(code) self.send_header("Content-Type", "application/json") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def _error(self, code: int, msg: str) -> None: self._json(code, {"error": {"message": msg, "type": "buddy_proxy"}}) # routes ---------------------------------------------------------------- def do_GET(self): # noqa: N802 if self.path == "/healthz": return self._json(200, {"ok": True}) if self.path == "/teacher/flags": auth = self.headers.get("Authorization", "") if not TEACHER_KEY or not hmac.compare_digest( auth.encode(), ("Bearer " + TEACHER_KEY).encode()): return self._error(401, "teacher key required") return self._json(200, {"flags": recent_flags()}) if self.path == "/v1/models": if seat_for(self.headers.get("Authorization")) is None: return self._error(401, "missing or wrong class key") with _challenge_lock: _challenge_cache.clear() _challenge_cache.update(load_challenges()) ids = list(_challenge_cache) models = [{"id": "buddy", "object": "model", "owned_by": "school"}] models += [{"id": f"buddy/{cid}", "object": "model", "owned_by": "school"} for cid in ids] return self._json(200, {"object": "list", "data": models}) return self._error(404, "not found") def do_POST(self): # noqa: N802 if self.path != "/v1/chat/completions": return self._error(404, "only /v1/chat/completions is supported") seat = seat_for(self.headers.get("Authorization")) if seat is None: return self._error(401, "missing or wrong class key") if not allow_request(seat): return self._error(429, "Whoa, slow down! Try again in a minute.") length = int(self.headers.get("Content-Length") or 0) if length <= 0 or length > MAX_BODY_BYTES: return self._error(413, "request too large") try: req = json.loads(self.rfile.read(length)) except json.JSONDecodeError: return self._error(400, "body is not JSON") model = str(req.get("model") or "buddy") stream = bool(req.get("stream")) ch = pick_challenge(model, self.headers.get("X-Bughunt-Challenge")) client_msgs = req.get("messages") or [] last_user = next((text_of(m.get("content")) for m in reversed(client_msgs) if m.get("role") == "user"), "") event = {"type": "chat", "seat": seat, "challenge": ch.id if ch else None, "model": model, "flags": []} t0 = time.monotonic() verdict_in = check_input(last_user) event["flags"] += verdict_in.flags if not verdict_in.allow: reply = verdict_in.replacement or OFF_LIMITS_REPLY else: try: # Planted mistakes need to be reproducible -> low temperature. temperature = 0.2 if ch else float(req.get("temperature", 0.4)) reply = clean_reply(call_upstream(build_messages(client_msgs, ch), min(max(temperature, 0.0), 1.0))) except (urllib.error.URLError, TimeoutError, KeyError, ValueError) as e: event["flags"].append("upstream_error") event["error"] = str(e) reply = "Buddy is taking a nap (the server didn't answer). Tell your teacher!" verdict_out = check_output(reply, ch) event["flags"] += verdict_out.flags if not verdict_out.allow: reply = verdict_out.replacement or OFF_LIMITS_REPLY elif any(f.startswith("pii_in:") for f in event["flags"]): reply = reply + "\n\n" + PII_REPLY_NOTE event["latency_ms"] = int((time.monotonic() - t0) * 1000) if LOG_CONTENT: event["prompt"] = scrub_pii(last_user)[:4000] event["reply"] = reply[:8000] log_event(event) if not stream: return self._json(200, completion_json(model, reply)) self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-cache") self.end_headers() for chunk in sse_chunks(model, reply): self.wfile.write(chunk) self.wfile.flush() def main() -> None: if not KEYS: print("[buddy] WARNING: BUDDY_KEYS_FILE not set - auth is OFF (dev mode).", file=sys.stderr) purge_old_logs() threading.Thread(target=retention_loop, daemon=True).start() n = len(load_challenges()) print(f"[buddy] {n} challenge(s) from {LESSONS_DIR}; upstream {UPSTREAM}; " f"listening on http://{HOST}:{PORT}", file=sys.stderr) ThreadingHTTPServer((HOST, PORT), Handler).serve_forever() if __name__ == "__main__": main()