"""HTTP server for the Nexum runtime.""" from __future__ import annotations import argparse import base64 import hashlib import hmac import json import os import secrets import select import socket import threading import time import uuid import urllib.parse from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from typing import Any from .bundle import release_artifact_sha256, validate_bundle from .capabilities import capability_catalog from .executor import execute_tool_calls, execute_tool_text, list_tools from .harness import ( NNFXHarnessRequest, NNFXHarnessSession, external_execution_result, ) from .identity import candidate_identity_sha256, runtime_source_sha256 from .protocols import a2a_request, agent_card, mcp_request from .runner import NexumLocalRunner, _runtime_state_root, runtime_state_namespace from .status import SELF_CORRECTION_STATUS, SELF_IMPROVEMENT_STATUS, runtime_health from .templates import TEMPLATES from .tooling.artifacts import ArtifactStore from .tooling.contracts import tool_call_from_openai from .tooling.events import EventLog from .tooling.language_packs import language_pack_catalog from .tooling.security import ApprovalStore from .tooling.tasks import TaskStore from .visibility import private_reasoning_content, visible_completion _RUNNER: NexumLocalRunner | None = None _RUNNER_LOAD_THREAD: threading.Thread | None = None _RUNNER_LOAD_ERROR = "" _RUNNER_LOAD_LOCK = threading.Lock() def _runtime_instance_sha256( release_sha256: str, runtime_source_sha256: str, state_namespace: str, ) -> str: canonical = json.dumps( { "release_artifact_sha256": release_sha256, "runtime_source_sha256": runtime_source_sha256, "state_namespace": state_namespace, "instance_nonce": secrets.token_hex(32), }, ensure_ascii=True, separators=(",", ":"), sort_keys=True, ).encode("utf-8") return hashlib.sha256(canonical).hexdigest() def _runner(model_dir: str, device: str) -> NexumLocalRunner: global _RUNNER if not model_dir: raise RuntimeError("model directory is required") resolved = Path(model_dir).expanduser().resolve() state_namespace = runtime_state_namespace() state_root = _runtime_state_root() if ( _RUNNER is None or _RUNNER.model_dir.expanduser().resolve() != resolved or _RUNNER.device != device or _RUNNER.state_namespace != state_namespace or _RUNNER.state_root != state_root ): _RUNNER = NexumLocalRunner(resolved, device=device) return _RUNNER def _runner_load_state() -> dict[str, Any]: thread = _RUNNER_LOAD_THREAD return { "loading": bool(thread is not None and thread.is_alive()), "load_error": _RUNNER_LOAD_ERROR, } def _start_runner_load(runner: NexumLocalRunner) -> None: """Begin the same full model load without blocking HTTP health surfaces.""" global _RUNNER_LOAD_ERROR, _RUNNER_LOAD_THREAD with _RUNNER_LOAD_LOCK: if runner.status().get("loaded"): _RUNNER_LOAD_ERROR = "" return if _RUNNER_LOAD_THREAD is not None and _RUNNER_LOAD_THREAD.is_alive(): return _RUNNER_LOAD_ERROR = "" def load_target() -> None: global _RUNNER_LOAD_ERROR try: runner.load() except Exception as exc: # noqa: BLE001 - status surface owns startup failure _RUNNER_LOAD_ERROR = f"{type(exc).__name__}: {exc}" _RUNNER_LOAD_THREAD = threading.Thread( target=load_target, name="nexum-runtime-load", daemon=True, ) _RUNNER_LOAD_THREAD.start() def _responses_reasoning_text(item: dict[str, Any]) -> str: """Recover one Responses API reasoning item for the next assistant turn.""" for field, accepted_type in ( ("content", "reasoning_text"), ("summary", "summary_text"), ): parts = item.get(field) if not isinstance(parts, list): continue text = "\n".join( str(part.get("text") or "").strip() for part in parts if isinstance(part, dict) and part.get("type") == accepted_type and str(part.get("text") or "").strip() ) if text: return text return "" def _responses_chat_payload(payload: dict[str, Any]) -> dict[str, Any]: messages: list[dict[str, Any]] = [] pending_reasoning: list[str] = [] def take_pending_reasoning() -> str: text = "\n".join(pending_reasoning).strip() pending_reasoning.clear() return text def flush_pending_reasoning() -> None: reasoning = take_pending_reasoning() if reasoning: messages.append( { "role": "assistant", "content": "", "reasoning_content": reasoning, } ) instructions = payload.get("instructions") if isinstance(instructions, str) and instructions.strip(): messages.append({"role": "system", "content": instructions}) source = payload.get("input") if isinstance(source, str): messages.append({"role": "user", "content": source}) elif isinstance(source, list): for item in source: if not isinstance(item, dict): continue item_type = item.get("type") if item_type == "reasoning": reasoning = _responses_reasoning_text(item) if reasoning: pending_reasoning.append(reasoning) elif item_type == "function_call": message: dict[str, Any] = { "role": "assistant", "content": "", "tool_calls": [ { "id": str(item.get("call_id") or item.get("id") or ""), "type": "function", "function": { "name": str(item.get("name") or ""), "arguments": str(item.get("arguments") or "{}"), }, } ], } reasoning = take_pending_reasoning() if reasoning: message["reasoning_content"] = reasoning messages.append(message) elif item_type == "function_call_output": flush_pending_reasoning() call_id = str(item.get("call_id") or "") output = item.get("output", "") messages.append( { "role": "tool", "tool_call_id": call_id, "content": output, } ) elif item_type == "message" or isinstance(item.get("role"), str): role = str(item.get("role") or "user") if role != "assistant": flush_pending_reasoning() message = { "role": role, "content": item.get("content", ""), } if role == "assistant": reasoning_parts = list(pending_reasoning) explicit_reasoning = item.get("reasoning_content") if ( isinstance(explicit_reasoning, str) and explicit_reasoning.strip() ): reasoning_parts.append(explicit_reasoning.strip()) pending_reasoning.clear() reasoning = "\n".join(dict.fromkeys(reasoning_parts)).strip() if reasoning: message["reasoning_content"] = reasoning messages.append(message) flush_pending_reasoning() conversation = payload.get("conversation") if isinstance(conversation, dict): conversation = conversation.get("id") chat = { "model": str(payload.get("model") or "Nexum"), "messages": messages, "tools": payload.get("tools"), "temperature": float(payload.get("temperature") or 0.0), "session_id": str( conversation or payload.get("session_id") or uuid.uuid4().hex ), } maximum = payload.get("max_output_tokens") if maximum not in (None, ""): chat["max_completion_tokens"] = int(str(maximum)) return chat def _responses_caller_observations( payload: dict[str, Any], ) -> list[dict[str, Any]]: """Correlate caller-owned function outputs to exact selected actions.""" source = payload.get("input") if not isinstance(source, list): return [] selected_calls: dict[str, dict[str, Any]] = {} resolved_ids: set[str] = set() observations: list[dict[str, Any]] = [] failure_statuses = {"cancelled", "error", "failed", "incomplete", "rejected"} for item in source: if not isinstance(item, dict): continue item_type = item.get("type") if item_type == "function_call": call_id = str(item.get("call_id") or item.get("id") or "").strip() if not call_id: raise ValueError("Responses function call has no identifier") if call_id in selected_calls: raise ValueError("Responses function call identifiers are duplicated") raw_call = { "id": call_id, "type": "function", "function": { "name": str(item.get("name") or ""), "arguments": item.get("arguments") or "{}", }, } tool_call_from_openai(raw_call) selected_calls[call_id] = raw_call continue if item_type != "function_call_output": continue call_id = str(item.get("call_id") or "").strip() pending_call = selected_calls.get(call_id) if pending_call is None: continue if call_id in resolved_ids: raise ValueError("Responses function output was supplied more than once") resolved_ids.add(call_id) selected = tool_call_from_openai(pending_call) raw_output = item.get("output", "") rendered_output = ( raw_output if isinstance(raw_output, str) else json.dumps(raw_output, sort_keys=True, default=str) ) try: decoded = json.loads(rendered_output) except json.JSONDecodeError: decoded = None structured = decoded if isinstance(decoded, dict) else {} status = str( structured.get("status") or item.get("status") or "completed" ).lower() error = str(structured.get("error") or item.get("error") or "") explicit_ok = structured.get("ok") ok = ( explicit_ok if isinstance(explicit_ok, bool) else not error and status not in failure_statuses ) explicit_executed = structured.get("executed") executed = ( explicit_executed if isinstance(explicit_executed, bool) else status not in {"cancelled", "rejected"} ) result = external_execution_result( { "tool_call_id": call_id, "name": structured.get("name", structured.get("tool", selected.name)), "args": structured.get("args", selected.args), "ok": ok, "executed": executed, "output": str(structured.get("output", rendered_output)), "stdout": str(structured.get("stdout") or ""), "stderr": str(structured.get("stderr") or ""), "error": error, "exit_code": structured.get("exit_code", structured.get("return_code")), "elapsed_s": structured.get("elapsed_s", 0.0), "status": status, }, pending_call, ) observations.append( { **result.to_dict(), "receipt_source": "caller_attested", "source_trust": "caller_owned", } ) return observations def _responses_chat(runner: Any, payload: dict[str, Any]) -> dict[str, Any]: """Run one Responses turn with authenticated caller-owned tool evidence.""" chat_payload = _responses_chat_payload(payload) observations = _responses_caller_observations(payload) if observations: session_id = str(chat_payload["session_id"]) chat_payload["nexum_observations"] = [ runner.sign_observation(session_id, observation) for observation in observations ] result = runner.chat(chat_payload) if not isinstance(result, dict): raise TypeError("Responses runner must return a completion object") return result def _session_bound_chat_result( chat: dict[str, Any], *, session_id: str, ) -> dict[str, Any]: """Return the visible completion with its reusable session identity.""" if not session_id: raise ValueError("chat session identity is required") projected = _visible_chat_result(chat) metadata = projected.get("nexum") if isinstance(metadata, dict): projected["nexum"] = {**metadata, "session_id": session_id} return projected def _responses_result(chat: dict[str, Any]) -> dict[str, Any]: choice = chat["choices"][0] message = choice["message"] output: list[dict[str, Any]] = [] reasoning = message.get("reasoning_content") if isinstance(reasoning, str) and reasoning.strip(): reasoning = reasoning.strip() output.append( { "type": "reasoning", "id": "rs_" + uuid.uuid4().hex, "summary": [{"type": "summary_text", "text": reasoning}], "content": [{"type": "reasoning_text", "text": reasoning}], "status": "completed", } ) calls = message.get("tool_calls") if isinstance(calls, list) and calls: for call in calls: function = call.get("function") if isinstance(call, dict) else None if not isinstance(function, dict): continue call_id = str(call.get("id") or "call_" + uuid.uuid4().hex) output.append( { "type": "function_call", "id": call_id, "call_id": call_id, "name": str(function.get("name") or ""), "arguments": str(function.get("arguments") or "{}"), "status": "completed", } ) else: output.append( { "type": "message", "id": "msg_" + uuid.uuid4().hex, "role": "assistant", "status": "completed", "content": [ { "type": "output_text", "text": str(message.get("content") or ""), "annotations": [], } ], } ) raw_usage = chat.get("usage") usage: dict[str, Any] = dict(raw_usage) if isinstance(raw_usage, dict) else {} return { "id": "resp_" + uuid.uuid4().hex, "object": "response", "created_at": int(time.time()), "status": "completed", "model": str(chat.get("model") or "Nexum"), "output": output, "usage": { "input_tokens": int(usage.get("prompt_tokens") or 0), "output_tokens": int(usage.get("completion_tokens") or 0), "total_tokens": int(usage.get("total_tokens") or 0), }, } def _visible_chat_result(chat: dict[str, Any]) -> dict[str, Any]: """Project visible text without mutating the private runner trajectory.""" projected = dict(chat) raw_choices = chat.get("choices") if not isinstance(raw_choices, list): return projected choices: list[Any] = [] for raw_choice in raw_choices: if not isinstance(raw_choice, dict): choices.append(raw_choice) continue choice = dict(raw_choice) raw_message = raw_choice.get("message") if isinstance(raw_message, dict): message = dict(raw_message) content = raw_message.get("content") if isinstance(content, str): message["content"] = visible_completion(content) reasoning = raw_message.get("reasoning_content") if not isinstance(reasoning, str) and isinstance(content, str): reasoning = private_reasoning_content(content) if isinstance(reasoning, str) and reasoning.strip(): message["reasoning_content"] = reasoning.strip() choice["message"] = message choices.append(choice) projected["choices"] = choices return projected def _json_response( handler: BaseHTTPRequestHandler, status: int, payload: dict[str, Any] ) -> None: data = json.dumps(payload, indent=2, sort_keys=True).encode("utf-8") try: handler.send_response(status) handler.send_header("Content-Type", "application/json") handler.send_header("Content-Length", str(len(data))) handler.end_headers() handler.wfile.write(data) except (BrokenPipeError, ConnectionResetError): return class NexumHandler(BaseHTTPRequestHandler): model_dir: str = "" workspace: str = "." device: str = "cuda:0" api_key: str = "" enable_tools: bool = False bundle_verified: bool = False release_artifact_sha256: str = "" runtime_source_sha256: str = "" candidate_identity_sha256: str = "" runtime_instance_sha256: str = "" state_namespace: str = "default" def log_message(self, fmt: str, *args: object) -> None: return def _read_payload(self) -> dict[str, Any]: length = int(self.headers.get("Content-Length") or "0") if length <= 0: return {} raw = self.rfile.read(length) value = json.loads(raw.decode("utf-8")) if not isinstance(value, dict): raise ValueError("payload must be an object") return value def _authorized(self) -> bool: if not self.api_key: return True headers = getattr(self, "headers", None) value = headers.get("Authorization", "") if headers is not None else "" scheme, separator, token = value.partition(" ") return ( bool(separator) and scheme.lower() == "bearer" and hmac.compare_digest(token.strip(), self.api_key) ) def _require_authorization(self) -> bool: if self._authorized(): return True _json_response( self, 401, {"ok": False, "error": "authorization_required"}, ) return False def _require_tool_execution(self) -> bool: if self.enable_tools: return True _json_response( self, 403, {"ok": False, "error": "server_tool_execution_disabled"}, ) return False def _workspace(self, requested: Any = None) -> str: root = Path(self.workspace).expanduser().resolve() target = Path(str(requested or root)).expanduser().resolve() try: target.relative_to(root) except ValueError as exc: raise ValueError( "requested workspace is outside the configured server workspace" ) from exc return str(target) def _client_disconnected(self) -> bool: try: readable, _, _ = select.select([self.connection], [], [], 0.0) if not readable: return False peeked: bytes = self.connection.recv( 1, socket.MSG_PEEK | socket.MSG_DONTWAIT ) return peeked == b"" except (BlockingIOError, OSError): return False def _route(self) -> tuple[str, dict[str, list[str]]]: parsed = urllib.parse.urlsplit(self.path) path = parsed.path.rstrip("/") or "/" return path, urllib.parse.parse_qs(parsed.query, keep_blank_values=True) def _base_url(self) -> str: forwarded = self.headers.get("X-Forwarded-Proto", "").strip().lower() scheme = forwarded if forwarded in {"http", "https"} else "http" host = self.headers.get("Host", "").strip() or "localhost" return f"{scheme}://{host}" def do_GET(self) -> None: try: self._do_GET() except (PermissionError, TypeError, ValueError) as exc: _json_response( self, 422, {"ok": False, "error": f"{type(exc).__name__}: {exc}"} ) except Exception as exc: _json_response(self, 500, {"ok": False, "error": type(exc).__name__}) def _do_GET(self) -> None: path, query = self._route() if path == "/health": runner_status = _RUNNER.status() if _RUNNER is not None else {} load_state = _runner_load_state() loaded = bool(runner_status.get("loaded")) payload = runtime_health( bundle_ok=self.bundle_verified, model_loaded=loaded and self.bundle_verified, self_correction_ready=bool(runner_status.get("self_correction_ready")), self_improvement_ready=bool( runner_status.get("self_improvement_ready") ), context_intent_action_ready=bool( runner_status.get("context_intent_action_ready") ), learning_generation=int(runner_status.get("learning_generation") or 0), ) payload.update(load_state) _json_response(self, 200 if payload["ready"] else 503, payload) return if path == "/.well-known/agent-card.json": _json_response(self, 200, agent_card(self._base_url())) return if not self._require_authorization(): return if path == "/release-status": runner_status = _RUNNER.status() if _RUNNER is not None else {} load_state = _runner_load_state() full_model_active = bool(runner_status.get("full_model_active")) tensor_packages_loaded = int( runner_status.get("tensor_packages_loaded") or 0 ) tokenizer_acceleration_ready = bool( runner_status.get("fast_tokenizer_active") ) self_correction_ready = bool(runner_status.get("self_correction_ready")) self_improvement_ready = bool(runner_status.get("self_improvement_ready")) context_intent_action_ready = bool( runner_status.get("context_intent_action_ready") ) context_intent_effect_ready = bool( runner_status.get("context_intent_effect_ready") ) context_action_effect_ready = bool( runner_status.get("context_action_effect_ready") ) context_action_conditioning_ready = bool( runner_status.get("context_action_conditioning_ready") ) ready = bool( self.bundle_verified and runner_status.get("loaded") and full_model_active and tensor_packages_loaded == 113 and tokenizer_acceleration_ready and self_correction_ready and self_improvement_ready and context_intent_action_ready and context_intent_effect_ready and context_action_effect_ready and context_action_conditioning_ready and self.release_artifact_sha256 and self.runtime_source_sha256 and self.candidate_identity_sha256 and self.runtime_instance_sha256 ) _json_response( self, 200 if ready else 503, { "schema": "nexum.release-status.v4", "ready": ready, "full_model_active": full_model_active, "tensor_packages_loaded": tensor_packages_loaded, "tokenizer_acceleration_ready": ( tokenizer_acceleration_ready ), "self_correction_ready": self_correction_ready, "self_improvement_ready": self_improvement_ready, "context_intent_action_ready": context_intent_action_ready, "context_intent_effect_ready": context_intent_effect_ready, "context_action_effect_ready": context_action_effect_ready, "context_action_conditioning_ready": ( context_action_conditioning_ready ), "release_artifact_sha256": self.release_artifact_sha256, "runtime_source_sha256": self.runtime_source_sha256, "candidate_identity_sha256": self.candidate_identity_sha256, "runtime_instance_sha256": self.runtime_instance_sha256, "state_namespace": self.state_namespace, "state_pristine": bool(runner_status.get("state_pristine")), **load_state, "learning_generation": int( runner_status.get("learning_generation") or 0 ), }, ) return if path == "/capabilities": _json_response(self, 200, capability_catalog()) return if path == "/language-packs": requested = query.get("query", query.get("q", [""])) _json_response( self, 200, language_pack_catalog(requested[0] if requested else ""), ) return if path == "/tools": _json_response(self, 200, {"tools": list_tools()}) return if path == "/v1/models": runner_status = _RUNNER.status() if _RUNNER is not None else {} ready = bool( self.bundle_verified and runner_status.get("loaded") and runner_status.get("self_correction_ready") and runner_status.get("self_improvement_ready") and runner_status.get("context_intent_action_ready") ) _json_response( self, 200 if ready else 503, { "object": "list", "data": ( [ { "id": "Nexum", "object": "model", "owned_by": "namenotfound.ai", } ] if ready else [] ), "ready": ready, }, ) return if path == "/self-correction/status": runner_status = _RUNNER.status() if _RUNNER is not None else {} _json_response( self, 200, { **SELF_CORRECTION_STATUS, "active": bool(runner_status.get("self_correction_ready")), }, ) return if path == "/self-improvement/status": runner_status = _RUNNER.status() if _RUNNER is not None else {} learning = runner_status.get("learning") _json_response( self, 200, { **SELF_IMPROVEMENT_STATUS, "active": bool(runner_status.get("self_improvement_ready")), "learning": (dict(learning) if isinstance(learning, dict) else {}), }, ) return if path == "/templates": _json_response(self, 200, {"templates": TEMPLATES}) return if path == "/approvals": if not self._require_tool_execution(): return session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") records = [ record.__dict__ for record in ApprovalStore(self.workspace).list(session_id=session_id) ] _json_response(self, 200, {"approvals": records}) return if path.startswith("/approvals/"): if not self._require_tool_execution(): return approval_id = path.removeprefix("/approvals/") session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") record = ApprovalStore(self.workspace).get( approval_id, session_id=session_id ) _json_response(self, 200, {"approval": record.__dict__}) return if path == "/tasks": if not self._require_tool_execution(): return session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") records = [ record.to_dict() for record in TaskStore(self.workspace).list(session_id=session_id) ] _json_response(self, 200, {"tasks": records}) return if path.startswith("/tasks/"): if not self._require_tool_execution(): return task_id = path.removeprefix("/tasks/") session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") task = TaskStore(self.workspace).status(task_id, session_id=session_id) _json_response(self, 200, {"task": task.to_dict()}) return if path == "/events": if not self._require_tool_execution(): return session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") after = int(str((query.get("after") or ["0"])[0])) events = [ event.to_dict() for event in EventLog(self.workspace).read( session_id=session_id, after_sequence=after, ) ] _json_response(self, 200, {"events": events}) return if path == "/artifacts": if not self._require_tool_execution(): return session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") records = [ record.to_dict() for record in ArtifactStore(self.workspace).list(session_id=session_id) ] _json_response(self, 200, {"artifacts": records}) return if path.startswith("/artifacts/"): if not self._require_tool_execution(): return artifact_id = path.removeprefix("/artifacts/") offset = int(str((query.get("offset") or ["0"])[0])) raw_length = str((query.get("length") or [""])[0]) length = int(raw_length) if raw_length else None session_id = str((query.get("session_id") or [""])[0]).strip() if not session_id: raise ValueError("session_id is required") artifact_record, data = ArtifactStore(self.workspace).read( artifact_id, offset=offset, length=length, session_id=session_id, ) _json_response( self, 200, { "artifact": artifact_record.to_dict(), "offset": offset, "data_base64": base64.b64encode(data).decode("ascii"), }, ) return _json_response(self, 404, {"ok": False, "error": "not_found"}) def do_POST(self) -> None: try: if not self._require_authorization(): return payload = self._read_payload() path, _query = self._route() if path == "/model/validate": deep = bool(payload.get("deep", False)) report = validate_bundle(self.model_dir, deep=deep) _json_response(self, 200 if report.ok else 503, report.to_dict()) return if path == "/self-improvement/rollback": if not self._require_tool_execution(): return expected_candidate = str( payload.get("expected_candidate_identity_sha256") or "" ) if not hmac.compare_digest( expected_candidate, self.candidate_identity_sha256, ): raise ValueError("candidate identity changed before rollback") target_generation = payload.get("target_generation") expected_generation = payload.get("expected_generation") if ( isinstance(target_generation, bool) or not isinstance(target_generation, int) or isinstance(expected_generation, bool) or not isinstance(expected_generation, int) ): raise ValueError( "target_generation and expected_generation must be integers" ) result = _runner(self.model_dir, self.device).rollback_learning( target_generation=target_generation, expected_generation=expected_generation, ) _json_response( self, 200, { **result, "candidate_identity_sha256": self.candidate_identity_sha256, "state_namespace": self.state_namespace, }, ) return if path == "/tools/execute": if not self._require_tool_execution(): return cwd = self._workspace(payload.get("workspace")) timeout_s = float(payload.get("timeout_s") or 0.0) session_id = str(payload.get("session_id") or "").strip() raw_calls = payload.get("tool_calls") structured_calls = raw_calls is not None if structured_calls: if ( not isinstance(raw_calls, list) or not raw_calls or any(not isinstance(item, dict) for item in raw_calls) ): raise ValueError("tool_calls must be a non-empty object array") executed = execute_tool_calls( tuple(tool_call_from_openai(item) for item in raw_calls), cwd=cwd, timeout_s=timeout_s, session_id=session_id, ) else: if session_id: raise ValueError( "signed execution requires exact structured tool_calls" ) text = str(payload.get("tool_text") or payload.get("text") or "") executed = tuple( execute_tool_text( text, cwd=cwd, timeout_s=timeout_s, ) ) results = [item.to_dict() for item in executed] observations: list[dict[str, Any]] = [] if session_id and structured_calls: runner = _runner(self.model_dir, self.device) observations = [ runner.sign_observation( session_id, { **result, "receipt_source": ( "runtime_execution" if item.executed else "runtime_rejection" ), }, ) for item, result in zip(executed, results, strict=True) ] ok = bool(results) and all(item["ok"] for item in results) _json_response( self, 200 if ok else 422, { "ok": ok, "results": results, "observations": observations, }, ) return if path == "/agent/run": if not self._require_tool_execution(): return request_payload = dict(payload) request_payload["workspace"] = self._workspace(payload.get("workspace")) request = NNFXHarnessRequest.from_json(request_payload) runner = _runner(self.model_dir, self.device) response = NNFXHarnessSession( complete=runner.chat, cancelled=self._client_disconnected, sign_observation=runner.sign_observation, ).execute(request) _json_response(self, 200 if response.ok else 422, response.to_dict()) return if path == "/observations/attest": if not self._require_tool_execution(): return session_id = str(payload.get("session_id") or "").strip() observation = payload.get("observation") if not session_id: raise ValueError("session_id is required") if not isinstance(observation, dict): raise ValueError("observation must be an object") observation = { **observation, "receipt_source": "caller_attested", "source_trust": "caller_owned", } signed = _runner(self.model_dir, self.device).sign_observation( session_id, observation ) _json_response(self, 200, {"ok": True, "observation": signed}) return if path == "/approvals/decide": if not self._require_tool_execution(): return approval_id = str(payload.get("approval_id") or "") session_id = str(payload.get("session_id") or "") if not session_id: raise ValueError("session_id is required") record = ApprovalStore(self.workspace).decide( approval_id, approved=bool(payload.get("approved")), session_id=session_id, ) EventLog(self.workspace).append( "approval_decision", session_id=session_id, status=record.status, detail={"approval_id": approval_id, "tool": record.tool}, ) _json_response(self, 200, {"ok": True, "approval": record.__dict__}) return if path == "/tasks/cancel": if not self._require_tool_execution(): return session_id = str(payload.get("session_id") or "").strip() if not session_id: raise ValueError("session_id is required") task = TaskStore(self.workspace).cancel( str(payload.get("task_id") or ""), session_id=session_id, ) _json_response(self, 200, {"ok": True, "task": task.to_dict()}) return if path == "/artifacts": if not self._require_tool_execution(): return session_id = str(payload.get("session_id") or "").strip() if not session_id: raise ValueError("session_id is required") if "data_base64" in payload: data = base64.b64decode( str(payload.get("data_base64") or ""), validate=True ) else: data = str(payload.get("text") or "").encode("utf-8") artifact_record = ArtifactStore(self.workspace).put( data, media_type=str( payload.get("media_type") or "application/octet-stream" ), source="caller_upload", session_id=session_id, ) _json_response( self, 200, {"ok": True, "artifact": artifact_record.to_dict()}, ) return if path == "/mcp": result = mcp_request( payload, workspace=self.workspace, session_id=str(payload.get("session_id") or uuid.uuid4().hex), tools_enabled=self.enable_tools, ) _json_response(self, 200, result) return if path == "/a2a": if not self._require_tool_execution(): return runner = _runner(self.model_dir, self.device) def run_agent(agent_payload: dict[str, Any]) -> dict[str, Any]: request_payload = dict(agent_payload) request_payload["workspace"] = self.workspace request = NNFXHarnessRequest.from_json(request_payload) return ( NNFXHarnessSession( complete=runner.chat, cancelled=self._client_disconnected, sign_observation=runner.sign_observation, ) .run(request) .to_dict() ) _json_response(self, 200, a2a_request(payload, run_agent=run_agent)) return if path == "/v1/chat/completions": runner = _runner(self.model_dir, self.device) request_payload = dict(payload) session_id = str(request_payload.get("session_id") or "").strip() if not session_id: session_id = uuid.uuid4().hex request_payload["session_id"] = session_id result = runner.chat(request_payload) result = _session_bound_chat_result( result, session_id=session_id, ) _json_response(self, 200, result) return if path == "/v1/responses": chat = _responses_chat(_runner(self.model_dir, self.device), payload) _json_response( self, 200, _responses_result(_visible_chat_result(chat)), ) return _json_response(self, 404, {"ok": False, "error": "not_found"}) except (PermissionError, TypeError, ValueError) as exc: _json_response( self, 422, {"ok": False, "error": f"{type(exc).__name__}: {exc}"} ) except Exception as exc: _json_response(self, 500, {"ok": False, "error": type(exc).__name__}) def serve( *, host: str, port: int, model_dir: str, workspace: str, device: str, api_key: str = "", enable_tools: bool = False, state_dir: str = "", state_namespace: str = "", ) -> None: api_key = api_key.strip() if host not in {"127.0.0.1", "::1", "localhost"} and not api_key: raise ValueError("a bearer API key is required for non-loopback binding") if enable_tools and not api_key: raise ValueError("a bearer API key is required when server tools are enabled") if not model_dir.strip(): raise ValueError("model directory is required") if state_dir.strip(): os.environ["NEXUM_STATE_DIR"] = str(Path(state_dir).expanduser().resolve()) if state_namespace.strip(): os.environ["NEXUM_STATE_NAMESPACE"] = state_namespace.strip() resolved_model_dir = str(Path(model_dir).expanduser().resolve()) report = validate_bundle(resolved_model_dir, deep=True) if not report.ok: raise RuntimeError("Nexum model bundle validation failed") artifact_sha256 = release_artifact_sha256( resolved_model_dir, validated_report=report ) source_sha256 = runtime_source_sha256() candidate_sha256 = candidate_identity_sha256( artifact_sha256, source_sha256, ) namespace = runtime_state_namespace() instance_sha256 = _runtime_instance_sha256( artifact_sha256, source_sha256, namespace, ) runner = _runner(resolved_model_dir, device) _start_runner_load(runner) NexumHandler.model_dir = resolved_model_dir NexumHandler.workspace = str(Path(workspace).expanduser()) NexumHandler.device = device NexumHandler.api_key = api_key NexumHandler.enable_tools = enable_tools NexumHandler.bundle_verified = bool(report.tensor_hashes_verified) NexumHandler.release_artifact_sha256 = artifact_sha256 NexumHandler.runtime_source_sha256 = source_sha256 NexumHandler.candidate_identity_sha256 = candidate_sha256 NexumHandler.runtime_instance_sha256 = instance_sha256 NexumHandler.state_namespace = namespace server = ThreadingHTTPServer((host, port), NexumHandler) server.daemon_threads = True server.serve_forever() def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Run the Nexum HTTP server") parser.add_argument("--host", default="127.0.0.1") parser.add_argument("--port", type=int, default=8080) parser.add_argument("--model", default="") parser.add_argument("--workspace", default=".") parser.add_argument("--device", default=os.environ.get("NEXUM_DEVICE", "cuda:0")) parser.add_argument("--state-dir", default=os.environ.get("NEXUM_STATE_DIR", "")) parser.add_argument( "--state-namespace", default=os.environ.get("NEXUM_STATE_NAMESPACE", "default"), ) parser.add_argument( "--enable-tools", action="store_true", default=os.environ.get("NEXUM_ENABLE_TOOLS", "").lower() in {"1", "true", "yes"}, ) args = parser.parse_args(argv) serve( host=args.host, port=args.port, model_dir=args.model, workspace=args.workspace, device=args.device, api_key=os.environ.get("NEXUM_API_KEY", ""), enable_tools=bool(args.enable_tools), state_dir=args.state_dir, state_namespace=args.state_namespace, ) return 0 if __name__ == "__main__": raise SystemExit(main())