| |
| from __future__ import annotations |
|
|
| import argparse |
| import hashlib |
| import json |
| import os |
| import signal |
| import threading |
| import time |
| from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer |
|
|
|
|
| class Handler(BaseHTTPRequestHandler): |
| server_version = "HarnessDummyRuntime/2" |
|
|
| def _send(self, status, obj): |
| raw = json.dumps(obj).encode("utf-8") |
| self.send_response(status) |
| self.send_header("Content-Type", "application/json") |
| self.send_header("Content-Length", str(len(raw))) |
| self.end_headers() |
| self.wfile.write(raw) |
|
|
| def do_GET(self): |
| if self.path == "/health": |
| self._send(200, { |
| "status": "ok", |
| "pid": os.getpid(), |
| "runtime": "dummy", |
| "nonce": self.server.runtime_nonce, |
| "model_id": self.server.model_id, |
| "model_sha256": self.server.model_sha256, |
| }) |
| return |
|
|
| if self.path == "/provenance": |
| self._send(200, { |
| "schema_version": 1, |
| "runtime": "dummy", |
| "pid": os.getpid(), |
| "nonce": self.server.runtime_nonce, |
| "model_id": self.server.model_id, |
| "model_sha256": self.server.model_sha256, |
| "runtime_id": self.server.runtime_id, |
| }) |
| return |
|
|
| if self.path == "/v1/models": |
| self._send(200, { |
| "object": "list", |
| "data": [{ |
| "id": self.server.model_id, |
| "object": "model", |
| }], |
| }) |
| return |
|
|
| self._send(404, {"error": "not_found"}) |
|
|
| def do_POST(self): |
| |
| |
| |
| if self.path == "/v1/chat/completions": |
| length = int(self.headers.get("Content-Length", "0")) |
| raw = self.rfile.read(length) if length else b"{}" |
| try: |
| req = json.loads(raw.decode("utf-8")) |
| except Exception: |
| self._send(400, {"error": "bad_json"}) |
| return |
|
|
| if str(req.get("model")) != self.server.model_id: |
| self._send(400, {"error": "model_mismatch"}) |
| return |
|
|
| content = "DUMMY_BOUND_RESPONSE" |
| self._send(200, { |
| "id": "dummy-chat", |
| "object": "chat.completion", |
| "choices": [{ |
| "index": 0, |
| "finish_reason": "stop", |
| "message": { |
| "role": "assistant", |
| "content": content, |
| }, |
| }], |
| "usage": { |
| "prompt_tokens": 1, |
| "completion_tokens": 1, |
| "total_tokens": 2, |
| }, |
| }) |
| return |
|
|
| self._send(404, {"error": "not_found"}) |
|
|
| def log_message(self, fmt, *args): |
| return |
|
|
|
|
| def main(): |
| ap = argparse.ArgumentParser() |
| ap.add_argument("--host", required=True) |
| ap.add_argument("--port", required=True, type=int) |
| ap.add_argument("--nonce", required=True) |
| ap.add_argument("--model-id", default="dummy-model") |
| ap.add_argument("--model-sha256", required=True) |
| ap.add_argument("--runtime-id", default="dummy-runtime-v2") |
| ap.add_argument("--exit-after", type=float, default=0.0) |
| args = ap.parse_args() |
|
|
| server = ThreadingHTTPServer( |
| (args.host, args.port), |
| Handler, |
| ) |
| server.runtime_nonce = args.nonce |
| server.model_id = args.model_id |
| server.model_sha256 = args.model_sha256 |
| server.runtime_id = args.runtime_id |
|
|
| stop_event = threading.Event() |
|
|
| def stop_handler(signum, frame): |
| stop_event.set() |
| threading.Thread( |
| target=server.shutdown, |
| daemon=True, |
| ).start() |
|
|
| signal.signal(signal.SIGTERM, stop_handler) |
| signal.signal(signal.SIGINT, stop_handler) |
|
|
| if args.exit_after > 0: |
| def timer(): |
| time.sleep(args.exit_after) |
| stop_handler(signal.SIGTERM, None) |
| threading.Thread(target=timer, daemon=True).start() |
|
|
| server.serve_forever(poll_interval=0.1) |
| server.server_close() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
