game/interpreter.py

"""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}, '<townlet>', '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