server.py

"""Local Switchyard route designer; Python 3.11+, no third-party dependencies."""
import json
import os
import re
import ssl
import sys
import atexit
import signal
import socket
import threading
import time
import subprocess
import tempfile
import tomllib
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

ROOT = Path(__file__).parent


def gateway_opener():
    """Use trusted macOS keychain certificates without disabling TLS verification."""
    context = ssl.create_default_context()
    if sys.platform == 'darwin':
        for keychain in ('/System/Library/Keychains/SystemRootCertificates.keychain',
                         '/Library/Keychains/System.keychain'):
            result = subprocess.run(['/usr/bin/security', 'find-certificate', '-a', '-p', keychain],
                                    capture_output=True, text=True, timeout=10)
            if result.returncode == 0 and result.stdout.strip():
                context.load_verify_locations(cadata=result.stdout)
    return urllib.request.build_opener(NoRedirect, urllib.request.HTTPSHandler(context=context))


def defaults():
    """Discover connection settings locally without returning any credential."""
    url = os.environ.get('OPENAI_BASE_URL', '')
    name = 'OPENAI_API_KEY'
    local = ROOT.parent / 'Switchyard' / 'dev-server' / 'config.toml'
    if not url and local.is_file():
        try:
            clients = tomllib.loads(local.read_text()).get('llm_clients', {})
            client = next(iter(clients.values()), {})
            url = client.get('base_url', '')
            name = client.get('api_key_env', name)
        except (ValueError, OSError):
            pass
    binary = ROOT.parent / 'Switchyard' / 'target' / 'debug' / 'switchyard-server'
    return {'url': url, 'env': name, 'has_key': bool(os.environ.get(name)),
            'binary': str(binary) if binary.is_file() else ''}


def config(data):
    """Render a single route deployment with validated inputs and TOML-safe strings."""
    url = data.get('url', '').rstrip('/')
    parsed = urllib.parse.urlsplit(url)
    if parsed.scheme not in ('http', 'https') or not parsed.hostname or parsed.username or parsed.password or parsed.query or parsed.fragment:
        raise ValueError('Enter an HTTP(S) gateway base URL without credentials, query, or fragment.')
    env = data.get('env', 'OPENAI_API_KEY')
    if not re.fullmatch(r'[A-Za-z_][A-Za-z0-9_]*', env):
        raise ValueError('Invalid API key environment variable name.')
    algorithm = data.get('algorithm')
    roles = {'ensemble': ['candidate_a', 'candidate_b', 'synthesizer'],
             'escalation': ['efficient', 'capable', 'judge'],
             'capability': ['efficient', 'capable', 'judge'],
             'passthrough': ['model']}.get(algorithm)
    if not roles:
        raise ValueError('Unknown algorithm.')
    quote = lambda value: json.dumps(value, ensure_ascii=False)
    route = data.get('route', '').strip()
    if not route:
        raise ValueError('Route model ID is required.')
    wire = data.get('format', 'openai_responses')
    if wire not in ('openai_responses', 'openai_chat', 'anthropic_messages'):
        raise ValueError('Unsupported wire format.')
    lines = ['schema_version = 1', '', '[llm_clients.gateway]', f'format = {quote(wire)}',
             f'base_url = {quote(url)}', f'api_key_env = {quote(env)}']
    for role in roles:
        model = data.get('models', {}).get(role, '').strip()
        if not model:
            raise ValueError(f'Select a model for {role.replace("_", " ")}.')
        lines += ['', f'[targets.{role}]', f'id = {quote(model)}', 'llm_client = "gateway"']
    lines += ['', '[routes.designed]', f'id = {quote(route)}']
    if algorithm == 'ensemble':
        cap = int(data.get('tokens', 2048))
        minimum = int(data.get('minimum', 2))
        if not 1 <= cap <= 1000000 or minimum not in (1, 2):
            raise ValueError('Invalid ensemble budget or minimum candidate count.')
        lines += ['type = "ensemble"', 'candidates = ["candidate_a", "candidate_b"]',
                  'synthesizer_target = "synthesizer"', f'candidate_max_output_tokens = {cap}',
                  f'minimum_successful_candidates = {minimum}']
    elif algorithm == 'passthrough':
        lines += ['type = "passthrough"', 'target = "model"']
    else:
        lines += ['type = "llm_classifier"', f'mode = {quote(algorithm)}',
                  'weak_target = "efficient"', 'strong_target = "capable"', 'classifier_target = "judge"']
        if algorithm == 'escalation':
            confirmations = int(data.get('confirmations', 2))
            if not 1 <= confirmations <= 100:
                raise ValueError('Confirmations must be between 1 and 100.')
            lines += ['', '[routes.designed.escalation]', f'confirmations = {confirmations}']
        else:
            threshold = float(data.get('threshold', .5))
            if not 0 <= threshold <= 1:
                raise ValueError('Threshold must be between zero and one.')
            lines += [f'base_threshold = {threshold}']
    result = '\n'.join(lines) + '\n'
    tomllib.loads(result)
    return result


class NoRedirect(urllib.request.HTTPRedirectHandler):
    """Never forward credentials through a gateway redirect."""
    def redirect_request(self, *args, **kwargs):
        return None


class RouterRuntime:
    """Own only the native router launched by this dashboard."""
    def __init__(self):
        self.lock = threading.RLock()
        self.process = None
        self.directory = None
        self.log = None
        self.port = None
        self.route = None
        self.opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect)

    def status(self):
        with self.lock:
            running = self.process is not None and self.process.poll() is None
            return {'running': running, 'route': self.route,
                    'base_url': f'http://127.0.0.1:{self.port}/v1' if running else None}

    def stop(self):
        with self.lock:
            if self.process is not None and self.process.poll() is None:
                self.process.terminate()
                try:
                    self.process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    self.process.kill()
                    self.process.wait()
            if self.log:
                self.log.close()
            if self.directory:
                self.directory.cleanup()
            self.process = self.directory = self.log = self.port = self.route = None
            return self.status()

    def start(self, data):
        rendered = config(data)
        binary = Path(data.get('binary', '')).expanduser()
        if not binary.is_absolute() or not binary.is_file() or not os.access(binary, os.X_OK):
            raise ValueError('No built Switchyard server found. Select its executable under runtime settings.')
        key = data.get('key', '').strip() or os.environ.get(data['env'], '')
        if not key:
            raise ValueError('Paste your gateway API key before starting the router.')
        with self.lock:
            if self.status()['running']:
                raise ValueError('Stop the current router before applying different settings.')
            self.stop()
            self.directory = tempfile.TemporaryDirectory(prefix='switchyard-studio-runtime-')
            path = Path(self.directory.name) / 'routes.toml'
            path.write_text(rendered)
            # Temporary port selection; readiness also requires this child to stay alive.
            with socket.socket() as listener:
                listener.bind(('127.0.0.1', 0))
                self.port = listener.getsockname()[1]
            self.log = tempfile.TemporaryFile(mode='w+b')
            self.route = data['route'].strip()
            try:
                self.process = subprocess.Popen([str(binary), '--config', str(path), '--host', '127.0.0.1',
                                                 '--port', str(self.port), '--shutdown-timeout', '2s'],
                                                stdout=self.log, stderr=self.log,
                                                env={**os.environ, data['env']: key})
                deadline = time.monotonic() + 12
                while time.monotonic() < deadline:
                    if self.process.poll() is not None:
                        self.log.seek(0)
                        reason = self.log.read()[-6000:].decode(errors='replace').replace(key, '[redacted]')
                        raise ValueError('Router failed to start: ' + reason)
                    try:
                        with self.opener.open(f'http://127.0.0.1:{self.port}/v1/models', timeout=.5) as response:
                            models = json.load(response)
                        if self.route in [m['id'] for m in models.get('data', [])] and self.process.poll() is None:
                            return self.status()
                    except (OSError, ValueError):
                        pass
                    time.sleep(.1)
                raise ValueError('Router did not become ready within 12 seconds.')
            except Exception:
                self.stop()
                raise

    def call(self, data):
        with self.lock:
            state = self.status()
            if not state['running']:
                raise ValueError('Start the router before sending a request.')
        messages = data.get('messages')
        if not isinstance(messages, list) or not messages or len(messages) > 100:
            raise ValueError('Provide a conversation with 1–100 messages.')
        for message in messages:
            if not isinstance(message, dict) or message.get('role') not in ('user', 'assistant') or not isinstance(message.get('content'), str):
                raise ValueError('Conversation messages must contain a user/assistant role and text content.')
        tokens = int(data.get('max_tokens', 2048))
        if not 1 <= tokens <= 32768:
            raise ValueError('Output budget must be between 1 and 32768.')
        session = data.get('session', '')
        if not re.fullmatch(r'[A-Za-z0-9_-]{1,100}', session):
            raise ValueError('Invalid session ID.')
        body = {'model': state['route'], 'messages': messages, 'max_completion_tokens': tokens, 'stream': False}
        request = urllib.request.Request(state['base_url'] + '/chat/completions', data=json.dumps(body).encode(),
                                         headers={'Content-Type': 'application/json', 'x-switchyard-session-id': session})
        started = time.monotonic()
        try:
            with self.opener.open(request, timeout=300) as response:
                payload = json.load(response)
                served = response.headers.get('x-model-router-selected-model', payload.get('model'))
        except urllib.error.HTTPError as error:
            # Do not expose raw upstream error bodies, which may include credentials.
            raise ValueError(f'Router returned HTTP {error.code}. Check model permissions and gateway settings; no successful response was received.') from error
        return {'response': payload, 'served_model': served, 'seconds': round(time.monotonic() - started, 2)}


runtime = RouterRuntime()
atexit.register(runtime.stop)


class Handler(BaseHTTPRequestHandler):
    """Serve only this app and same-origin JSON actions on loopback."""
    def log_message(self, *args):
        pass

    def send(self, status, payload, mime='application/json'):
        body = json.dumps(payload).encode() if mime == 'application/json' else payload
        self.send_response(status)
        self.send_header('Content-Type', mime)
        self.send_header('Cache-Control', 'no-store')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.send_header('Content-Security-Policy', "default-src 'self'; style-src 'self'; script-src 'self'; connect-src 'self'; frame-ancestors 'none'")
        self.end_headers()
        self.wfile.write(body)

    def allowed(self):
        return self.headers.get('Host') == f'127.0.0.1:{self.server.server_port}'

    def do_GET(self):
        if not self.allowed():
            return self.send(403, {'error': 'Open this app through its printed loopback URL.'})
        if self.path == '/api/defaults':
            return self.send(200, defaults())
        if self.path == '/api/status':
            return self.send(200, runtime.status())
        files = {'/': ('index.html', 'text/html'), '/app.js': ('app.js', 'text/javascript'), '/style.css': ('style.css', 'text/css')}
        if self.path not in files:
            return self.send(404, {'error': 'Not found'})
        name, mime = files[self.path]
        self.send(200, (ROOT / name).read_bytes(), mime)

    def do_POST(self):
        if not self.allowed() or self.headers.get('Origin') != f'http://127.0.0.1:{self.server.server_port}' or self.headers.get('Content-Type') != 'application/json':
            return self.send(403, {'error': 'Same-origin JSON requests required.'})
        try:
            length = int(self.headers.get('Content-Length', 0))
            if not 0 < length <= 65536:
                raise ValueError('Invalid request size.')
            data = json.loads(self.rfile.read(length))
            if self.path == '/api/start':
                return self.send(200, runtime.start(data))
            if self.path == '/api/stop':
                return self.send(200, runtime.stop())
            if self.path == '/api/chat':
                return self.send(200, runtime.call(data))
            if self.path == '/api/models':
                # Validate connection fields using a minimal route before any network request.
                config({**data, 'algorithm': 'passthrough', 'route': 'probe', 'models': {'model': 'probe'}})
                key = data.get('key', '').strip() or os.environ.get(data.get('env', 'OPENAI_API_KEY'), '')
                if not key:
                    raise ValueError('Paste your API key into the connection field.')
                request = urllib.request.Request(data['url'].rstrip('/') + '/models', headers={'Authorization': 'Bearer ' + key})
                with gateway_opener().open(request, timeout=20) as response:
                    payload = json.load(response)
                models = sorted({item['id'] for item in payload['data'] if isinstance(item.get('id'), str)})
                return self.send(200, {'models': models})
            rendered = config(data)
            if self.path == '/api/config':
                return self.send(200, {'toml': rendered})
            if self.path == '/api/validate':
                binary = Path(data.get('binary', '')).expanduser()
                if not binary.is_absolute() or not binary.is_file() or not os.access(binary, os.X_OK):
                    raise ValueError('Select an absolute path to a built switchyard-server executable.')
                with tempfile.TemporaryDirectory(prefix='switchyard-dashboard-') as directory:
                    path = Path(directory) / 'routes.toml'
                    path.write_text(rendered)
                    env = dict(os.environ)
                    env.setdefault(data['env'], 'validation-only')
                    result = subprocess.run([str(binary), '--config', str(path), '--dry-run'], capture_output=True, text=True, timeout=30, env=env)
                output = (result.stdout + result.stderr)[-12000:]
                key = os.environ.get(data['env'])
                if key:
                    output = output.replace(key, '[redacted]')
                return self.send(200, {'ok': result.returncode == 0, 'output': output})
            self.send(404, {'error': 'Not found'})
        except urllib.error.HTTPError as error:
            self.send(400, {'error': f'Gateway returned HTTP {error.code}. Check the URL, credentials, and model access.'})
        except urllib.error.URLError as error:
            if isinstance(error.reason, ssl.SSLCertVerificationError):
                message = 'Gateway TLS certificate is not trusted. Check your system certificates or corporate VPN certificate installation.'
            else:
                message = 'Cannot reach the gateway. Check the gateway URL, VPN, and network connection.'
            self.send(400, {'error': message})
        except subprocess.TimeoutExpired:
            self.send(400, {'error': 'The operation timed out. Check the selected build or gateway connection.'})
        except (ValueError, KeyError, TypeError, OSError, subprocess.TimeoutExpired) as error:
            self.send(400, {'error': str(error) if isinstance(error, ValueError) else 'Operation failed. Check connection settings and executable path.'})


if __name__ == '__main__':
    signal.signal(signal.SIGTERM, lambda *_: sys.exit(0))
    server = ThreadingHTTPServer(('127.0.0.1', 4780), Handler)
    print('Switchyard Studio: http://127.0.0.1:4780', flush=True)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        server.server_close()