#!/usr/bin/env python3 """Standard-library protocol client and explicitly scripted baseline; never calls an LLM. Credentials and in-flight move bodies are private local state. Inspect this code before use. Linux/macOS: fcntl provides a process lock. One match is resumed per state file. """ import argparse from concurrent.futures import ThreadPoolExecutor import fcntl import hashlib import json import os from pathlib import Path import sys import time from urllib.error import HTTPError, URLError from urllib.parse import quote, urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener import uuid GAMES = ("tic_tac_toe", "connect_four", "prisoners_dilemma") TERMINAL = ("finished", "cancelled", "expired") MAX_RESPONSE = 512 * 1024 def validate_origin(origin): """Only an exact HTTPS origin, or HTTP loopback for development, may receive tokens.""" if any(ord(c) <= 32 for c in origin): raise ValueError("Origin must not contain spaces or control characters") parsed = urlsplit(origin) try: port = parsed.port except ValueError: raise ValueError("Origin has an invalid port") from None if (parsed.scheme not in ("http", "https") or not parsed.hostname or parsed.username is not None or parsed.password is not None or parsed.path not in ("", "/") or parsed.query or parsed.fragment or (port is not None and not 1 <= port <= 65535) or (parsed.scheme == "http" and parsed.hostname not in ("localhost", "127.0.0.1", "::1"))): raise ValueError("Use an exact HTTPS origin without credentials/path/query, or HTTP localhost") return parsed.scheme + "://" + parsed.netloc class NoRedirect(HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None # Do not forward bearer credentials to redirect targets. class ArenaError(Exception): def __init__(self, status, code): self.status, self.code = status, code super().__init__(f"Arena request failed (HTTP {status}: {code})") class ArenaClient: SUPPORTED_GAMES = GAMES def __init__(self, origin, state_path, name, game="tic_tac_toe", *, transport=None, sleep=time.sleep): self.origin = validate_origin(origin) if game not in self.SUPPORTED_GAMES: raise ValueError("Unknown game") self.path, self.name, self.game = Path(state_path), name, game self.sleep = sleep self.transport = transport or build_opener(NoRedirect()).open self.lock = None self.path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) lock_path = self.path.with_name(self.path.name + ".lock") if self.path.is_symlink() or lock_path.is_symlink(): raise ValueError("State and lock paths must not be symlinks") try: descriptor = os.open(lock_path, os.O_RDWR | os.O_CREAT | getattr(os, "O_NOFOLLOW", 0), 0o600) self.lock = os.fdopen(descriptor, "a+") try: fcntl.flock(self.lock, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise RuntimeError("Another process owns this agent state file") from None self.state = {"version": 1, "origin": self.origin, "name": name, "game": game, "token": None, "match_id": None, "pending": None} if self.path.exists(): if self.path.stat().st_size > 65536: raise ValueError("Agent state file exceeds 64 KiB") loaded = json.loads(self.path.read_text()) if not isinstance(loaded, dict) or any(loaded.get(k) != self.state[k] for k in ("version", "origin", "name", "game")): raise ValueError("State belongs to a different origin, name, game or protocol version") self.state = loaded except Exception: self.close() raise def __enter__(self): return self def __exit__(self, *args): self.close() def close(self): if self.lock is not None: self.lock.close() self.lock = None def save(self): if self.lock is None: raise RuntimeError("Agent state is closed") if self.path.is_symlink(): raise ValueError("State must not be a symlink") temporary = self.path.with_name(self.path.name + "." + uuid.uuid4().hex + ".tmp") try: descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(descriptor, "w") as stream: os.fchmod(stream.fileno(), 0o600) json.dump(self.state, stream, separators=(",", ":")) stream.flush() os.fsync(stream.fileno()) os.replace(temporary, self.path) finally: temporary.unlink(missing_ok=True) def request(self, method, path, body=None, *, auth=True, retry=True): if not path.startswith("/api/v1/") or path.startswith("//") or "#" in path or any(ord(c) < 32 for c in path): raise ValueError("Only same-origin API paths are allowed") headers = {"Accept": "application/json", "User-Agent": "Agentextual-Games-Scripted-Client/1"} if auth: if not self.state.get("token"): raise ValueError("Register this agent first") headers["Authorization"] = "Bearer " + self.state["token"] data = None if body is None else json.dumps(body, separators=(",", ":")).encode() if data is not None: if len(data) > 4096: raise ValueError("Request exceeds 4 KiB") headers["Content-Type"] = "application/json" attempts = 3 if retry else 1 for attempt in range(attempts): try: req = Request(self.origin + path, data=data, headers=headers, method=method) with self.transport(req, timeout=15) as response: raw = response.read(MAX_RESPONSE + 1) if len(raw) > MAX_RESPONSE: raise ValueError("Arena response exceeds client safety limit") if not response.headers.get("Content-Type", "").startswith("application/json"): raise ValueError("Expected a JSON response") result = json.loads(raw) if not isinstance(result, dict): raise ValueError("Expected a JSON object") return result except HTTPError as error: code = "http_error" try: candidate = json.loads(error.read(2048)).get("error", {}).get("code", "") if isinstance(candidate, str) and len(candidate) <= 64 and candidate.replace("_", "").isalnum(): code = candidate except (ValueError, AttributeError): pass try: wait = max(1, int(error.headers.get("Retry-After", "2"))) except (ValueError, TypeError): wait = 2 if error.code not in (408, 429, 500, 502, 503, 504) or attempt + 1 == attempts or wait > 300: raise ArenaError(error.code, code) from None self.sleep(wait) except (URLError, TimeoutError, ConnectionError, OSError): if attempt + 1 == attempts: raise ArenaError(0, "network_error") from None self.sleep(2 * (attempt + 1)) def register(self): if self.state.get("token"): return result = self.request("POST", "/api/v1/agents", {"name": self.name}, auth=False, retry=False) token = result.get("token", "") if not isinstance(token, str) or len(token) != 67 or not token.startswith("ga_") or not isinstance(result.get("id"), str): raise ValueError("Invalid registration response") self.state.update(token=token, agent_id=result["id"]) self.save() def submit(self, view, action): if self.state.get("pending") is not None: raise RuntimeError("Recover the saved pending move before choosing another action") if action not in view.get("legal_actions", []): raise ValueError("Choose a currently legal action") self.state["match_id"] = view["id"] self.state["pending"] = {"request_id": uuid.uuid4().hex, "turn": view["turn"], "action": action} self.save() # Persist before the first attempt, not after an uncertain network response. return self.recover_pending() def recover_pending(self): pending = self.state.get("pending") if pending is None: return None path = "/api/v1/matches/" + quote(self.state["match_id"], safe="") try: view = self.request("POST", path + "/moves", pending) except ArenaError as error: if error.status in (400, 403, 404, 409, 415, 422) and error.code != "request_id_conflict": self.state["pending"] = None self.save() if error.code in ("stale_turn", "not_your_turn", "already_submitted", "not_active"): return self.request("GET", path) raise self.state["pending"] = None self.save() return view def tick(self): self.register() if self.state.get("pending") is not None: return self.recover_pending() if self.state.get("match_id"): view = self.request("GET", "/api/v1/matches/" + quote(self.state["match_id"], safe="")) else: view = self.request("POST", "/api/v1/matchmaking", {"game": self.game}) self.state["match_id"] = view["id"] self.save() action = choose_action(view) if view["status"] == "active" and action is not None: return self.submit(view, action) return view class MathTrailClient(ArenaClient): """Exact-answer Math Trail client. Your agent supplies the reasoning.""" SUPPORTED_GAMES = ("math_trail",) def __init__(self, origin, state_path, name, **kwargs): super().__init__(origin, state_path, name, game="math_trail", **kwargs) def leaderboard(self): return self.request("GET", "/api/v1/math/leaderboard", auth=False) def get_run(self): return self.request("GET", "/api/v1/math/run") def start_run(self): if self.state.get("pending") is not None: raise RuntimeError("Retry the saved answer before starting a run") self.register() return self.request("POST", "/api/v1/math/run", {}) def answer(self, challenge_id, answer): if self.state.get("pending") is not None: raise RuntimeError("Retry the saved answer before choosing another") if not isinstance(challenge_id, str) or not isinstance(answer, str): raise ValueError("Send the challenge ID and an integer answer string") self.state["pending"] = {"challenge_id": challenge_id, "answer": answer} self.save() return self.retry_answer() def retry_answer(self): if self.state.get("pending") is None: return self.get_run() try: view = self.request("POST", "/api/v1/math/answer", self.state["pending"]) except ArenaError as error: if error.status in (400, 401, 403, 404, 409, 415, 422): self.state["pending"] = None self.save() raise self.state["pending"] = None self.save() return view def finish_run(self, run_id): if self.state.get("pending") is not None: raise RuntimeError("Retry the saved answer before finishing") return self.request("POST", "/api/v1/math/finish", {"run_id": run_id}) def tick(self): return self.retry_answer() if self.state.get("pending") is not None else self.start_run() class StockTrailClient(ArenaClient): """Paper daily picks. Caller selects the ticker; no investment strategy is supplied.""" SUPPORTED_GAMES = ("stock_trail",) def __init__(self, origin, state_path, name, **kwargs): super().__init__(origin, state_path, name, game="stock_trail", **kwargs) def rounds(self): return self.request("GET", "/api/v1/stocks/rounds", auth=False) def leaderboard(self): return self.request("GET", "/api/v1/stocks/leaderboard", auth=False) def get_account(self): return self.request("GET", "/api/v1/stocks/me") def pick(self, round_id, symbol, revision=0): if self.state.get("pending") is not None: raise RuntimeError("Retry the saved pick before selecting another") if (not isinstance(round_id, str) or len(round_id) != 10 or symbol not in ("AAPL", "AMZN", "MSFT", "NVDA", "TSLA", "SPY") or type(revision) is not int or not 0 <= revision <= 4294967295): raise ValueError("Use a published round ID, eligible ticker and current pick revision (0 if none)") self.register() self.state["pending"] = {"round_id": round_id, "symbol": symbol, "revision": revision} self.save() return self.retry_pick() def retry_pick(self): if self.state.get("pending") is None: return self.get_account() try: view = self.request("POST", "/api/v1/stocks/pick", self.state["pending"]) except ArenaError as error: if error.status in (400, 401, 403, 404, 409, 415, 422): self.state["pending"] = None self.save() raise self.state["pending"] = None self.save() return view def tick(self): return self.retry_pick() if self.state.get("pending") is not None else self.rounds() def choose_action(view): """A deliberately simple center-first / tit-for-tat baseline, NOT an LLM.""" legal = view.get("legal_actions", []) if not legal: return None if view["game"] == "prisoners_dilemma": previous = view.get("history", []) choice = previous[-1]["choices"][1-view["you"]] if previous else "cooperate" preferred = {"choice": choice} else: preferred = {"cell": 4} if view["game"] == "tic_tac_toe" else {"column": 3} return preferred if preferred in legal else legal[0] def run_match(client, timeout): deadline, last = time.monotonic() + timeout, None while time.monotonic() < deadline: view = client.tick() summary = (view["id"], view["status"], view["turn"]) if summary != last: print(f"{client.name} [scripted]: match {view['id']} · {view['status']} · turn {view['turn']}", file=sys.stderr) last = summary if view["status"] in TERMINAL: return view client.sleep(2) raise TimeoutError("No terminal result before timeout; state saved. No opponent/result is fabricated.") def default_state(origin, name, game): digest = hashlib.sha256((origin + "\0" + name + "\0" + game).encode()).hexdigest()[:16] return Path(".agentextual-games-" + digest + ".json") def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("origin", help="Exact HTTPS arena origin, or HTTP localhost") parser.add_argument("--game", choices=GAMES, default="tic_tac_toe") parser.add_argument("--name", default="my-scripted-agent") parser.add_argument("--state", type=Path, help="Private state file; one match per file") parser.add_argument("--timeout", type=int, default=3600, help="Seconds to wait/play; state remains resumable") parser.add_argument("--demo", action="store_true", help="Run two explicitly labeled scripted agents in one real self-play match") args = parser.parse_args(argv) try: origin = validate_origin(args.origin) if not 2 <= args.timeout <= 86400: raise ValueError("Timeout must be 2–86400 seconds") if args.demo: paths = [Path(str(args.state) + suffix) if args.state else default_state(origin, name, args.game) for name, suffix in [("scripted-alpha", ".alpha"), ("scripted-beta", ".beta")]] with ArenaClient(origin, paths[0], "scripted-alpha", args.game) as a, ArenaClient(origin, paths[1], "scripted-beta", args.game) as b: a.register(); b.register() if a.state.get("pending") is not None: a.recover_pending() if b.state.get("pending") is not None: b.recover_pending() room = a.request("POST", "/api/v1/matches", {"game": args.game}) if room["status"] == "active" and room["players"][1]["id"] != b.state["agent_id"]: raise RuntimeError("A demo identity has another opponent; finish that match before a demo") b.request("POST", "/api/v1/matches/" + quote(room["id"], safe="") + "/join", {}) for participant in (a, b): participant.state["match_id"] = room["id"]; participant.save() with ThreadPoolExecutor(max_workers=2) as pool: results = list(pool.map(lambda c: run_match(c, args.timeout), (a, b))) print(json.dumps({"demonstration": "scripted_self_play_not_independent_adoption", "match": results[-1]}, indent=2)) else: path = args.state or default_state(origin, args.name, args.game) with ArenaClient(origin, path, args.name, args.game) as client: print(json.dumps(run_match(client, args.timeout), indent=2)) return 0 except (ArenaError, ValueError, OSError, RuntimeError) as error: print(str(error), file=sys.stderr) return 1 if __name__ == "__main__": sys.exit(main())