super-solid-system 1.0.2

A lightweight, type-safe microkernel & plug-and-play framework for Python.

Super-Solid System

A robust, generic infrastructure for building modular software systems with plug-and-play engines, dynamic plugin discovery, thread-safe component registries, and fault-tolerant system lifecycle management.

A lightweight, type-safe microkernel & plug-and-play framework for Python.

What is Super-Solid System?

Super-Solid System (S³) gives you a small set of building blocks for assembling modular applications:

• Registries hold your components — type-checked, thread-safe, and cached.
• Engines group registries together and handle boot/shutdown logic.
• Systems orchestrate engines in dependency order with automatic rollback.
• Events let everything talk to each other through a reactive signal bus.
• Security controls who can load what and from where.
• Health provides a standard way for components to report their status.

No magic, no hidden state, no framework lock-in. Just clean abstractions you subclass and wire up.

Architecture

The core loop is simple:

1. Register component classes into typed registries (or let discovery find them).
2. Load components on demand — the registry instantiates, caches, and returns them.
3. Wire registries into engines, engines into a system, and call start().

Installation

Pip (editable, for development)

pip install -e /path/to/super-solid-system

Poetry (as a local dependency)

poetry add --editable /path/to/super-solid-system

Docker (Multi-stage build)

# Build the production Docker image
docker build -t super-solid-system:latest .

# Run container & verify installation
docker run --rm super-solid-system:latest

Quickstart

1. Define a base interface and a registry

from abc import ABC, abstractmethod
from supersolid.core import super_solid_registry, SuperSolidRegistry

class BaseAdapter(ABC):
    @abstractmethod
    def connect(self) -> None: ...

# [QUICK] Create a typed component registry using super_solid_registry
Adapters = super_solid_registry(name="adapters", component_class=BaseAdapter)

# [ALTERNATIVE] Create a class component registry by subclassing SuperSolidRegistry
class AdapterRegistry(SuperSolidRegistry[BaseAdapter]):
    def __init__(self):
        super().__init__(name="adapters", component_class=BaseAdapter)

Adapters = AdapterRegistry()

SuperSolidRegistry is generic — it enforces that only BaseAdapter subclasses can be registered here. Anything else raises a TypeError at registration time.

2. Register components

You can register manually:

Adapters.register("database", "sqlite", SqliteAdapter)

Or use the @super_solid_component decorator for a more declarative style:

from supersolid.core import super_solid_component

@super_solid_component(Adapters, namespace="database", name="postgres", version="1.0.0")
class PostgresAdapter(BaseAdapter):
    def __init__(self, dsn: str = "postgresql://localhost/db"):
        self.dsn = dsn

    def connect(self) -> None:
        print(f"Connected to {self.dsn}")

    def disconnect(self) -> None:
        print("Disconnected.")

The decorator registers the class and attaches a ComponentMetadata model you can inspect later:

PostgresAdapter._metadata.version   # "1.0.0"
PostgresAdapter._metadata.namespace # "database"

3. Load components

db = Adapters.load("database", "postgres", dsn="postgresql://prod/mydb")
db.connect()  # Connected to postgresql://prod/mydb

What happens under the hood:

1. Checks the cache — if already loaded with the same parameters, returns the cached instance.
2. If not registered, runs the discovery chain (more on that below).
3. Introspects the constructor signature and filters keyword arguments automatically.
4. Calls lifecycle hooks (initialize() or boot()) if the component defines them.
5. Caches and returns the instance.

When you're done:

Adapters.unload("database", "postgres")
# Calls disconnect() or stop() automatically if the component defines them

4. Build an engine

A SuperSolidEngine groups registries and provides a unified loading interface:

from supersolid.core import SuperSolidEngine

class CoreEngine(SuperSolidEngine):
    def __init__(self):
        super().__init__()
        self.connect(Adapters)  # attach registries

    def boot(self, **kwargs) -> None:
        self._running = True
        db = self.load("adapters", "database", "postgres")
        db.connect()

    def mount(self, app, **kwargs) -> None:
        pass  # hook for attaching to a web framework, etc.

    def shutdown(self, **kwargs) -> None:
        self._running = False

Need async? Subclass SuperSolidAsyncEngine instead — it provides boot_async(), mount_async(), and shutdown_async() that automatically bridge to the sync interface via asyncio.run().

5. Orchestrate with a system

from supersolid.core import SuperSolidSystem

class MyApp(SuperSolidSystem):
    pass

app = MyApp()
app.add_engine("core", CoreEngine())
app.start()  # boots engines in dependency order
# ... your app runs ...
app.stop()   # shuts down in reverse order

If any engine fails during start(), all previously booted engines are shut down in reverse (LIFO) order automatically.

Engine Dependencies (Dependency Resolution)

Engines can declare dependencies on other engines:

class ConsensusEngine(SuperSolidEngine):
    depends_on = ["network"]  # boot network engine first
    # ...

class NetworkEngine(SuperSolidEngine):
    depends_on = []
    # ...

system = MyApp()
system.add_engine("consensus", ConsensusEngine())
system.add_engine("network", NetworkEngine())
system.start()
# Boot order: network → consensus (resolved via topological sort)

The system uses the internal graphlibs TopologicalSorter so circular dependencies raise a CycleError immediately.

Events

SuperSolidSystem comes with a built-in SuperSolidEventBus that publishes lifecycle signals automatically:

Event	When it fires
EngineBootingEvent	Right before an engine boots
EngineBootedEvent	After an engine boots successfully
EngineShutdownEvent	After an engine shuts down
SystemErrorEvent	On boot failure or shutdown error
PluginLoadedEvent	When a plugin is loaded

Subscribing to events

from supersolid.core import SuperSolidEventBus, EngineBootedEvent, super_solid_subscriber

bus = SuperSolidEventBus()

@super_solid_subscriber(bus)
def on_boot(event: EngineBootedEvent):
    print(f"Engine '{event.engine_name}' is up!")

The decorator infers the event type from the parameter annotation. You can also be explicit:

@super_solid_subscriber(bus, event_type=EngineBootedEvent)
def on_boot(event):
    print(f"Engine '{event.engine_name}' is up!")

Publishing events

The @super_solid_publisher decorator auto-publishes a function's return value if it's a SystemEvent:

from supersolid.core import super_solid_publisher, EngineBootedEvent

@super_solid_publisher(bus)
def finish_boot(name: str) -> EngineBootedEvent:
    # ... do boot work ...
    return EngineBootedEvent(engine_name=name)

finish_boot("consensus")  # automatically published to bus

Works with both sync and async functions.

Custom events

All events are Pydantic models with extra = "allow", so you can add any fields:

from supersolid.core import SystemEvent

class NodeSyncEvent(SystemEvent):
    node_id: str
    epoch: int

# Extra fields work too — they serialize to JSON just fine
event = NodeSyncEvent(node_id="node_01", epoch=42, custom_field="whatever")
event.model_dump_json()

Plugin Discovery

When you call registry.load() for something that isn't registered yet, the registry runs a discovery chain to try to find it automatically. The default chain has three strategies:

Strategy	Where it looks
InternalLib	{root_pkg}.lib.{registry_name}.{namespace}_{name} (trusted imports)
EntryPoints	Setuptools entry points in group {root_pkg}.{registry_name}.{namespace}
PluginDirectory	plugins.{registry_name}.{namespace}_{name} (local files)

Writing a custom discovery strategy

from supersolid.core import DiscoveryStrategy

class RemoteDiscovery(DiscoveryStrategy):
    def discover(self, registry, item_namespace, item_name) -> bool:
        # fetch plugin from remote source, register it
        return registry.is_registered(item_namespace, item_name)

Adapters.add_discovery(RemoteDiscovery())

Security

SecurityPolicy

Attach a SecurityPolicy to any registry to control access:

from supersolid.core import SecurityPolicy

policy = SecurityPolicy(
    allowed_namespaces={"database", "cache"},  # only these namespaces can be loaded
    allowed_callers={"core_engine"},            # only these caller IDs are authorized
    allow_internal_lib=True,                    # trusted internal imports still work
    allow_dynamic_discovery=False,              # block external plugins (EntryPoints, PluginDirectory)
)

Adapters.set_policy(policy)

Now Adapters.load("auth", "oauth") raises PermissionError because "auth" isn't in the allowed set. And external discovery strategies are skipped entirely, while InternalLib still runs as a trusted fallback.

Plugin verification

Before importing a plugin file, you can verify its checksum:

from supersolid.core import PluginVerifier

if PluginVerifier.verify_sha256("plugins/consensus/my_plugin.py", expected_hash):
    import plugins.consensus.my_plugin

Health Checks

Any component can implement health reporting by defining a health_check() method:

from supersolid.core import HealthStatus, HealthState, HealthCheckable

class DatabaseAdapter:
    def health_check(self) -> HealthStatus:
        return HealthStatus(
            state=HealthState.HEALTHY,
            details={"connections": 42, "pool_size": 100}
        )

db = DatabaseAdapter()
isinstance(db, HealthCheckable)  # True — structural typing via Protocol
db.health_check().model_dump_json()
# {"state": "healthy", "details": {"connections": 42, "pool_size": 100}, "timestamp": ...}

HealthCheckable is a @runtime_checkable Protocol — no need to inherit from anything.

The three states are HEALTHY, DEGRADED, and UNHEALTHY.

Project Structure

super-solid-system/
├── supersolid/
│   └── core/
│       ├── __init__.py       # Public API exports
│       ├── registry.py       # SuperSolidRegistry, super_solid_component, ComponentMetadata
│       ├── engine.py         # SuperSolidEngine, AsyncSuperSolidEngine
│       ├── system.py         # SuperSolidSystem (orchestrator)
│       ├── events.py         # SuperSolidEventBus, SystemEvent, super_solid_subscriber, super_solid_publisher
│       ├── discovery.py      # DiscoveryStrategy, InternalLib, EntryPoints, PluginDirectory
│       ├── security.py       # SecurityPolicy, PluginVerifier
│       └── health.py         # HealthStatus, HealthState, HealthCheckable
├── tests/
│   ├── test_registry.py      # Registration, loading, type enforcement, thread safety, unload
│   ├── test_events.py        # Event bus pub/sub, @open_subscriber, @open_publisher
│   ├── test_security.py      # Namespace ACL, caller auth, discovery fallbacks, SHA-256
│   ├── test_health.py        # Health models, Protocol duck typing
│   └── test_system.py        # DAG resolution, start/stop, rollback on failure
├── pyproject.toml
├── LICENSE
└── README.md

---

API Reference

Export	Module	What it does
SuperSolidSystem	core.system	Orchestrator — boots/stops engines in dependency order (SolidSystem, OpenSystem)
SuperSolidEngine	core.engine	Domain hub — groups registries, handles boot/mount/shutdown (SolidEngine, OpenEngine)
SuperSolidAsyncEngine	core.engine	Async variant supporting boot_async(), mount_async() (SolidAsyncEngine, AsyncOpenEngine)
SuperSolidRegistry	core.registry	Generic registry — type checks, caching, discovery (SolidRegistry, OpenRegistry)
super_solid_registry	core.registry	Factory function — instantiates a SuperSolidRegistry (solid_registry, open_registry)
super_solid_component	core.registry	Decorator — registers a class with metadata (solid_component, open_component)
ComponentMetadata	core.registry	Pydantic model — name, namespace, version, and extras
SuperSolidEventBus	core.events	In-memory pub/sub signal bus (SolidEventBus, OpenEventBus)
SystemEvent	core.events	Base event model (Pydantic, extra fields allowed)
super_solid_subscriber	core.events	Decorator — subscribes handler with type inference (solid_subscriber, open_subscriber)
super_solid_publisher	core.events	Decorator — auto-publishes return values (solid_publisher, open_publisher)
DiscoveryStrategy	core.discovery	Abstract base — subclass to write custom plugin discovery
SecurityPolicy	core.security	Pydantic model — namespace/caller ACL, discovery toggles
PluginVerifier	core.security	SHA-256 file checksum verification
HealthStatus	core.health	Pydantic model — state, details, timestamp
HealthState	core.health	Enum — HEALTHY, DEGRADED, UNHEALTHY
HealthCheckable	core.health	Protocol — any class with health_check() satisfies it

Running Tests

Run the test suite with coverage report:

poetry run python -m pytest --cov=supersolid --cov-report=term-missing

Name                           Stmts   Miss  Cover
--------------------------------------------------
supersolid/__init__.py             2      0   100%
supersolid/core/__init__.py        8      0   100%
supersolid/core/discovery.py      43      0   100%
supersolid/core/engine.py         49      0   100%
supersolid/core/events.py         85      0   100%
supersolid/core/health.py         14      0   100%
supersolid/core/registry.py      154      0   100%
supersolid/core/security.py       24      0   100%
supersolid/core/system.py         59      0   100%
--------------------------------------------------
TOTAL                            438      0   100%

============================= 43 passed in 1.12s ==============================

License

Apache-2.0