contracts

app.py

@@ -13,14 +13,13 @@
               or: gunicorn ... "app:app"
 """
 from kernel_ai.config import Config
+from kernel_ai.services.core_observability import get_system_info
 from kernel_ai.webapp import app, create_app
 
 __all__ = ["app", "create_app"]
 
 
 if __name__ == "__main__":
-    from kernel_ai.webapp import get_system_info
-
     system_info = get_system_info()
     print("🚀 Linux Kernel Visualization Backend")
     print(f"📍 Platform: {system_info['platform']}")

kernel_ai/contracts/__init__.py

@@ -0,0 +1,2 @@
+"""API and domain-level response contracts."""
+

kernel_ai/contracts/api_contracts.py

@@ -0,0 +1,108 @@
+"""Runtime validators for stable HTTP response contracts."""
+
+from __future__ import annotations
+
+from typing import Any, TypedDict
+
+
+class KernelDataResponse(TypedDict):
+    timestamp: str
+    syscalls: list
+    subsystems: dict
+    processes: int
+    system_stats: dict
+
+
+class NetworkStackRealtimeResponse(TypedDict):
+    timestamp: str
+    layer_metrics: dict
+    layer_activity: dict
+    signals: dict
+    throughput_mb_s: float
+    tcp_counters: dict
+
+
+class CryptoRealtimeResponse(TypedDict):
+    items: list
+    processes: list
+    meta: dict
+
+
+class SecurityRealtimeResponse(TypedDict):
+    timestamp: str
+    pipeline: dict
+    trust_graph: list
+    attack_surface: list
+    security_tools: dict
+    security_core: dict
+    meta: dict
+
+
+class FilesystemBlocksResponse(TypedDict):
+    timestamp: str
+    rows: int
+    cols: int
+    zones: list
+    blocks: list
+    meta: dict
+
+
+def _expect_dict(payload: Any, name: str) -> dict:
+    if not isinstance(payload, dict):
+        raise ValueError(f"{name} must be a JSON object")
+    return payload
+
+
+def _expect_key(payload: dict, key: str, expected_type: type | tuple[type, ...], name: str) -> None:
+    if key not in payload:
+        raise ValueError(f"{name}: missing key '{key}'")
+    if not isinstance(payload[key], expected_type):
+        raise ValueError(f"{name}: key '{key}' has invalid type")
+
+
+def validate_kernel_data_response(payload: Any) -> None:
+    data = _expect_dict(payload, "kernel_data")
+    _expect_key(data, "timestamp", str, "kernel_data")
+    _expect_key(data, "syscalls", list, "kernel_data")
+    _expect_key(data, "subsystems", dict, "kernel_data")
+    _expect_key(data, "processes", int, "kernel_data")
+    _expect_key(data, "system_stats", dict, "kernel_data")
+
+
+def validate_network_stack_realtime_response(payload: Any) -> None:
+    data = _expect_dict(payload, "network_stack_realtime")
+    _expect_key(data, "timestamp", str, "network_stack_realtime")
+    _expect_key(data, "layer_metrics", dict, "network_stack_realtime")
+    _expect_key(data, "layer_activity", dict, "network_stack_realtime")
+    _expect_key(data, "signals", dict, "network_stack_realtime")
+    _expect_key(data, "throughput_mb_s", (int, float), "network_stack_realtime")
+    _expect_key(data, "tcp_counters", dict, "network_stack_realtime")
+
+
+def validate_crypto_realtime_response(payload: Any) -> None:
+    data = _expect_dict(payload, "crypto_realtime")
+    _expect_key(data, "items", list, "crypto_realtime")
+    _expect_key(data, "processes", list, "crypto_realtime")
+    _expect_key(data, "meta", dict, "crypto_realtime")
+    _expect_key(data["meta"], "active_flows", int, "crypto_realtime.meta")
+
+
+def validate_security_realtime_response(payload: Any) -> None:
+    data = _expect_dict(payload, "security_realtime")
+    _expect_key(data, "timestamp", str, "security_realtime")
+    _expect_key(data, "pipeline", dict, "security_realtime")
+    _expect_key(data, "trust_graph", list, "security_realtime")
+    _expect_key(data, "attack_surface", list, "security_realtime")
+    _expect_key(data, "security_tools", dict, "security_realtime")
+    _expect_key(data, "security_core", dict, "security_realtime")
+    _expect_key(data, "meta", dict, "security_realtime")
+
+
+def validate_filesystem_blocks_response(payload: Any) -> None:
+    data = _expect_dict(payload, "filesystem_blocks")
+    _expect_key(data, "timestamp", str, "filesystem_blocks")
+    _expect_key(data, "rows", int, "filesystem_blocks")
+    _expect_key(data, "cols", int, "filesystem_blocks")
+    _expect_key(data, "zones", list, "filesystem_blocks")
+    _expect_key(data, "blocks", list, "filesystem_blocks")
+    _expect_key(data, "meta", dict, "filesystem_blocks")

kernel_ai/http/api.py

@@ -0,0 +1,67 @@
+"""REST API handlers facade."""
+
+from kernel_ai.http.api_handlers.kernel import (
+    get_execution_context,
+    kernel_data,
+    kernel_dna,
+    nginx_files,
+    process_kernel_map,
+    syscalls_realtime,
+)
+from kernel_ai.http.api_handlers.network_system import (
+    active_connections,
+    devices_realtime,
+    filesystem_blocks,
+    isolation_context,
+    network_stack_realtime,
+    traceroute_info,
+)
+from kernel_ai.http.api_handlers.processes import (
+    get_ipc_links,
+    get_proc_graph,
+    get_proc_matrix,
+    get_proc_timeline,
+    get_process_cpu,
+    get_process_fds,
+    get_process_files,
+    get_process_threads,
+    get_processes,
+    get_processes_detailed,
+    processes_realtime,
+)
+from kernel_ai.http.api_handlers.security_logs import (
+    crypto_realtime,
+    ingest_frontend_logs,
+    security_realtime,
+)
+
+__all__ = [
+    "active_connections",
+    "crypto_realtime",
+    "devices_realtime",
+    "filesystem_blocks",
+    "get_execution_context",
+    "get_ipc_links",
+    "get_proc_graph",
+    "get_proc_matrix",
+    "get_proc_timeline",
+    "get_process_cpu",
+    "get_process_fds",
+    "get_process_files",
+    "get_process_kernel_map",
+    "get_process_threads",
+    "get_processes",
+    "get_processes_detailed",
+    "ingest_frontend_logs",
+    "isolation_context",
+    "kernel_data",
+    "kernel_dna",
+    "network_stack_realtime",
+    "nginx_files",
+    "process_kernel_map",
+    "processes_realtime",
+    "security_realtime",
+    "syscalls_realtime",
+    "traceroute_info",
+]
+

kernel_ai/http/common.py

@@ -0,0 +1,20 @@
+"""Common HTTP helpers."""
+
+from flask import jsonify
+
+
+def api_json(producer, error_status=500, error_extra=None, exception_statuses=None):
+    """Execute producer and serialize response/error as JSON."""
+    try:
+        return jsonify(producer())
+    except Exception as e:
+        status = error_status
+        if exception_statuses:
+            for exc_type, exc_status in exception_statuses:
+                if isinstance(e, exc_type):
+                    status = exc_status
+                    break
+        payload = {"error": str(e)}
+        if error_extra:
+            payload.update(error_extra)
+        return jsonify(payload), status

kernel_ai/http/pages.py

@@ -0,0 +1,70 @@
+"""HTTP handlers for site pages."""
+
+from datetime import datetime
+
+from flask import jsonify, redirect, render_template, send_from_directory
+
+from kernel_ai.config import PROJECT_ROOT
+from kernel_ai.services import core_observability as _core_observability_service
+
+
+def get_system_info():
+    return _core_observability_service.get_system_info()
+
+
+def index():
+    return send_from_directory(str(PROJECT_ROOT), "index.html")
+
+
+def linux_crypto_subsystem_page():
+    return render_template("linux-crypto-subsystem.html")
+
+
+def crypto_page_legacy():
+    return redirect("/linux-crypto-subsystem", code=301)
+
+
+def linux_security_subsystem_page():
+    return render_template("linux-security-subsystem.html")
+
+
+def security_page_legacy():
+    return redirect("/linux-security-subsystem", code=301)
+
+
+def linux_processes_subsystem_page():
+    return render_template("linux-processes-subsystem.html")
+
+
+def processes_page_legacy():
+    return redirect("/linux-processes-subsystem", code=301)
+
+
+def linux_crypto_subsystem_html():
+    return redirect("/linux-crypto-subsystem", code=301)
+
+
+def linux_security_subsystem_html():
+    return redirect("/linux-security-subsystem", code=301)
+
+
+def linux_processes_subsystem_html():
+    return redirect("/linux-processes-subsystem", code=301)
+
+
+def linux_memory_subsystem_page():
+    return render_template("linux-memory-subsystem.html")
+
+
+def linux_memory_subsystem_html():
+    return redirect("/linux-memory-subsystem", code=301)
+
+
+def health_check():
+    return jsonify(
+        {
+            "status": "healthy",
+            "timestamp": datetime.now().isoformat(),
+            "system_info": get_system_info(),
+        }
+    )

kernel_ai/http/register.py

@@ -1,9 +1,12 @@
-"""Register Flask blueprints (called after all view implementations are defined in webapp)."""
+"""Register Flask blueprints for pages and API modules."""
 
 
 def register_http_routes(app):
-    from kernel_ai.api.rest import bp as api_bp
-    from kernel_ai.views.pages import bp as pages_bp
+    # Keep import-callables lazy to avoid import cycles at app bootstrap.
+    blueprint_loaders = [
+        lambda: __import__("kernel_ai.views.pages", fromlist=["bp"]).bp,
+        lambda: __import__("kernel_ai.api.rest", fromlist=["bp"]).bp,
+    ]
 
-    app.register_blueprint(pages_bp)
-    app.register_blueprint(api_bp)
+    for load_bp in blueprint_loaders:
+        app.register_blueprint(load_bp())