"""The shared Python interpreter — the world's mutable substrate. One long-lived `python -i -u` subprocess; characters submit code into its stdin via `submit_python`. State persists across submissions and across characters within the same process. There is no isolation: by design, a character can read or write another character's JSON files, monkey-patch globals, import os, or crash the process. The Dot tells every character this is possible. Read-until-sentinel protocol: each submitted block is wrapped so the interpreter prints a unique end marker once it finishes; this lets us collect stdout/stderr for that block without ambiguity. == Fork semantics on HF Spaces == `@spaces.GPU` runs the decorated function inside `multiprocessing.fork()` (`spaces==0.50.4`, `spaces/zero/wrappers.py:57`). The child fork inherits the parent's Popen object via copy-on-write memory, but inherited threads do NOT survive fork — only the calling thread continues in the child. That means after fork: - The parent's pump threads are still draining the interpreter's stdout pipe into the PARENT's in-memory queue. - The child has its own copy of the queue (empty at fork time) but no thread is filling it. - If the child writes to the interpreter's stdin (which works because fds are shared), the interpreter's output goes into the pipe, then into the PARENT's queue, never the child's. The fix: detect when we're in a process that did not start this Popen (via `os.getpid()` comparison) and start a fresh interpreter, pump threads, and queues local to this process. Each `@spaces.GPU` fork ends up with its own interpreter session. Cross-fork persistence happens via the filesystem (the `characters/` JSON files), not via the interpreter's in-memory state. Crash detection is lazy. We can't trust `_proc.poll()` for a process the parent started, so we only flag `crashed=True` when `stdin.write` itself raises BrokenPipeError. Real crashes are detected on the next submission. """ from __future__ import annotations import os import queue import subprocess import sys import threading import time import uuid from dataclasses import dataclass from pathlib import Path TOWNLET_HOME = Path("/tmp/townlet") RUN_TIMEOUT_S = 8.0 @dataclass class ShellResult: stdout: str stderr: str crashed: bool class SharedInterpreter: def __init__(self, cwd: Path | None = None) -> None: self.cwd = cwd or TOWNLET_HOME self.cwd.mkdir(parents=True, exist_ok=True) self._proc: subprocess.Popen | None = None self._stdout_q: queue.Queue[str] = queue.Queue() self._stderr_q: queue.Queue[str] = queue.Queue() self._lock = threading.Lock() # PID of the process that started self._proc. If we detect a # mismatch (we've been forked) we'll start a fresh interpreter for # this process so its output ends up in queues we can drain. self._started_pid: int | None = None def _owned_by_this_process(self) -> bool: return self._proc is not None and self._started_pid == os.getpid() def start(self) -> None: if self._owned_by_this_process(): return # Either never started, or started in a different process (fork). # In the fork case the parent still owns its own interpreter; we # just need our own. Re-init everything so the new pipes feed # OUR queues from OUR pump threads. Don't touch the parent's # _proc — let it keep running in the parent. self._proc = None self._stdout_q = queue.Queue() self._stderr_q = queue.Queue() self._lock = threading.Lock() env = os.environ.copy() env["PYTHONUNBUFFERED"] = "1" self._proc = subprocess.Popen( [sys.executable, "-i", "-u", "-q"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, cwd=str(self.cwd), text=True, bufsize=1, env=env, ) self._started_pid = os.getpid() threading.Thread( target=self._pump, args=(self._proc.stdout, self._stdout_q), daemon=True ).start() threading.Thread( target=self._pump, args=(self._proc.stderr, self._stderr_q), daemon=True ).start() def _pump(self, stream, q: queue.Queue[str]) -> None: for line in iter(stream.readline, ""): q.put(line) def is_alive(self) -> bool: return self._owned_by_this_process() and self._proc.poll() is None # type: ignore[union-attr] def submit(self, code: str) -> ShellResult: with self._lock: if not self._owned_by_this_process(): # Either first call ever, or first call in this fork. Spin # up a fresh interpreter so subsequent output lands in # queues we own. self.start() assert self._proc is not None and self._proc.stdin is not None marker = f"__townlet_{uuid.uuid4().hex}__" # Wrap user code as ONE exec() call so `python -i` sees exactly # one top-level statement. Compound statements written across # multiple physical lines trigger the REPL's continuation prompt # which never resolves over a piped stdin. inner = ( "import sys as __s, traceback as __tb\n" f"try: exec(compile({code!r}, '', 'exec'))\n" "except Exception: __tb.print_exc()\n" f"print({marker!r}, flush=True)\n" f"__s.stderr.write({marker!r} + chr(10)); __s.stderr.flush()\n" ) wrapped = f"exec({inner!r})\n" try: self._proc.stdin.write(wrapped) self._proc.stdin.flush() except BrokenPipeError: return ShellResult("", "interpreter pipe closed", True) out, err = self._collect(marker) return ShellResult(stdout=out, stderr=err, crashed=False) def _collect(self, marker: str) -> tuple[str, str]: deadline = time.monotonic() + RUN_TIMEOUT_S out_buf: list[str] = [] err_buf: list[str] = [] out_done = err_done = False while not (out_done and err_done): remaining = deadline - time.monotonic() if remaining <= 0: err_buf.append(f"[townlet] timeout after {RUN_TIMEOUT_S}s\n") break if not out_done: try: line = self._stdout_q.get(timeout=0.05) if marker in line: out_done = True else: out_buf.append(line) except queue.Empty: pass if not err_done: try: line = self._stderr_q.get(timeout=0.05) if marker in line: err_done = True else: err_buf.append(line) except queue.Empty: pass if not self.is_alive(): break return "".join(out_buf), "".join(err_buf) def stop(self) -> None: if self._proc is None: return try: self._proc.terminate() self._proc.wait(timeout=2) except Exception: self._proc.kill() self._proc = None self._started_pid = None