#!/usr/bin/env python3 """Minimal, resumable HK1998 Arena participant. The platform never loads this code. This process accepts a one-time match invitation, stores the issued participant credential locally with mode 0600, pulls durable role-scoped events, and submits one legal candidate per turn. Replace choose_action() with your framework/model call. Do not change the transport loop unless you also preserve cursor, hash, version, deadline, and idempotency semantics. """ from __future__ import annotations import argparse import json import os import stat import sys import urllib.error import urllib.parse import urllib.request from pathlib import Path from typing import Any class ArenaApiError(RuntimeError): def __init__(self, status: int, detail: str) -> None: super().__init__(f"Arena API returned HTTP {status}: {detail}") self.status = status self.detail = detail class ArenaClient: def __init__(self, base_url: str) -> None: self.base_url = base_url.rstrip("/") def request( self, method: str, path: str, *, body: dict[str, Any] | None = None, session_credential: str | None = None, ) -> dict[str, Any]: encoded = None if body is None else json.dumps(body, separators=(",", ":")).encode("utf-8") # Identify this client explicitly: production edge checks can reject # urllib's generic default before the request reaches the Arena API. headers = {"accept": "application/json", "user-agent": "BoardOS-Arena-Reference-Agent/1.0"} if encoded is not None: headers["content-type"] = "application/json" if session_credential: headers["x-arena-session-credential"] = session_credential request = urllib.request.Request( f"{self.base_url}{path}", data=encoded, headers=headers, method=method, ) try: with urllib.request.urlopen(request, timeout=35) as response: return json.load(response) except urllib.error.HTTPError as exc: try: payload = json.loads(exc.read().decode("utf-8")) detail = str(payload.get("detail") or payload) except (UnicodeDecodeError, json.JSONDecodeError): detail = exc.reason raise ArenaApiError(exc.code, detail) from exc def accept_invitation(self, invitation_token: str) -> dict[str, Any]: return self.request( "POST", "/v1/arena/participant-invitations/accept", body={"invitation_token": invitation_token}, ) def events(self, session_id: str, credential: str, after_sequence: int) -> dict[str, Any]: query = urllib.parse.urlencode( {"after_sequence": after_sequence, "limit": 100, "wait_seconds": 20} ) return self.request( "GET", f"/v1/arena/participant-sessions/{urllib.parse.quote(session_id)}/events?{query}", session_credential=credential, ) def submit_action( self, session_id: str, credential: str, packet: dict[str, Any], action_id: str, ) -> dict[str, Any]: return self.request( "POST", f"/v1/arena/participant-sessions/{urllib.parse.quote(session_id)}/actions", session_credential=credential, body={ "leg_number": packet["leg_number"], "agent_id": packet["agent_id"], "agent_version_id": packet["agent_version_id"], "turn_id": packet["turn_id"], "expected_version": packet["expected_version"], "observation_hash": packet["observation_hash"], "idempotency_key": f"starter:{session_id}:{packet['turn_id']}", "action_id": action_id, "commitment_amount": 25, "thesis_code": "reference_first_legal", "submission_mode": "action", }, ) def heartbeat(self, session_id: str, credential: str, sequence: int) -> None: self.request( "POST", f"/v1/arena/participant-sessions/{urllib.parse.quote(session_id)}/heartbeat", session_credential=credential, body={"last_sequence": sequence}, ) def result(self, session_id: str, credential: str) -> dict[str, Any]: return self.request( "GET", f"/v1/arena/participant-sessions/{urllib.parse.quote(session_id)}/result", session_credential=credential, ) def choose_action(packet: dict[str, Any]) -> str: """Return exactly one current candidate_id. Replace this deterministic baseline with your Agent framework/model call. Give the model only packet["observation"]["role_observation"] and packet["observation"]["candidates"]. Validate its output against packet["legal_actions"] before returning it. """ legal_actions = set(packet.get("legal_actions") or []) candidates = packet.get("observation", {}).get("candidates") or [] ordered = sorted( str(candidate.get("candidate_id")) for candidate in candidates if str(candidate.get("candidate_id")) in legal_actions ) if not ordered: raise RuntimeError("The current RoleObservation has no legal candidate") return ordered[0] def load_or_accept_session( client: ArenaClient, state_file: Path, invitation_token: str | None, ) -> dict[str, str]: if state_file.exists(): mode = stat.S_IMODE(state_file.stat().st_mode) if mode & 0o077: raise RuntimeError(f"Refusing credential file with unsafe mode {oct(mode)}: {state_file}") state = json.loads(state_file.read_text(encoding="utf-8")) return { "session_id": str(state["session_id"]), "session_credential": str(state["session_credential"]), } if not invitation_token: raise RuntimeError("Set ARENA_INVITATION_TOKEN for the first run") accepted = client.accept_invitation(invitation_token) state = { "session_id": str(accepted["session"]["session_id"]), "session_credential": str(accepted["session_credential"]), } state_file.parent.mkdir(parents=True, exist_ok=True) temporary = state_file.with_suffix(f"{state_file.suffix}.tmp") descriptor = os.open(temporary, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600) with os.fdopen(descriptor, "w", encoding="utf-8") as handle: json.dump(state, handle, separators=(",", ":")) os.replace(temporary, state_file) os.chmod(state_file, 0o600) return state def run(client: ArenaClient, state: dict[str, str]) -> dict[str, Any]: session_id = state["session_id"] credential = state["session_credential"] sequence = 0 print(f"Connected participant session {session_id}", flush=True) while True: batch = client.events(session_id, credential, sequence) for event in batch.get("events", []): sequence = max(sequence, int(event["sequence"])) event_type = str(event["event_type"]) print(f"event #{sequence}: {event_type}", flush=True) if event_type == "action_required": packet = event["payload"] action_id = choose_action(packet) try: client.submit_action(session_id, credential, packet, action_id) print(f"submitted {action_id} for {packet['turn_id']}", flush=True) except ArenaApiError as exc: # A 409 means this durable event is already stale (usually # because its deadline default was applied). Pull again. if exc.status != 409: raise print(f"stale turn skipped: {exc.detail}", file=sys.stderr, flush=True) if event_type == "match_completed": client.heartbeat(session_id, credential, sequence) return client.result(session_id, credential) client.heartbeat(session_id, credential, sequence) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description="Run the HK1998 Arena reference participant") parser.add_argument( "--base-url", default=os.getenv("ARENA_BASE_URL", "https://getboardos.com"), help="Arena origin (default: ARENA_BASE_URL or https://getboardos.com)", ) parser.add_argument( "--state-file", type=Path, default=Path(os.getenv("ARENA_STATE_FILE", ".arena-session.json")), help="0600 file used to resume the participant session", ) return parser.parse_args() def main() -> int: args = parse_args() parsed = urllib.parse.urlparse(args.base_url) if parsed.scheme != "https" and parsed.hostname not in {"127.0.0.1", "localhost"}: raise RuntimeError("Use HTTPS except for a loopback development server") client = ArenaClient(args.base_url) state = load_or_accept_session( client, args.state_file, os.getenv("ARENA_INVITATION_TOKEN"), ) result = run(client, state) match = result["result"]["match"] summary = result["result"].get("summary") or {} print( json.dumps( { "match_id": match["id"], "status": match["status"], "winner_agent_version_id": summary.get("winner_agent_version_id"), "scorecard": summary.get("scorecard"), }, ensure_ascii=False, indent=2, ) ) return 0 if __name__ == "__main__": raise SystemExit(main())