import json

import inspect

import functools

from typing import Any, Dict, List, TypeVar, Callable, Optional

from dataclasses import dataclass

T = TypeVar("T")

# Context definition

@dataclass

class KernelContext:

"""Context object passed to action handlers"""

invocation_id: str

# Action definition

@dataclass

class KernelAction:

"""Action that can be invoked on a Kernel app"""

name: str

handler: Callable[..., Any]

# JSON interfaces

@dataclass

class KernelActionJson:

"""JSON representation of a Kernel action"""

name: str

@dataclass

class KernelAppJson:

"""JSON representation of a Kernel app"""

name: str

actions: List[KernelActionJson]

@dataclass

class KernelJson:

"""JSON representation of Kernel manifest"""

apps: List[KernelAppJson]

# App class

class KernelApp:

def __init__(self, name: str):

self.name = name

self.actions: Dict[str, KernelAction] = {}

# Register this app in the global registry

_app_registry.register_app(self)

def action(self, name: str) -> Callable[..., Any]:

"""

Decorator to register an action with the app

Usage:

@app.action("action-name")

def my_handler(ctx: KernelContext):

# ...

@app.action("action-with-payload")

def my_handler(ctx: KernelContext, payload: dict):

# ...

"""

def decorator(handler: Callable[..., Any]) -> Callable[..., Any]:

return self._register_action(name, handler)

return decorator

def _register_action(self, name: str, handler: Callable[..., Any]) -> Callable[..., Any]:

"""Internal method to register an action"""

# Validate handler signature

sig = inspect.signature(handler)

param_count = len(sig.parameters)

if param_count == 0:

raise TypeError("Action handler must accept at least the context parameter")

elif param_count > 2:

raise TypeError("Action handler can only accept context and payload parameters")

param_names = list(sig.parameters.keys())

@functools.wraps(handler)

def wrapper(*args: Any, **kwargs: Any) -> Any:

# Ensure the first argument is the context

if not args or not isinstance(args[0], KernelContext):

raise TypeError("First argument to action handler must be a KernelContext")

ctx = args[0]

if param_count == 1:

# Handler takes only context

return handler(ctx)

else: # param_count == 2

# Handler takes context and payload

if len(args) >= 2:

return handler(ctx, args[1])

else:

# Try to find payload in kwargs

payload_name = param_names[1]

if payload_name in kwargs:

return handler(ctx, kwargs[payload_name])

else:

raise TypeError(f"Missing required payload parameter '{payload_name}'")

action = KernelAction(name=name, handler=wrapper)

self.actions[name] = action

return wrapper

def get_actions(self) -> List[KernelAction]:

"""Get all actions for this app"""

return list(self.actions.values())

def get_action(self, name: str) -> Optional[KernelAction]:

"""Get an action by name"""

return self.actions.get(name)

def to_dict(self) -> KernelAppJson:

"""Export app information without handlers"""

return KernelAppJson(

name=self.name,

actions=[KernelActionJson(name=action.name) for action in self.get_actions()]

)

# Registry for storing Kernel apps

class KernelAppRegistry:

def __init__(self) -> None:

self.apps: Dict[str, KernelApp] = {}

def register_app(self, app: KernelApp) -> None:

self.apps[app.name] = app

def get_apps(self) -> List[KernelApp]:

return list(self.apps.values())

def get_app_by_name(self, name: str) -> Optional[KernelApp]:

return self.apps.get(name)

def export(self) -> KernelJson:

"""Export the registry as a KernelJson object"""

apps = [app.to_dict() for app in self.get_apps()]

return KernelJson(apps=apps)

def export_json(self) -> str:

"""Export the registry as JSON"""

kernel_json = self.export()

return json.dumps(kernel_json.__dict__, indent=2)

# Create singleton registry for apps

_app_registry = KernelAppRegistry()

# Create a simple function for creating apps

def App(name: str) -> KernelApp:

"""Create a new Kernel app"""

return KernelApp(name)

# Export the app registry for boot loader

app_registry = _app_registry

# Function to export registry as JSON

def export_registry() -> str:

"""Export the registry as JSON"""

return _app_registry.export_json()