| 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, | |
| ) | |