File size: 5,642 Bytes
b1b9014 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 | import hmac
import json
import math
from urllib.parse import urlsplit
from workers import Response, WorkerEntrypoint
MAX_BODY_CHARS = 1_000_000
MAX_PROMPT_CHARS = 16_000
ROUTE_OPERATIONS = {
"/subconscious": "subconscious.autonomous_step",
"/conscious": "conscious.deep_surgery",
}
def _json_response(data, status=200, origin="*"):
return Response(
json.dumps(data, separators=(",", ":"), allow_nan=False),
status=status,
headers={
"Content-Type": "application/json; charset=utf-8",
"Access-Control-Allow-Origin": origin,
"Access-Control-Allow-Methods": "POST, OPTIONS, GET",
"Access-Control-Allow-Headers": "Content-Type, X-API-KEY",
"Vary": "Origin",
"Cache-Control": "no-store",
},
)
def _bounded_number(payload, key, default):
value = payload.get(key, default)
if isinstance(value, bool) or not isinstance(value, (int, float)):
return None
if not math.isfinite(value) or value < 0 or value > 1:
return None
return value
def _validate_payload(route, payload):
if not isinstance(payload, dict):
return "Request body must be a JSON object."
expected_operation = ROUTE_OPERATIONS[route]
if payload.get("op") != expected_operation:
return "Request operation does not match the requested route."
text_key = "stimulus" if route == "/subconscious" else "prompt"
text = payload.get(text_key)
if not isinstance(text, str) or not text.strip():
return f"'{text_key}' must be a non-empty string."
if len(text) > MAX_PROMPT_CHARS:
return f"'{text_key}' exceeds the {MAX_PROMPT_CHARS}-character limit."
number_keys = (
("friction",)
if route == "/subconscious"
else ("prediction_error", "rho_virtue")
)
for key in number_keys:
if _bounded_number(payload, key, 0.5) is None:
return f"'{key}' must be a finite number between 0 and 1."
return None
class Default(WorkerEntrypoint):
async def fetch(self, request):
origin = getattr(self.env, "CORS_ORIGIN", "*") or "*"
method = request.method.upper()
path = urlsplit(request.url).path.rstrip("/") or "/"
if method == "OPTIONS":
return Response(
"",
status=204,
headers={
"Access-Control-Allow-Origin": origin,
"Access-Control-Allow-Methods": "POST, OPTIONS, GET",
"Access-Control-Allow-Headers": "Content-Type, X-API-KEY",
"Access-Control-Max-Age": "86400",
"Vary": "Origin",
},
)
if path == "/health":
if method != "GET":
return _json_response({"error": "Method not allowed."}, 405, origin)
return _json_response(
{
"status": "ok",
"routes": list(ROUTE_OPERATIONS),
"core_binding_configured": getattr(self.env, "XELA_CORE", None) is not None,
},
origin=origin,
)
if path not in ROUTE_OPERATIONS:
return _json_response({"error": "Route not found."}, 404, origin)
if method != "POST":
return _json_response({"error": "Method not allowed."}, 405, origin)
configured_key = getattr(self.env, "XELA_API_KEY", None)
if configured_key:
supplied_key = request.headers.get("X-API-KEY", "")
if not isinstance(supplied_key, str) or not hmac.compare_digest(
supplied_key, str(configured_key)
):
return _json_response({"error": "Unauthorized."}, 401, origin)
try:
body_text = await request.clone().text()
if len(body_text) > MAX_BODY_CHARS:
return _json_response({"error": "Request body is too large."}, 413, origin)
payload = json.loads(body_text)
except (ValueError, TypeError):
return _json_response({"error": "Request body must contain valid JSON."}, 400, origin)
validation_error = _validate_payload(path, payload)
if validation_error:
return _json_response({"error": validation_error}, 400, origin)
core = getattr(self.env, "XELA_CORE", None)
if core is None:
return _json_response(
{"error": "XELA_CORE service binding is not configured."},
503,
origin,
)
try:
upstream = await core.fetch(request)
content_type = upstream.headers.get("Content-Type") or "application/json"
response_body = await upstream.text()
return Response(
response_body,
status=upstream.status,
headers={
"Content-Type": content_type,
"Access-Control-Allow-Origin": origin,
"Access-Control-Allow-Methods": "POST, OPTIONS, GET",
"Access-Control-Allow-Headers": "Content-Type, X-API-KEY",
"Vary": "Origin",
"Cache-Control": "no-store",
},
)
except Exception:
return _json_response(
{"error": "The XELA core service could not be reached."},
502,
origin,
)
|