"""Standard-library HTTP adapter. No model is loaded until main().""" from __future__ import annotations import argparse import json from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path if __package__: from .core import ItemFailure, ServiceFailure, UnsupportedRequest, Service, ROOT, load_json, validate_config, apply_profile, validate_calibration, validate_threshold else: from core import ItemFailure, ServiceFailure, UnsupportedRequest, Service, ROOT, load_json, validate_config, apply_profile, validate_calibration, validate_threshold BINDS = ("127.0.0.1", "0.0.0.0") def make_handler(service): def service_dead(): check = getattr(service, 'is_dead', None) return check() if callable(check) else getattr(service, 'dead', False) class Handler(BaseHTTPRequestHandler): protocol_version='HTTP/1.1' def handle_expect_100(self): if self.command == 'POST' and service_dead(): self.close_connection=True self.respond(503, {'error':'service failure'}) return False try: n=int(self.headers.get('Content-Length','0')) if n<=0 or n>service.config['server']['max_request_bytes'] or self.headers.get('Transfer-Encoding'): raise ValueError except (ValueError,TypeError): self.close_connection=True self.respond(400,{'error':'Invalid request size'}) return False self.send_response_only(100) self.end_headers() return True def setup(self): super().setup() self.connection.settimeout(service.config["server"]["socket_timeout_seconds"]) def log_message(self, format, *args): # Avoid logging request contents, model paths or network endpoints. pass def respond(self, status, value): data = json.dumps(value, ensure_ascii=False, allow_nan=False).encode("utf-8") self.send_response(status) self.send_header("Content-Type", "application/json; charset=utf-8") self.send_header("Content-Length", str(len(data))) self.end_headers() self.wfile.write(data) def do_GET(self): if self.path.split('?', 1)[0].rstrip('/') == '/health': if __package__: from .attention import attention_health else: from attention import attention_health health=service.health() # Attention backend and runtime versions are information, never warnings. info=health.setdefault('info',{}) info.update(attention_health()) info['runtime']=getattr(service.backend,'runtime_info',None) self.respond(200, health) elif self.path.split('?',1)[0].rstrip('/') == '/engine_stats': # Read-only engine counters for maintainer usage reconciliation; not used by evaluation. try: self.respond(200,service.engine_stats()) except Exception: self.respond(503,{'error':'Engine counters unavailable'}) else: self.respond(404, {'error':'Unknown route'}) def do_POST(self): # After an unrecoverable engine error or the third consecutive item failure, # every POST is answered 503 without reading or processing the request. if service_dead(): self.close_connection=True self.respond(503, {'error':'service failure'}) return if self.path.split("?", 1)[0].rstrip("/") != "/v1/systemone": self.close_connection=True self.respond(404, {"error": "Unknown route"}) return try: if self.headers.get("Transfer-Encoding"): raise ValueError("Transfer-Encoding is unsupported") length = int(self.headers.get("Content-Length", "0")) if length <= 0 or length > service.config["server"]["max_request_bytes"]: raise ValueError("Invalid request size") def reject(value): raise ValueError("Non-finite JSON value") raw = self.rfile.read(length) if len(raw) != length: raise ValueError("Incomplete request body") payload = json.loads(raw.decode("utf-8"), parse_constant=reject) # Validate before inference so model ValueError is never a client error. if __package__: from .core import normalize_request else: from core import normalize_request normalize_request(payload, service.config["server"]["max_questions"]) except UnsupportedRequest: self.close_connection=True self.respond(422,{"error":"Unsupported systemone request"}) return except (ValueError, TypeError, UnicodeError, OSError): self.close_connection=True self.respond(400, {"error": "Invalid systemone request"}) return # Failures never produce a substitute answer: 422 for one item, 503 for the service. try: result = service.answer(payload) json.dumps(result, allow_nan=False) except ItemFailure as exc: print("item failure: "+type(exc.__cause__ or exc).__name__, flush=True) self.respond(422, {"error": "item could not be processed"}) return except ServiceFailure as exc: print("service failure: "+str(exc), flush=True) self.respond(503, {"error": "service failure"}) return except Exception as exc: print("service failure: "+type(exc).__name__, flush=True) self.respond(503, {"error": "service failure"}) return self.respond(200, result) return Handler def install_extensions(service, config): """Serving extensions named in config.json, installed on the running service in this order: item_deadline (per-decision deadline of the drafting pass, with the item rules of the `item_rules` block: the question types that may take the drafting pass and the low-margin choice re-read), then post_calibration (per-type calibration of the reported probabilities). A configuration without these blocks serves the unchanged service.""" if __package__: from . import backends, core, item_deadline, post_calibration else: import backends, core, item_deadline, post_calibration if config.get('item_deadline') is not None: rules = config.get('item_rules') or {} types = rules.get('slow_types') service.warmup() # the drafting speed model is timed on a warmed engine, as in the measured setup item_deadline.install(core, backends, service.backend, config['item_deadline'], system_prompt=config['prompt']['system']).wrap_service( service, slow_types=None if types is None or set(types) == {'choice', 'noul', 'score'} else types, reread_tau=rules.get('reread_tau') or None) if config.get('post_calibration') is not None: post_calibration.wrap_service(service, config['post_calibration'], core) return service class QuietHTTPServer(ThreadingHTTPServer): # Request workers must not keep a stopped service alive while draining. daemon_threads = True def handle_error(self, request, client_address): # The base implementation prints a peer endpoint on disconnected clients. print("HTTP connection failed", flush=True) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--config", type=Path, default=ROOT / "config.json") parser.add_argument("--model") parser.add_argument("--quant", choices=("int8", "bf16", "awq")) parser.add_argument("--adapter") parser.add_argument("--backend", choices=("transformers", "vllm")) parser.add_argument("--bind", choices=BINDS, help="Listening address; defaults to server.bind in config") parser.add_argument("--port", type=int) parser.add_argument("--mode", choices=("routed", "uniform", "fast_only")) parser.add_argument("--calibration", type=Path) parser.add_argument('--state-dir', type=Path, help='Absolute directory for run state (required)') parser.add_argument('--run-id') parser.add_argument('--resume', action='store_true') parser.add_argument('--model-key') parser.add_argument('--quota-tier',choices=('normal','hard','long')) parser.add_argument('--cost-fuse',choices=('on','off')) parser.add_argument('--no-prefix-caching',action='store_true') parser.add_argument('--allow-placeholder-threshold',action='store_true') args = parser.parse_args(argv) if args.state_dir is None or not args.state_dir.is_absolute(): parser.error("--state-dir must be given as an absolute path") c = load_json(args.config) for arg, field in (("model", "path"), ("quant", "quant"), ("adapter", "adapter"), ("backend", "backend")): if getattr(args, arg) is not None: c["model"][field] = getattr(args, arg) for key in ("bind", "port"): if getattr(args, key) is not None: c["server"][key] = getattr(args, key) if args.mode is not None: c["routing"]["mode"] = args.mode if args.state_dir is not None: c['routing']['state_dir'] = str(args.state_dir) if args.run_id: c['routing']['run_id'] = args.run_id c['routing']['resume'] = args.resume if args.model_key: c['model']['model_key'] = args.model_key apply_profile(c) if __package__: from .core import apply_budget_tier else: from core import apply_budget_tier apply_budget_tier(c,args.quota_tier) if args.mode is not None:c["routing"]["mode"]=args.mode if args.quant is not None and args.quant!=c['model']['quant']:parser.error('Precision conflicts with model profile') if args.backend is not None and args.backend!=c['model']['backend']:parser.error('Backend conflicts with model profile') if args.no_prefix_caching: c["model"]["enable_prefix_caching"]=False if args.cost_fuse:c['routing']['cost_fuse']['enabled']=args.cost_fuse=='on' validate_config(c) validate_threshold(c,args.allow_placeholder_threshold) if not c["model"]["path"] or not c["server"]["bind"]: parser.error("Set --model and --bind, or their config fields") calibration = load_json(args.calibration or ROOT / c["calibration_file"]) validate_calibration(calibration, c["model"]["model_key"]) if __package__: from .backends import create_backend from .quota import RuntimeCounter else: from backends import create_backend from quota import RuntimeCounter counter = RuntimeCounter(c) service = Service(c, calibration, create_backend(c), counter, allow_placeholder_threshold=args.allow_placeholder_threshold) install_extensions(service, c) if c["server"].get("warmup", True): service.warmup() httpd = QuietHTTPServer((c["server"]["bind"], c["server"]["port"]), make_handler(service)) print("systemone service ready", flush=True) try: httpd.serve_forever() except KeyboardInterrupt: pass finally: httpd.server_close() if __name__ == "__main__": main()