"""Loopback HTTP server the patched game talks to. POST /da Line-oriented, for the game's old JVM, which has no JSON parser and cannot speak modern TLS (hence this sidecar). Request lines are "\\t" with keys text, context and said. Response is one "id charID p1 p2 p3" line per act. POST /classify JSON in, JSON out. For people and tools. GET / Live monitor page. GET /events Server-sent events feeding that page. GET /history Recent events as JSON. GET /health When Jev cannot be reached /da answers 502, and the game patch falls back to the original 2005 rule parser for that line, so the game stays playable. """ import json import queue import sys import time from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from .classify import Utterance, classify from .config import load_settings from .dialogue import readable from .jev_client import JevClient, JevError from .monitor import PAGE_PATH, EventLog # Fixed on purpose: the game patch has this address compiled in. HOST, PORT = "127.0.0.1", 8765 MAX_BODY_BYTES = 64 * 1024 SSE_HEARTBEAT_SECONDS = 15 LOOPBACK_HOSTS = ("127.0.0.1", "localhost") def parse_da_request(body): text, contexts, said = "", [], [] for line in body.splitlines(): key, _, value = line.partition("\t") value = value.strip() if key == "text": text = value elif key == "context" and value: contexts.append(value) elif key == "said" and value: said.append(value) spoken = [line for line in (readable(s) for s in said) if line] return Utterance(text=text, contexts=tuple(contexts), recent_dialogue=tuple(spoken)) def format_da_response(acts): return "".join( "%d %d %d %d %d\n" % (a.da_id, a.char_id, a.param1, a.param2, a.param3) for a in acts ) def describe(acts): parts = [] for act in acts: params = ",".join(p for p in act.param_names if p) parts.append("%s(%s) %.2f" % (act.name, params, act.probability)) return " ".join(parts) def _local_hosts(port): return {"%s:%d" % (host, port) for host in LOOPBACK_HOSTS} def has_local_host(headers, port): """False for a request that reached us under a foreign name (DNS rebinding), which would let a web page read what the player typed.""" return headers.get("Host") in _local_hosts(port) def has_local_origin(headers, port): """False for a cross-site fetch. Every POST spends the player's OpenRouter credit. The game and command-line tools send no Origin at all.""" origin = headers.get("Origin") return origin is None or origin in {"http://" + host for host in _local_hosts(port)} def make_handler(client, settings): events = EventLog() class Handler(BaseHTTPRequestHandler): def log_message(self, *args): pass # one line per classification is logged below instead def _send(self, status, content_type, payload): data = payload.encode("utf-8") self.send_response(status) self.send_header("Content-Type", content_type + "; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.send_header("Cache-Control", "no-store") self.end_headers() self.wfile.write(data) def _body(self): try: length = int(self.headers.get("Content-Length") or 0) except ValueError: length = 0 return self.rfile.read(max(0, min(length, MAX_BODY_BYTES))).decode("utf-8", "replace") def _classify(self, source, utterance): """Classify, and record the outcome either way. Re-raises JevError.""" started = time.monotonic() try: result = classify(utterance, client, settings) except JevError as error: elapsed_ms = (time.monotonic() - started) * 1000 events.record(source, utterance, error=str(error), elapsed_ms=elapsed_ms) print("JEV ERROR %r: %s" % (utterance.text, error), file=sys.stderr, flush=True) raise elapsed_ms = (time.monotonic() - started) * 1000 events.record(source, utterance, result=result, elapsed_ms=elapsed_ms) print("%4d ms %r -> %s" % (elapsed_ms, utterance.text, describe(result.acts)), flush=True) return result def do_GET(self): if not has_local_host(self.headers, self.server.server_address[1]): return self._send(403, "text/plain", "local requests only") if self.path == "/": return self._send(200, "text/html", PAGE_PATH.read_text(encoding="utf-8")) if self.path == "/events": return self._stream_events() if self.path == "/history": return self._send(200, "application/json", json.dumps(events.history())) if self.path == "/health": return self._send(200, "application/json", json.dumps({"ok": True, "model": settings.model})) self._send(404, "text/plain", "not found\n") def do_POST(self): port = self.server.server_address[1] if not (has_local_host(self.headers, port) and has_local_origin(self.headers, port)): return self._send(403, "text/plain", "local requests only") if self.path == "/da": return self._post_da() if self.path == "/classify": return self._post_classify() self._send(404, "text/plain", "not found\n") def _stream_events(self): # Subscribe before answering, so nothing recorded in between is lost. subscriber = events.subscribe() try: self.send_response(200) self.send_header("Content-Type", "text/event-stream") self.send_header("Cache-Control", "no-store") self.end_headers() while True: try: event = subscriber.get(timeout=SSE_HEARTBEAT_SECONDS) chunk = "data: %s\n\n" % json.dumps(event) except queue.Empty: chunk = ": keepalive\n\n" self.wfile.write(chunk.encode("utf-8")) self.wfile.flush() except OSError: pass # the page was closed finally: events.unsubscribe(subscriber) def _post_da(self): try: acts = self._classify("game", parse_da_request(self._body())).acts except JevError: # Any non-200 tells the game patch to use its original parser. return self._send(502, "text/plain", "") self._send(200, "text/plain", format_da_response(acts)) def _post_classify(self): try: request = json.loads(self._body()) if not isinstance(request["text"], str): raise TypeError("text must be a string") utterance = Utterance( text=request["text"], contexts=tuple(str(c) for c in request.get("contexts", ())), recent_dialogue=tuple(str(line) for line in request.get("recent_dialogue", ())), ) source = "monitor" if request.get("source") == "monitor" else "api" except (ValueError, KeyError, TypeError) as error: return self._send(400, "application/json", json.dumps({"error": "bad request: %s" % error})) try: result = self._classify(source, utterance) except JevError as error: return self._send(502, "application/json", json.dumps({"error": str(error)})) self._send(200, "application/json", json.dumps({ "understood": result.understood, "acts": [act.as_dict() for act in result.acts], })) return Handler class SidecarServer(ThreadingHTTPServer): # The default (True) lets a second sidecar bind the same port on Windows, # after which requests go to either one at random. allow_reuse_address = False MONITOR_URL = "http://%s:%d/" % (HOST, PORT) def create_server(): """A sidecar that is already listening; call serve() to run it. Exits with a message if the key is missing or the port is taken.""" settings = load_settings() client = JevClient.from_settings(settings) try: server = SidecarServer((HOST, PORT), make_handler(client, settings)) except OSError as error: sys.exit("cannot listen on %s:%d (is a sidecar already running?): %s" % (HOST, PORT, error)) print("jev-facade sidecar (model %s)" % settings.model) print("monitor: %s" % MONITOR_URL, flush=True) return server def serve(server): try: server.serve_forever() except KeyboardInterrupt: pass def main(): serve(create_server()) if __name__ == "__main__": main()