Something went wrong. Try again.
This repository has no description
Something went wrong. Try again.
9.1 kB · 231 lines
Python
at main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232"""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 "<key>\\t<value>" 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 tothe original 2005 rule parser for that line, so the game stays playable."""import jsonimport queueimport sysimport timefrom http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from .classify import Utterance, classifyfrom .config import load_settingsfrom .dialogue import readablefrom .jev_client import JevClient, JevErrorfrom .monitor import PAGE_PATH, EventLog
# Fixed on purpose: the game patch has this address compiled in.HOST, PORT = "127.0.0.1", 8765MAX_BODY_BYTES = 64 * 1024SSE_HEARTBEAT_SECONDS = 15LOOPBACK_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()