PC3 Phase 2.2: game-server orchestration + POST /api/worlds/{id}/join

- backend/app/services/game_servers.py: an in-process supervisor. One
  headless Godot dedicated-server process per world, keyed by world_id,
  spawned on demand from the join endpoint, port-allocated from a
  configured range, with a background reaper that kills a world's server
  once its status file reports 0 players for the idle timeout. Off by
  default (GAME_SERVERS_ENABLED); the checkpoint/playtest launcher enables it.
- POST /api/worlds/{id}/join -> {world_id, host, port}; 404 for
  unknown/hidden worlds, 503 when orchestration is disabled. Auth required.
- main.py lifespan starts/stops the supervisor.
- config.py: GODOT_BIN, GAME_PROJECT_DIR, GAME_SERVER_HOST, port range,
  idle timeout.
- tests/test_world_join.py (auth / 404 / visibility / 503) -- suite 72 passed.
- checkpoint runner also runs the 2p multiplayer shell test.

Verified end to end: enabled orchestration, POST /join spawned the Godot
server, a real client connected to the returned address and moved; backend
shutdown reaped the process.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015jtxAiu5dAW4WxCNdGkrkJ

backend/app/main.py

@@ -9,12 +9,15 @@ from app.config import settings
 from app.routers import auth, users, worlds, assets, avatar
 
 
+from app.services.game_servers import supervisor
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     """Application lifespan: startup and shutdown events."""
-    # Startup: initialize database, connect to Redis, etc.
+    supervisor.start()
     yield
-    # Shutdown: close connections, cleanup
+    await supervisor.shutdown()
 
 
 app = FastAPI(

backend/app/routers/worlds.py

@@ -12,6 +12,7 @@ from sqlalchemy.orm import selectinload
 from app.database import get_db
 from app.dependencies import CurrentUser
 from app.models.world import World, WorldVisibility
+from app.services.game_servers import supervisor
 
 router = APIRouter()
 
@@ -106,6 +107,40 @@ async def get_world(
     return world
 
 
+class JoinResponse(BaseModel):
+    """Address of the dedicated server hosting this world."""
+    world_id: int
+    host: str
+    port: int
+
+
+@router.post("/{world_id}/join", response_model=JoinResponse)
+async def join_world(
+    world_id: int,
+    current_user: CurrentUser,
+    db: DbDep,
+):
+    """Ensure a dedicated game server is running for this world and return its
+    ENet address for the client to connect to.
+
+    503 if game-server orchestration isn't enabled on this backend (e.g. a
+    plain dev/test instance with no Godot binary).
+    """
+    result = await db.execute(select(World).where(World.id == world_id))
+    world = result.scalar_one_or_none()
+    if not world:
+        raise HTTPException(status_code=404, detail="World not found")
+    if world.visibility != WorldVisibility.PUBLIC and world.owner_id != current_user.id:
+        raise HTTPException(status_code=404, detail="World not found")
+
+    try:
+        info = await supervisor.ensure_server(str(world_id))
+    except RuntimeError as exc:
+        raise HTTPException(status_code=503, detail=str(exc))
+
+    return JoinResponse(world_id=world_id, host=info["host"], port=info["port"])
+
+
 @router.post("/", response_model=WorldResponse, status_code=201)
 async def create_world(
     world_data: WorldCreate,

backend/app/services/game_servers.py

@@ -0,0 +1,182 @@
+"""Game-server orchestration.
+
+On ``POST /api/worlds/{id}/join`` the backend ensures a headless Godot
+dedicated-server process is running for that world and returns its address.
+One process per world, keyed by ``world_id``. A background reaper kills a
+world's server once its status file reports zero players for
+``GAME_SERVER_IDLE_TIMEOUT_SECONDS``.
+
+This lives in-process (a module-level singleton) rather than as a separate
+service, so the backend MUST run single-worker -- which it does
+(``uvicorn app.main:app`` with no ``--workers``). The Godot binary and
+project are on the host (see ``settings.GODOT_BIN`` / ``GAME_PROJECT_DIR``);
+the backend spawns them as subprocesses.
+
+Design notes / limits (PC3):
+ - All worlds currently load the same test level (per-world content is M2/M7).
+   ``world_id`` still scopes a distinct server process + session.
+ - Port allocation is a linear scan of the configured range.
+ - No crash-restart: if a server process dies, the next join respawns it.
+"""
+from __future__ import annotations
+
+import asyncio
+import json
+import socket
+import subprocess
+import tempfile
+import time
+from dataclasses import dataclass, field
+from pathlib import Path
+
+from app.config import settings
+
+
+@dataclass
+class _WorldServer:
+    world_id: str
+    port: int
+    proc: subprocess.Popen
+    status_path: Path
+    started_at: float = field(default_factory=time.time)
+    last_join: float = field(default_factory=time.time)
+
+    def status(self) -> dict:
+        try:
+            return json.loads(self.status_path.read_text())
+        except (OSError, ValueError):
+            return {}
+
+    def player_count(self) -> int:
+        return int(self.status().get("players", 0))
+
+    def alive(self) -> bool:
+        return self.proc.poll() is None
+
+    def stop(self) -> None:
+        if self.alive():
+            self.proc.terminate()
+            try:
+                self.proc.wait(timeout=10)
+            except subprocess.TimeoutExpired:
+                self.proc.kill()
+        self.status_path.unlink(missing_ok=True)
+
+
+class GameServerSupervisor:
+    def __init__(self) -> None:
+        self._servers: dict[str, _WorldServer] = {}
+        self._lock = asyncio.Lock()
+        self._reaper: asyncio.Task | None = None
+        self._runtime_dir = Path(tempfile.gettempdir()) / "agp-game-servers"
+        self._runtime_dir.mkdir(parents=True, exist_ok=True)
+
+    # ── lifecycle ────────────────────────────────────────────────────────
+    def start(self) -> None:
+        if not settings.GAME_SERVERS_ENABLED:
+            return
+        if self._reaper is None or self._reaper.done():
+            self._reaper = asyncio.create_task(self._reap_loop())
+
+    async def shutdown(self) -> None:
+        if self._reaper:
+            self._reaper.cancel()
+        async with self._lock:
+            for srv in list(self._servers.values()):
+                srv.stop()
+            self._servers.clear()
+
+    # ── join ─────────────────────────────────────────────────────────────
+    async def ensure_server(self, world_id: str) -> dict:
+        """Return ``{"host", "port", "world_id"}`` for the world's dedicated
+        server, spawning it if needed. Raises RuntimeError if orchestration
+        is disabled or no port/engine is available."""
+        if not settings.GAME_SERVERS_ENABLED:
+            raise RuntimeError("game servers are not enabled on this backend")
+
+        async with self._lock:
+            srv = self._servers.get(world_id)
+            if srv is not None and not srv.alive():
+                srv.stop()
+                srv = None
+                del self._servers[world_id]
+
+            if srv is None:
+                srv = self._spawn(world_id)
+                self._servers[world_id] = srv
+                # Give the engine a moment to bind the ENet port before we
+                # hand the address to a client.
+                await self._await_port(srv.port)
+
+            srv.last_join = time.time()
+            return {"host": settings.GAME_SERVER_HOST, "port": srv.port, "world_id": world_id}
+
+    # ── internals ────────────────────────────────────────────────────────
+    def _spawn(self, world_id: str) -> _WorldServer:
+        port = self._free_port()
+        status_path = self._runtime_dir / f"world_{world_id}_{port}.json"
+        status_path.unlink(missing_ok=True)
+        cmd = [
+            settings.GODOT_BIN, "--headless",
+            "--path", settings.GAME_PROJECT_DIR,
+            "res://scenes/server.tscn", "--",
+            "--port", str(port),
+            "--world", str(world_id),
+            "--status-file", str(status_path),
+        ]
+        proc = subprocess.Popen(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+        return _WorldServer(world_id=world_id, port=port, proc=proc, status_path=status_path)
+
+    def _free_port(self) -> int:
+        used = {s.port for s in self._servers.values()}
+        for p in range(settings.GAME_SERVER_PORT_MIN, settings.GAME_SERVER_PORT_MAX + 1):
+            if p in used:
+                continue
+            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+                try:
+                    s.bind(("0.0.0.0", p))
+                    return p
+                except OSError:
+                    continue
+        raise RuntimeError("no free game-server port in the configured range")
+
+    async def _await_port(self, port: int, timeout: float = 8.0) -> None:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            # ENet is UDP; a bind failure means something (our server) has it.
+            with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
+                try:
+                    s.bind(("0.0.0.0", port))
+                except OSError:
+                    return
+            await asyncio.sleep(0.25)
+
+    async def _reap_loop(self) -> None:
+        try:
+            while True:
+                await asyncio.sleep(30)
+                async with self._lock:
+                    now = time.time()
+                    for wid, srv in list(self._servers.items()):
+                        if not srv.alive():
+                            srv.stop()
+                            del self._servers[wid]
+                            continue
+                        idle_for = now - srv.last_join
+                        if idle_for > settings.GAME_SERVER_IDLE_TIMEOUT_SECONDS and srv.player_count() == 0:
+                            srv.stop()
+                            del self._servers[wid]
+        except asyncio.CancelledError:
+            pass
+
+    def debug_state(self) -> list[dict]:
+        return [
+            {
+                "world_id": s.world_id, "port": s.port, "alive": s.alive(),
+                "players": s.player_count(), "uptime_s": round(time.time() - s.started_at),
+            }
+            for s in self._servers.values()
+        ]
+
+
+supervisor = GameServerSupervisor()

backend/tests/test_world_join.py

@@ -0,0 +1,54 @@
+"""Tests for POST /api/worlds/{id}/join.
+
+The happy path spawns a real Godot process, so it's covered by a manual /
+checkpoint integration test, not here. These cover the envelope: auth,
+visibility, and the 503 when orchestration is disabled (the default).
+"""
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.main import app
+
+
+def _auth(client: TestClient, name: str) -> tuple[dict, int]:
+    r = client.post("/api/auth/register", json={
+        "username": name, "email": f"{name}@e.com", "password": "secret123", "over_18": True,
+    })
+    assert r.status_code == 201, r.text
+    headers = {"Authorization": f"Bearer {r.json()['access_token']}"}
+    me = client.get("/api/users/me", headers=headers)
+    return headers, me.json()["id"]
+
+
+def _make_world(client: TestClient, headers: dict, visibility: str = "public") -> int:
+    r = client.post("/api/worlds/", headers=headers, json={"name": "W", "visibility": visibility})
+    assert r.status_code == 201, r.text
+    return r.json()["id"]
+
+
+class TestJoinWorld:
+    def test_requires_auth(self):
+        client = TestClient(app)
+        assert client.post("/api/worlds/1/join").status_code == 401
+
+    def test_unknown_world_is_404(self):
+        client = TestClient(app)
+        headers, _ = _auth(client, "join_404")
+        assert client.post("/api/worlds/999999/join", headers=headers).status_code == 404
+
+    def test_private_world_of_another_user_is_404(self):
+        client = TestClient(app)
+        owner_headers, _ = _auth(client, "join_owner")
+        wid = _make_world(client, owner_headers, "private")
+        other_headers, _ = _auth(client, "join_other")
+        assert client.post(f"/api/worlds/{wid}/join", headers=other_headers).status_code == 404
+
+    def test_disabled_orchestration_returns_503(self):
+        # GAME_SERVERS_ENABLED defaults to False in tests.
+        client = TestClient(app)
+        headers, _ = _auth(client, "join_503")
+        wid = _make_world(client, headers, "public")
+        r = client.post(f"/api/worlds/{wid}/join", headers=headers)
+        assert r.status_code == 503
+        assert "not enabled" in r.json()["detail"]

scripts/run_checkpoint_tests.py

@@ -156,6 +156,16 @@ def main():
                             except subprocess.TimeoutExpired:
                                 game_server.kill()
                 results["godot smoke tests"] = godot_ok
+
+                # 2-player multiplayer test (its own shell harness: server + 2
+                # real client processes).
+                mp_script = ROOT / "scripts" / "run_multiplayer_2p_test.sh"
+                if mp_script.exists():
+                    print("\n--- scripts/run_multiplayer_2p_test.sh ---")
+                    mp_env = os.environ.copy()
+                    mp_env["GODOT_BIN"] = godot_bin
+                    r = run(["bash", str(mp_script)], env=mp_env)
+                    results["multiplayer 2p test"] = r.returncode == 0
         finally:
             server_proc.terminate()
             try: