| """JSONL and local HTTP entry points sharing the supported decision engine.""" |
| import argparse |
| from contextlib import contextmanager |
| from datetime import datetime,timezone |
| from http.server import BaseHTTPRequestHandler,ThreadingHTTPServer |
| import json |
| from pathlib import Path |
| import signal,sys,threading |
| import urllib.error |
| from .backend import Client,NativeServer |
| from .decision import DecisionEngine |
|
|
| CAPABILITIES=dict(api_version='1',input_fields=['state','question','options'],min_options=2,max_options=26, |
| option_ids='opaque',numeric_values='parsed from option descriptions only',reasoning=False, |
| modalities=['text'],agent_execution='experimental; not exposed by this HTTP service', |
| compatibility='This is a decision API, not complete Jev/OpenJev API compatibility.') |
|
|
|
|
| def server(engine,host='0.0.0.0',port=9304): |
| slots=threading.BoundedSemaphore(2) |
| class Handler(BaseHTTPRequestHandler): |
| def log_message(self,*args):pass |
| def send_json(self,status,payload): |
| data=json.dumps(payload,ensure_ascii=False).encode() |
| self.send_response(status);self.send_header('Content-Type','application/json');self.send_header('Content-Length',str(len(data)));self.end_headers() |
| try:self.wfile.write(data) |
| except (BrokenPipeError,ConnectionResetError):pass |
| def do_GET(self): |
| if self.path=='/health': |
| try: |
| engine.client.health() |
| self.send_json(200,dict(status='ready',profile='base' if engine.client.adapter_id is None else 'adapter')) |
| except (OSError,ValueError):self.send_json(503,dict(status='backend-unavailable')) |
| elif self.path=='/v1/capabilities':self.send_json(200,CAPABILITIES) |
| else:self.send_json(404,dict(error='Unknown endpoint')) |
| def do_POST(self): |
| if self.path!='/v1/decide':self.send_json(404,dict(error='Unknown endpoint'));return |
| if not slots.acquire(blocking=False):self.send_json(503,dict(error='Decision worker and bounded queue are busy'));return |
| try: |
| self.connection.settimeout(125) |
| length=int(self.headers.get('Content-Length','0')) |
| if not 0<length<=65536:self.send_json(413,dict(error='Provide a JSON body of at most 64 KiB'));return |
| data=self.rfile.read(length) |
| if len(data)!=length:raise ValueError('Incomplete request body') |
| request=json.loads(data) |
| self.send_json(200,engine.decide(request)) |
| except (ValueError,UnicodeError,TypeError) as exc:self.send_json(400,dict(error=str(exc))) |
| except (OSError,RuntimeError,KeyError,IndexError) as exc:self.send_json(502,dict(error='Backend decision failed',detail=str(exc))) |
| finally:slots.release() |
| return ThreadingHTTPServer((host,port),Handler) |
|
|
|
|
| def arguments(description): |
| p=argparse.ArgumentParser(description=description) |
| p.add_argument('--base',default='http://127.0.0.1:5991');p.add_argument('--profile',choices=['base','current'],default='base') |
| p.add_argument('--adapter-id',type=int,default=0);p.add_argument('--launch',action='store_true') |
| p.add_argument('--model',default='models/Ternary-Bonsai-2-27B-PQ2_0.gguf') |
| p.add_argument('--adapter',help='Explicit adapter path required to launch profile current') |
| p.add_argument('--binary',default='llama.cpp-b2/build/bin/llama-server');p.add_argument('--gpu',type=int,default=0) |
| p.add_argument('--native-port',type=int,default=5991);p.add_argument('--run-dir') |
| return p |
|
|
|
|
| @contextmanager |
| def engine_for(args): |
| if args.launch: |
| if args.profile=='current' and not args.adapter:raise ValueError('Current profile requires an explicit --adapter path; its weight license differs from the base') |
| if args.profile=='base' and args.adapter:raise ValueError('Base profile does not load adapters') |
| if args.adapter_id!=0:raise ValueError('Launching one adapter requires adapter ID 0') |
| run=Path(args.run_dir) if args.run_dir else Path('runs')/datetime.now(timezone.utc).strftime('%Y%m%dT%H%M%S.%fZ') |
| run.mkdir(parents=True,exist_ok=False) |
| with NativeServer(run,args.gpu,args.native_port,args.model,args.adapter if args.profile=='current' else None,args.binary) as client: |
| yield DecisionEngine(client) |
| else: |
| yield DecisionEngine(Client(args.base,None if args.profile=='base' else args.adapter_id)) |
|
|
|
|
| @contextmanager |
| def graceful_termination(): |
| """Let service-manager termination unwind owned backend contexts.""" |
| def interrupt(signum,frame):raise KeyboardInterrupt |
| previous=signal.signal(signal.SIGTERM,interrupt) |
| try:yield |
| finally:signal.signal(signal.SIGTERM,previous) |
|
|
|
|
| def cli(): |
| p=arguments('Bonsai text-choice decisions: JSONL input and JSONL output.');p.add_argument('--input');args=p.parse_args() |
| try: |
| with graceful_termination(),engine_for(args) as engine: |
| stream=open(args.input) if args.input else sys.stdin |
| try: |
| for line in stream: |
| if not line.strip():continue |
| try:result=engine.decide(json.loads(line)) |
| except (ValueError,TypeError,OSError,RuntimeError) as exc:result=dict(error=str(exc)) |
| print(json.dumps(result),flush=True) |
| finally: |
| if args.input:stream.close() |
| except KeyboardInterrupt:pass |
| except (ValueError,RuntimeError) as exc:p.error(str(exc)) |
|
|
|
|
| def http_cli(): |
| p=arguments('Serve the Bonsai decision API.');p.add_argument('--host',default='0.0.0.0');p.add_argument('--port',type=int,default=9304);args=p.parse_args() |
| try: |
| with graceful_termination(),engine_for(args) as engine: |
| httpd=server(engine,args.host,args.port) |
| print(json.dumps(dict(listening=f'http://{args.host}:{args.port}',profile=args.profile)),file=sys.stderr,flush=True) |
| try:httpd.serve_forever(poll_interval=.2) |
| except KeyboardInterrupt:pass |
| finally:httpd.server_close() |
| except KeyboardInterrupt:pass |
| except (ValueError,RuntimeError) as exc:p.error(str(exc)) |
|
|
|
|
| if __name__=='__main__':cli() |
|
|
