Nexum-Expanded/runtime/src/nexum_runtime/server.py

"""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())