"""server.py -- the page, plus the buttons (Ivy, Oct 23 2026: "There should be a play button that lets me start a new creation when it's not busy. And there should be co op mode so it can play with me.") python3 server.py # http://localhost:8767/viewer.html (127.0.0.1 only) Serves reef/ as before, and: GET /api/status {busy, run, stage, coop} GET /api/catalog?q=frog search the whole LDraw catalog POST /api/bin_add {part} put a catalog piece in its bin for good POST /api/play start a solo creation (refused while something is building) POST /api/coop start a creation you play together, taking turns POST /api/stop end the game this server started POST /api/ghost {card, p} legal placements of a card nearest point p (our frame), for the page's ghost POST /api/pictionary {name} Pin gets a secret and builds it; you guess (Ivy, Oct 23 2026) POST /api/guess {text} your guess in pictionary -> {right} POST /api/giveup pictionary: show the answer POST /api/act {type, ...} your move in co-op: card {card} | place {i} | place_at {card, conn, idx, spin} | repaint {j, colour} | swap {card} | back | pass | done One game at a time. A game started here runs in this process; a build started from the command line still shows up as busy. Written by Iris.""" import json, os, sys, threading, time, queue, random from http.server import ThreadingHTTPServer, SimpleHTTPRequestHandler HERE = os.path.dirname(os.path.abspath(__file__)); sys.path.insert(0, HERE) import imagine as I GAME = {'thread': None, 'reef': None, 'guesses': [], 'secret': None} LOCK = threading.Lock() # PICTIONARY secrets (Ivy, Oct 23 2026: "it gets a prompt and it has to make me guess it, and my guesses give it feedback like # pictionary"): things a kid could name, that a handful of bricks can say WORDS = ['giraffe', 'snail', 'rocket', 'lighthouse', 'duck', 'palm tree', 'robot', 'snowman', 'castle', 'crab', 'penguin', 'ice cream cone', 'sailboat', 'car', 'airplane', 'house', 'flower', 'mushroom', 'dog', 'fish', 'whale', 'dinosaur', 'spider', 'bridge', 'train', 'helicopter', 'umbrella', 'chair', 'lamp', 'guitar', 'cactus', 'volcano', 'ladybug', 'butterfly', 'owl', 'frog', 'turtle', 'elephant', 'camel', 'windmill', 'tent', 'campfire', 'birthday cake', 'crown', 'sword', 'key', 'teapot', 'mug', 'tower', 'satellite', 'snake', 'bee', 'pig', 'tractor', 'bed', 'flamingo', 'octopus', 'tree', 'bird', 'boat'] def _norm(s): w = [x.rstrip('s') if len(x) > 3 else x for x in ''.join(c if c.isalnum() else ' ' for c in str(s).lower()).split()] return [x for x in w if x not in ('a', 'an', 'the', 'it', 'is', 'its', 'thats', 'that', 'i', 'think', 'maybe', 'some', 'kind', 'of')] def right_guess(guess, secret): """plain match first (every word of the secret in the guess, plurals and 'a'/'the' ignored); a near miss ('boat' for 'sailboat', 'puppy' for 'dog') goes to Haiku, and only a clear yes counts""" g, sec = _norm(guess), _norm(secret) if sec and all(w in g for w in sec): return True if not g: return False head = sec[-1] if sec else '' # the thing itself: palm TREE, birthday CAKE, ice cream CONE if len(g) == 1 and (g[0] == head or (head.endswith(g[0]) and len(g[0]) >= 4 and 2 * len(g[0]) >= len(head))): return True # 'tree' for palm tree, 'boat' for sailboat -- not 'ant' for elephant, not 'cream' # meaning, measured (Ivy, Oct 23 2026: "Can we use that thing that looks for semantic similarity?"). MiniLM, at 0.72: on 25 true # synonyms and 29 related-but-wrong guesses it caught 14 synonyms (puppy/dog, plane/airplane, serpent/snake, cup/mug...) and # accepted NO wrong ones; the worst wrong pair was shark/whale at 0.69. Qwen3-Embedding-0.6B, newer and 70.5 on MTEB, did # worse here (cat/dog 0.86 without an instruction; 11 caught + 1 wrong with one) -- retrieval embedders measure relatedness. try: import teach, numpy as np E = teach.emb([' '.join(g), ' '.join(sec)]); E = E / np.linalg.norm(E, axis=1, keepdims=True) if float(E[0] @ E[1]) >= 0.72: return True except Exception: pass sys.path.insert(0, os.path.join(os.path.dirname(HERE), 'favorites')); import haiku as H out = {} def ask(): try: t, _ = H.ask(f'Pictionary. The answer is "{secret}". Someone guessed "{guess}". Would a fair Pictionary judge accept that guess as ' f'naming the answer (same thing, or a plain synonym; NOT just something similar or related)? Answer only YES or NO.', [], 5) out['yes'] = str(t).strip().upper().startswith('YES') except Exception: pass th = threading.Thread(target=ask, daemon=True); th.start(); th.join(8) # Haiku retries can take minutes; a guess can't return out.get('yes', None) # None = couldn't check (counts as not right) def reveal(r, how): st = {'secret': GAME['secret'], 'turn': getattr(r, 'turn', 0), 'guesses': len(GAME['guesses']), 'how': how} r.solved = st; r.state.update(solved=st, request=GAME['secret']); r.save() def latest_state(): try: run = json.load(open(os.path.join(HERE, 'runs', 'latest.json')))['run'] return run, json.load(open(os.path.join(HERE, 'runs', run, 'state.json'))) except (OSError, ValueError, KeyError): return None, {} def busy(): if GAME['thread'] and GAME['thread'].is_alive() and not getattr(GAME['reef'], 'stop', False): return True # a stopped game is only tidying up run, st = latest_state() # a build started from the command line if not st: return False fresh = time.time() - os.path.getmtime(os.path.join(HERE, 'runs', run, 'state.json')) < 600 return st.get('stage') not in ('done', 'stopped') and fresh def start(coop, player='Ivy', request=None, secret=None): # load the newest engine code for every new game (Oct 23 2026: Ivy kept starting games back to back, so a 'restart when idle' # never found a gap and she played an hour on old code). Server changes themselves still need a restart. import importlib, to_glb, play for mod in (to_glb, play, I): importlib.reload(mod) name = ('coop_' if coop else 'play_') + time.strftime('%m%d_%H%M%S'); seed = random.randrange(10**6) # a start-over that never got to start (Oct 23 2026: the server was on old code and dropped the flamingo's do-over without a # word): asking for the same thing again picks its plan back up instead of starting cold pending = None run_, st_ = latest_state() if not coop and request and st_.get('restarted') and (st_.get('request') or '').strip().lower() == request.strip().lower(): so = st_['restarted'] if so.get('first_piece') and so.get('first_colour'): pending = {'piece': so['first_piece'], 'colour': so['first_colour'], 'why': f"starting it over: {so.get('would', '')}", 'idea': (st_.get('history') or [{}])[-1].get('might_be') or request, 'would': so.get('would', ''), 'restart': True} def career(): first, budget, k = pending, 100000, 0 # no turn limit: Pin decides when it's done (Ivy: "Make it clear Pin can't run out of turns"); Stop still ends it while budget > 0: again = bool(first and first.get('restart')) # a start-over keeps the request it was building r = I.Imagine(seed + k, os.path.join(HERE, 'runs', name + (f'_{k}' if k else '')), first=first, request=request if (k == 0 or again) else None, secret=secret) r.coop = coop; r.player = (player or 'Ivy').strip()[:24] or 'Ivy'; r.inbox = queue.Queue(); GAME['reef'] = r if secret: r.guesses = GAME['guesses']; r.d20 = GAME.get('d20') # one list across start-overs: the guesses are about the secret first = r.play(budget); budget -= len(r.notebook); k += 1 if not first or coop or (request and not first.get('restart')) or getattr(r, 'solved', None): break # only clear the slot if it is still OURS: a stopped game tidying up used to clear the NEW game's slot, so co-op lost # its ghosts and moves (Ivy, Oct 23 2026: "It's not letting me place a piece in co op. No ghost") if GAME['reef'] is r: GAME['reef'] = None t = threading.Thread(target=career, daemon=True); GAME['thread'] = t; t.start(); return name class H(SimpleHTTPRequestHandler): def __init__(s, *a, **k): super().__init__(*a, directory=HERE, **k) def log_message(s, *a): pass def send_json(s, obj, code=200): b = json.dumps(obj).encode(); s.send_response(code); s.send_header('Content-Type', 'application/json') s.send_header('Content-Length', str(len(b))); s.end_headers(); s.wfile.write(b) def do_GET(s): if s.path.startswith('/api/catalog'): # Ivy searches the whole LDraw catalog for pieces to give it from urllib.parse import urlparse, parse_qs import partbin qs = parse_qs(urlparse(s.path).query); q = (qs.get('q') or [''])[0]; off = int((qs.get('offset') or ['0'])[0] or 0) try: if q.strip(): res = partbin.catalog(q, 13, off) else: # an empty search shows what's already in Pin's bin, newest first import play; names = {p: n for p, n, _ in partbin.index()} allp = list(reversed(play.PALETTE)); res = [{'part': p, 'name': names.get(p, play.CX.NAMES.get(p, p)), 'in_bin': True} for p in allp[off:off + 13]] return s.send_json({'ok': True, 'results': res[:12], 'more': len(res) > 12}) except Exception as e: return s.send_json({'ok': False, 'why': str(e)[:200], 'results': []}) if s.path.startswith('/api/preview'): # a picture of any catalog part, drawn once and kept from urllib.parse import urlparse, parse_qs import partbin p = partbin.preview((parse_qs(urlparse(s.path).query).get('part') or [''])[0]) if not p: s.send_response(404); s.end_headers(); return b = open(p, 'rb').read(); s.send_response(200); s.send_header('Content-Type', 'image/png'); s.send_header('Content-Length', str(len(b))) s.send_header('Cache-Control', 'max-age=86400'); s.end_headers(); s.wfile.write(b); return if s.path.startswith('/api/status'): run, st = latest_state() try: nm = json.load(open(os.path.join(HERE, 'self.json'))).get('name') or 'it' except (OSError, ValueError): nm = 'it' return s.send_json({'busy': busy(), 'run': run, 'stage': st.get('stage'), 'coop': bool(GAME['reef'] and GAME['reef'].coop), 'name': nm}) return super().do_GET() def do_POST(s): n = int(s.headers.get('Content-Length') or 0); body = json.loads(s.rfile.read(n) or b'{}') with LOCK: if s.path in ('/api/play', '/api/coop'): if busy(): return s.send_json({'ok': False, 'why': 'it is busy building right now'}, 409) return s.send_json({'ok': True, 'run': start(coop=s.path == '/api/coop', player=body.get('name'), request=body.get('request'))}) if s.path == '/api/pictionary': if busy(): return s.send_json({'ok': False, 'why': 'it is busy building right now'}, 409) import nouns # a random noun; a d20 decides how common (Ivy: "Barring critical failure") GAME['d20'], GAME['secret'], _ = nouns.roll(); GAME['guesses'] = [] return s.send_json({'ok': True, 'run': start(coop=False, player=body.get('name'), secret=GAME['secret'])}) if s.path in ('/api/guess', '/api/giveup'): r = GAME['reef'] if not (r and getattr(r, 'pictionary', False)): return s.send_json({'ok': False, 'why': 'no pictionary game is running'}, 409) if r.solved: return s.send_json({'ok': True, 'right': True, 'secret': GAME['secret']}) if s.path == '/api/giveup': reveal(r, 'gave up'); return s.send_json({'ok': True, 'secret': GAME['secret']}) text = str(body.get('text', '')).strip()[:80] if not text: return s.send_json({'ok': False}) g = {'text': text, 'turn': getattr(r, 'turn', 0), 'right': None}; GAME['guesses'].append(g) r.state['guesses'] = list(GAME['guesses']); r.save() def judge(g=g, r=r): # outside the lock: Haiku can take a few seconds # warmth: how close in MEANING (Ivy, Oct 23 2026: "It should give a warmth indicator to each guess that stays # visible"); related counts here -- cat for dog is warm even though it isn't right. The page shows each guess # against the one before: red up / blue down a little, fire / shiver for a big jump. Pin can override it. try: import teach, numpy as np E = teach.emb([g['text'], GAME['secret']]); E = E / np.linalg.norm(E, axis=1, keepdims=True) g['warm'] = round(float(E[0] @ E[1]), 3) except Exception: pass r.state['guesses'] = list(GAME['guesses']); r.save() ok = right_guess(g['text'], GAME['secret']); g['right'] = bool(ok) if ok is None: g['unchecked'] = True if g['right'] and not r.solved: reveal(r, 'guessed') r.state['guesses'] = list(GAME['guesses']); r.save() threading.Thread(target=judge, daemon=True).start() return s.send_json({'ok': True}) if s.path == '/api/stop': # INSTANT (Ivy: "Stop button is waiting for its turn to finish. It should be instant.") r = GAME['reef'] if r: r.stop = True; r.inbox.put({'type': 'done'}) p = getattr(r, '_proc', None) if p and p.poll() is None: p.kill() # its look, mid-thought try: r.state.update(stage='stopped', error='stopped'); r.save() except Exception: pass return s.send_json({'ok': bool(r)}) if s.path == '/api/bin_add': # put a catalog piece in its bin, for good import partbin ok, msg = partbin.add(str(body.get('part', '')), (body.get('name') or 'Ivy').strip()[:24] or 'Ivy', describe_it=False) if ok: threading.Thread(target=partbin.describe, args=(str(body.get('part', '')).strip().lower(),), daemon=True).start() r = GAME['reef'] if ok and r is not None and getattr(r, '_deck_w', None) is not None: r._deck_w = None # its deck includes it from now on return s.send_json({'ok': ok, 'why': msg}) if s.path == '/api/ghost': # legal placements of a card near the mouse (point in our frame) r = GAME['reef'] if not (r and r.coop and r.state.get('stage') in ('your_turn', 'your_options')): return s.send_json({'ok': False, 'ghosts': []}) try: return s.send_json({'ok': True, **r.ghosts(body['card'], body['p'], under=bool(body.get('under')), node=body.get('node'))}) except Exception as e: return s.send_json({'ok': False, 'why': str(e)[:200], 'ghosts': []}) if s.path == '/api/act': r = GAME['reef'] if not (r and r.coop): return s.send_json({'ok': False, 'why': 'no co-op game is running'}, 409) r.inbox.put(body); return s.send_json({'ok': True}) s.send_json({'ok': False, 'why': 'unknown'}, 404) if __name__ == '__main__': import play # names for the cards on the page # anything that pulls in cypari (via snappy) must be imported HERE, on the main thread: it installs signal handlers, and the # first catalog search from a request thread failed with "signal only works in main thread" import partbin; partbin.index() os.environ.setdefault('HIP_VISIBLE_DEVICES', ''); os.environ.setdefault('CUDA_VISIBLE_DEVICES', '') # CPU only (the iGPU aborts) try: import sentence_transformers # -> datasets -> fsspec -> snappy -> cypari (signal handlers): main thread only except Exception as e: print('sentence_transformers import failed:', e, flush=True) try: sys.path.insert(0, os.path.join(HERE, 'teacher')); import teach except Exception as e: print('teacher import failed:', e, flush=True) json.dump({n: play.CX.NAMES.get(n, n) for n in play.PALETTE}, open(os.path.join(HERE, 'parts', 'names.json'), 'w')) # warm every piece's connectors in the background (~20 s once), so nobody pays a piece's first-time cost mid-game def warm(): for n in play.PALETTE + list(play.BASEPLATES): play.G.fresh(n) try: sys.path.insert(0, os.path.join(HERE, 'teacher')); import teach; teach.emb(['warm up']) # the teacher's language model except Exception as e: print('teacher warmup failed:', e, flush=True) threading.Thread(target=warm, daemon=True).start() ThreadingHTTPServer(('127.0.0.1', 8767), H).serve_forever()