"""Interactive session manager for the compiler worker (Playground A1, Tasks 1-3). Manages interactive PTY sessions for Python and C programs: - Bounded output buffer with byte cursors (delta reads, no repetition). - Per-session tempdir, validated multi-file uploads. - PTY lifecycle: pty.openpty() + termios ECHO off + reader thread (select/os.read, incremental UTF-8 decoder) + parent-side watchdog (idle / max runtime / output overflow) — no preexec_fn (multi-threaded app). - C compile queue: threading.BoundedSemaphore(MAX_COMPILES), status queued → compiling → running, gcc without shell. - One daemon sweeper thread for retention cleanup + watchdog kills. """ import codecs import os import pty import re import secrets import select import shutil import signal import subprocess import sys import tempfile import termios import threading import time from collections import Counter, deque from dataclasses import dataclass, field # -------------------------------------------------------------------------- # Errors (mapped to HTTP codes in app.py) # -------------------------------------------------------------------------- class SessionError(Exception): """Base class for session manager errors.""" class SessionCapacityError(SessionError): """Registry full — HTTP 429.""" class SessionValidationError(SessionError): """Invalid request payload — HTTP 400.""" class SessionNotFoundError(SessionError): """Unknown session id — HTTP 404.""" class SessionNotRunningError(SessionError): """Input written to a session that is not running — HTTP 409.""" # -------------------------------------------------------------------------- # Configuration # -------------------------------------------------------------------------- def _env_int(name, default): raw = os.environ.get(name) if raw is None: return default try: return int(raw) except (TypeError, ValueError): return default TERMINAL_STATUSES = frozenset({"exited", "error", "stopped"}) # Explicit, validated status transitions. _ALLOWED_TRANSITIONS = { "queued": {"compiling", "running", "error", "stopped"}, "compiling": {"running", "error", "exited", "stopped"}, "running": {"exited", "error", "stopped"}, "exited": set(), "error": set(), "stopped": set(), } # -------------------------------------------------------------------------- # Session model # -------------------------------------------------------------------------- @dataclass class InteractiveSession: session_id: str language: str status: str created_at: float last_activity_at: float output_base_cursor: int = 0 output_next_cursor: int = 0 output_chunks: deque = field(default_factory=deque) process: subprocess.Popen | None = None master_fd: int | None = None exit_code: int | None = None error: str | None = None # --- extra bookkeeping ------------------------------------------------ tempdir: str | None = None started_at: float = 0.0 last_output_at: float = 0.0 input_pending: str = "" truncated: bool = False reader_thread: threading.Thread | None = None def _set_status(self, new_status): old = self.status if new_status == old: return if new_status not in _ALLOWED_TRANSITIONS.get(old, set()): raise SessionError( f"invalid status transition: {old} -> {new_status}" ) self.status = new_status # -------------------------------------------------------------------------- # Manager # -------------------------------------------------------------------------- class SessionManager: def __init__( self, *, max_sessions=None, max_compiles=None, queue_timeout=None, idle_timeout=None, max_runtime=None, terminal_retention=None, output_limit_bytes=None, max_files=None, max_source_bytes=None, sweeper_interval=1.0, autostart_sweeper=True, ): self.max_sessions = max_sessions or _env_int("INTERACTIVE_MAX_SESSIONS", 50) self.max_compiles = max_compiles or _env_int("INTERACTIVE_MAX_COMPILES", 2) self.queue_timeout = queue_timeout or _env_int("INTERACTIVE_QUEUE_TIMEOUT_SECONDS", 20) self.idle_timeout = idle_timeout or _env_int("INTERACTIVE_IDLE_TIMEOUT_SECONDS", 30) self.max_runtime = max_runtime or _env_int("INTERACTIVE_MAX_RUNTIME_SECONDS", 60) self.terminal_retention = terminal_retention or _env_int( "INTERACTIVE_TERMINAL_RETENTION_SECONDS", 15 ) self.output_limit_bytes = output_limit_bytes or _env_int( "INTERACTIVE_OUTPUT_LIMIT_BYTES", 262144 ) self.max_files = max_files or _env_int("INTERACTIVE_MAX_FILES", 20) self.max_source_bytes = max_source_bytes or _env_int( "INTERACTIVE_MAX_SOURCE_BYTES", 262144 ) self.sweeper_interval = sweeper_interval self._lock = threading.RLock() self._sessions = {} self._compile_semaphore = threading.BoundedSemaphore(self.max_compiles) self._compiling_count = 0 self._sweeper_stop = threading.Event() self._sweeper = None if autostart_sweeper: self._start_sweeper() # ------------------------------------------------------------------ # Sweeper daemon thread # ------------------------------------------------------------------ def _start_sweeper(self): self._sweeper = threading.Thread( target=self._sweeper_loop, name="session-sweeper", daemon=True, ) self._sweeper.start() def _sweeper_loop(self): while not self._sweeper_stop.wait(self.sweeper_interval): try: self.cleanup_expired() except Exception: # noqa: BLE001 — sweeper must never die pass def shutdown(self): """Stop sweeper, kill all child processes, remove tempdirs.""" self._sweeper_stop.set() if self._sweeper is not None and self._sweeper.is_alive(): self._sweeper.join(timeout=2) with self._lock: sessions = list(self._sessions.values()) self._sessions.clear() for s in sessions: if s.process is not None and s.process.poll() is None: self._kill_proc(s.process) self._cleanup_tempdir(s) # ------------------------------------------------------------------ # Output buffer (bounded, byte cursors) # ------------------------------------------------------------------ def _append_output(self, session, text): """Append decoded text; evict oldest chunks past the byte limit.""" if not text: return data = text if isinstance(text, str) else text.decode("utf-8", "replace") encoded_len = len(data.encode("utf-8")) limit = self.output_limit_bytes if encoded_len > limit: # Single chunk larger than the whole buffer: keep its head. data = data[:limit] encoded_len = limit with self._lock: session.output_chunks.clear() session.output_base_cursor = session.output_next_cursor session.truncated = True with self._lock: while ( session.output_chunks and (session.output_next_cursor - session.output_base_cursor) + encoded_len > limit ): old = session.output_chunks.popleft() session.output_base_cursor += len(old.encode("utf-8")) session.truncated = True session.output_chunks.append(data) session.output_next_cursor += encoded_len def _slice_from_cursor(self, chunks, base, next_cursor, cursor): """Return (text, missed_data) for byte cursor relative to chunk stream.""" if cursor < base: return "".join(chunks), True if cursor >= next_cursor: return "", False buf = [] offset = base missed = False for c in chunks: clen = len(c.encode("utf-8")) if offset + clen <= cursor: offset += clen continue if offset == cursor: buf.append(c) else: # Cursor points mid-chunk (pathological): reslice from byte offset. raw = c.encode("utf-8")[cursor - offset :] buf.append(raw.decode("utf-8", errors="replace")) missed = True offset += clen return "".join(buf), missed # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def create(self, language, files, active_file=None, stdin=""): """Validate, register, and launch a new interactive session.""" if language not in ("python", "c"): raise SessionValidationError(f"unsupported language: {language!r}") files = self._validate_files(language, files) with self._lock: if len(self._sessions) >= self.max_sessions: raise SessionCapacityError( f"session capacity reached: {self.max_sessions} active sessions" ) now = time.time() session = InteractiveSession( session_id=secrets.token_urlsafe(32), language=language, status="queued", created_at=now, last_activity_at=now, ) session.started_at = now session.input_pending = stdin if isinstance(stdin, str) else "" self._sessions[session.session_id] = session try: if language == "python": self._launch_python(session, files, active_file) else: self._launch_c(session, files) except SessionValidationError: with self._lock: self._sessions.pop(session.session_id, None) self._cleanup_tempdir(session) raise except Exception as e: # noqa: BLE001 — surface as error session with self._lock: session.status = "error" session.error = str(e) session.last_activity_at = time.time() return session def get_delta(self, session_id, cursor=0): with self._lock: session = self._sessions.get(session_id) if session is None: raise SessionNotFoundError(f"unknown session: {session_id}") chunks = list(session.output_chunks) base = session.output_base_cursor next_cursor = session.output_next_cursor buffer_truncated = session.truncated status = session.status language = session.language exit_code = session.exit_code error = session.error try: cursor = max(0, int(cursor)) except (TypeError, ValueError): cursor = 0 text, missed = self._slice_from_cursor(chunks, base, next_cursor, cursor) return { "session_id": session_id, "language": language, "status": status, "output": text, "cursor": next_cursor, "base_cursor": base, "truncated": missed, "buffer_truncated": buffer_truncated, "running": status == "running", "exit_code": exit_code, "error": error, } def write_input(self, session_id, text): if not isinstance(text, str): raise SessionValidationError("input must be a string") with self._lock: session = self._sessions.get(session_id) if session is None: raise SessionNotFoundError(f"unknown session: {session_id}") if session.status != "running" or session.master_fd is None: raise SessionNotRunningError( f"session {session_id} is not running (status={session.status})" ) fd = session.master_fd if not text: return True # input()/scanf() butuh Enter (newline) untuk submit — tambahkan bila belum ada normalized = text if text.endswith("\n") else text + "\n" ok = self._write_all(fd, normalized) if not ok: with self._lock: session.status = "stopped" session.error = "broken pipe while writing input" session.last_activity_at = time.time() raise SessionNotRunningError("session terminated while writing input") with self._lock: session.last_activity_at = time.time() return True def stop(self, session_id): """Stop a session; idempotent.""" with self._lock: session = self._sessions.get(session_id) if session is None: raise SessionNotFoundError(f"unknown session: {session_id}") if session.status == "stopped": return self._summary(session) if session.status in TERMINAL_STATUSES: # Already finished on its own — nothing to kill. return self._summary(session) proc = session.process session.status = "stopped" session.last_activity_at = time.time() if proc is not None and proc.poll() is None: self._kill_proc(proc) with self._lock: if session.exit_code is None and session.process is not None: session.exit_code = session.process.poll() return self._summary(session) def cleanup_expired(self): """Retention cleanup + watchdog kills. Returns number of sessions cleaned.""" now = time.time() to_remove = [] to_kill = [] # (session, reason) with self._lock: for sid, s in list(self._sessions.items()): if s.status in TERMINAL_STATUSES: if now - s.last_activity_at >= self.terminal_retention: to_remove.append(sid) elif s.status == "queued": if now - s.created_at >= self.queue_timeout: s.status = "error" s.error = f"queue timeout after {self.queue_timeout}s" s.last_activity_at = now to_remove.append(sid) elif s.status == "running": if now - s.last_activity_at >= self.idle_timeout: to_kill.append((s, "idle timeout")) elif now - s.started_at >= self.max_runtime: to_kill.append((s, "max runtime exceeded")) elif s.status == "compiling": if now - s.started_at >= self.max_runtime: to_kill.append((s, "max runtime exceeded")) for s, reason in to_kill: with self._lock: if s.status == "running" or s.status == "compiling": s.status = "stopped" s.error = reason s.last_activity_at = now if s.process is not None and s.process.poll() is None: self._kill_proc(s.process) with self._lock: for sid in to_remove: s = self._sessions.pop(sid, None) if s is not None: self._cleanup_tempdir(s) return len(to_remove) + len(to_kill) def stats(self): with self._lock: counts = Counter(s.status for s in self._sessions.values()) total = len(self._sessions) return { "active_sessions": total, "queued": counts.get("queued", 0), "compiling": counts.get("compiling", 0), "running": counts.get("running", 0), "exited": counts.get("exited", 0), "error": counts.get("error", 0), "stopped": counts.get("stopped", 0), "limits": { "max_sessions": self.max_sessions, "max_compiles": self.max_compiles, "queue_timeout_seconds": self.queue_timeout, "idle_timeout_seconds": self.idle_timeout, "max_runtime_seconds": self.max_runtime, "terminal_retention_seconds": self.terminal_retention, "output_limit_bytes": self.output_limit_bytes, "max_files": self.max_files, "max_source_bytes": self.max_source_bytes, }, } # ------------------------------------------------------------------ # Validation # ------------------------------------------------------------------ _NAME_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]*$") @staticmethod def _valid_name(name, allowed_exts): if not isinstance(name, str) or not name: return False if "/" in name or "\\" in name: return False if name in (".", "..") or name.startswith("..") or os.path.isabs(name): return False if not SessionManager._NAME_RE.match(name): return False return os.path.splitext(name)[1].lower() in allowed_exts def _validate_files(self, language, files): allowed_exts = {".py"} if language == "python" else {".c", ".h"} if not isinstance(files, list) or not files: raise SessionValidationError("files must be a non-empty list") if len(files) > self.max_files: raise SessionValidationError( f"too many files: {len(files)} > {self.max_files}" ) seen = {} total_bytes = 0 normalized = [] for item in files: if not isinstance(item, dict) or "name" not in item or "content" not in item: raise SessionValidationError("each file must be {name, content}") name = item["name"] content = item["content"] if not isinstance(content, str): raise SessionValidationError(f"content of {name!r} must be a string") if not self._valid_name(name, allowed_exts): raise SessionValidationError( f"invalid file name {name!r} (basename only, extension " f"{'/'.join(sorted(allowed_exts))})" ) key = name.lower() if key in seen: raise SessionValidationError( f"duplicate file name (case-insensitive): {name!r} and {seen[key]!r}" ) seen[key] = name total_bytes += len(content.encode("utf-8")) if total_bytes > self.max_source_bytes: raise SessionValidationError( f"total source exceeds {self.max_source_bytes} bytes" ) normalized.append({"name": name, "content": content}) return normalized # ------------------------------------------------------------------ # Launch: Python # ------------------------------------------------------------------ def _launch_python(self, session, files, active_file): tempdir = tempfile.mkdtemp(prefix="elemes-py-") session.tempdir = tempdir self._write_files(tempdir, files) names = [f["name"] for f in files] target = active_file if active_file in names else names[0] if not target.endswith(".py"): raise SessionValidationError(f"active file must be a .py file: {target!r}") self._spawn_pty(session, [sys.executable, target], tempdir) # ------------------------------------------------------------------ # Launch: C with compile queue # ------------------------------------------------------------------ def _launch_c(self, session, files): tempdir = tempfile.mkdtemp(prefix="elemes-c-") session.tempdir = tempdir self._write_files(tempdir, files) c_files = [f["name"] for f in files if f["name"].endswith(".c")] if not c_files: raise SessionValidationError("no .c source files to compile") acquired = self._compile_semaphore.acquire(timeout=self.queue_timeout) if not acquired: with self._lock: session.status = "error" session.error = f"compile queue timeout after {self.queue_timeout}s" session.last_activity_at = time.time() return with self._lock: if session.status != "queued": # Stopped/errored while waiting in queue. self._compile_semaphore.release() return session.status = "compiling" self._compiling_count += 1 try: res = subprocess.run( ["gcc", "-std=c11", "-Wall", "-Wextra", "-I", tempdir] + c_files + ["-o", "program"], capture_output=True, text=True, timeout=30, cwd=tempdir, shell=False, ) if res.stdout: self._append_output(session, res.stdout) if res.stderr: self._append_output(session, res.stderr) if res.returncode != 0: with self._lock: session.status = "error" session.error = "compilation failed" session.exit_code = res.returncode session.last_activity_at = time.time() return except subprocess.TimeoutExpired: with self._lock: session.status = "error" session.error = "compilation timed out" session.last_activity_at = time.time() return finally: with self._lock: self._compiling_count -= 1 self._compile_semaphore.release() self._spawn_pty(session, [os.path.join(tempdir, "program")], tempdir) # ------------------------------------------------------------------ # PTY lifecycle # ------------------------------------------------------------------ @staticmethod def _write_files(tempdir, files): for f in files: path = os.path.join(tempdir, f["name"]) with open(path, "w", encoding="utf-8") as fh: fh.write(f["content"]) def _spawn_pty(self, session, argv, cwd): """Spawn argv on a PTY (no shell, no preexec_fn), then start reader.""" with self._lock: if session.status in TERMINAL_STATUSES or session.status == "stopped": return None master_fd, slave_fd = pty.openpty() try: attrs = termios.tcgetattr(slave_fd) attrs[3] &= ~termios.ECHO termios.tcsetattr(slave_fd, termios.TCSANOW, attrs) proc = subprocess.Popen( argv, stdin=slave_fd, stdout=slave_fd, stderr=slave_fd, cwd=cwd, start_new_session=True, shell=False, close_fds=True, ) os.close(slave_fd) except Exception: os.close(master_fd) try: os.close(slave_fd) except OSError: pass raise session.process = proc session.master_fd = master_fd session.status = "running" session.started_at = time.time() session.last_output_at = time.time() session.last_activity_at = time.time() pending = session.input_pending session.input_pending = "" if pending: self._write_all(master_fd, pending) thread = threading.Thread( target=self._reader_loop, args=(session,), name=f"reader-{session.session_id[:8]}", daemon=True, ) session.reader_thread = thread thread.start() return session def _reader_loop(self, session): """select + os.read on the PTY master; incremental UTF-8 decoding.""" decoder = codecs.getincrementaldecoder("utf-8")(errors="replace") fd = session.master_fd overflow = False try: while True: try: ready, _, _ = select.select([fd], [], [], 0.5) except (OSError, ValueError): break if ready: try: data = os.read(fd, 4096) except OSError: break if not data: break text = decoder.decode(data) if text: self._append_output(session, text) with self._lock: session.last_output_at = time.time() session.last_activity_at = time.time() overflow = self._output_overflowed(session) if overflow: break else: proc = session.process if proc is not None and proc.poll() is not None: # Child exited: drain any residual buffered output. self._drain(session, fd, decoder) break finally: with self._lock: if overflow: session.status = "stopped" session.error = "output limit exceeded" elif session.status not in TERMINAL_STATUSES: session.status = "exited" if session.process is not None: session.exit_code = session.process.poll() session.last_activity_at = time.time() self._close_master_fd(session) if overflow: proc = session.process if proc is not None and proc.poll() is None: self._kill_proc(proc) def _drain(self, session, fd, decoder): deadline = time.time() + 0.5 while time.time() < deadline: try: ready, _, _ = select.select([fd], [], [], 0.1) except (OSError, ValueError): return if not ready: return try: data = os.read(fd, 4096) except OSError: return if not data: return text = decoder.decode(data) if text: self._append_output(session, text) def _output_overflowed(self, session): return ( session.output_next_cursor - session.output_base_cursor >= self.output_limit_bytes ) def _close_master_fd(self, session): fd = session.master_fd if fd is not None: try: os.close(fd) except OSError: pass session.master_fd = None @staticmethod def _write_all(fd, text): view = memoryview(text.encode("utf-8")) while view: try: written = os.write(fd, view) except OSError: return False if written <= 0: return False view = view[written:] return True @staticmethod def _kill_proc(proc): """SIGTERM the process group, escalate to SIGKILL after 1s.""" try: os.killpg(os.getpgid(proc.pid), signal.SIGTERM) except (ProcessLookupError, OSError): try: proc.terminate() except Exception: # noqa: BLE001 pass try: proc.wait(timeout=1.0) except subprocess.TimeoutExpired: try: os.killpg(os.getpgid(proc.pid), signal.SIGKILL) except (ProcessLookupError, OSError): try: proc.kill() except Exception: # noqa: BLE001 pass try: proc.wait(timeout=1.0) except Exception: # noqa: BLE001 pass def _cleanup_tempdir(self, session): if session.tempdir: shutil.rmtree(session.tempdir, ignore_errors=True) session.tempdir = None @staticmethod def _summary(session): return { "session_id": session.session_id, "language": session.language, "status": session.status, "exit_code": session.exit_code, "error": session.error, "created_at": session.created_at, "last_activity_at": session.last_activity_at, }