0. Layer Rules & Decision Guide

What you’re allowed to touch depends on where you are. This page answers: which API do I use?

Read this first. Every other tutorial teaches mechanics — how things work. This page tells you which mechanism to reach for when you’re writing real code.

The Three Layers

Every piece of code in a SignalPy system lives in exactly one of these layers. Each layer has a strict boundary — cross it and you’ve introduced a bug that won’t show up until something changes at runtime.

┌─────────────────────────────────────────────────────────┐
│  APP LAYER  — your business logic                       │
│  You ARE a component. You use decorators + self.rt.     │
├─────────────────────────────────────────────────────────┤
│  PLATFORM LAYER  — boot + wiring + HTTP shell           │
│  You ASSEMBLE components. You expose bus to transport.  │
├─────────────────────────────────────────────────────────┤
│  KERNEL  — the reactive engine (you don't touch this)   │
│  Signal, Computed, Effect, Registry, Bus, Lifecycle.    │
└─────────────────────────────────────────────────────────┘

What each layer is allowed to touch

	App Layer	Platform Layer	Kernel (don’t touch)
Imports from kernel	decorators, contracts	Kernel class only	—
Creates a Kernel?	Never	Yes, exactly once	—
Calls registry.require?	Never	Only at boot script level	—
Calls other components?	Via @requires + self.rt.X.method()	Via transport adapters	–
Reads config?	Via self.rt.config.get()	Via a runnable on a transport	–
Holds kernel reference?	Never	Only in boot script, never on app.state	—

---

The Core Mechanism: @requires + Direct Calls

Components interact through @requires injection and direct method calls. There is one mechanism for all inter-component communication within the kernel.

@requires – “I need this service”

@requires(config=IConfig, logger=ILogger)
class MyApp: ...

What it means: “Wire this service in before I activate.”

What happens:

1. The kernel reads your @requires at boot
2. It refuses to activate you until every required service exists
3. It injects the service as a Signal on self.rt
4. Every read is reactive – @effect and @computed auto-track it
5. If the service is hot-swapped later, your effects re-run automatically

Properties:

• Resolved at boot time (before your @lifecycle.activate runs)
• Guaranteed to exist when you access it – the kernel ensures this
• Reactive – reading self.rt.config.get("x") inside @effect subscribes you
• Direct method calls – self.rt.config.get("key"), type-safe and IDE-friendly
• Structural – shows up in dependency graph, affects activation order

Use it for: any service your component needs – infrastructure or peer.

# Infrastructure services
@requires(config=IConfig, logger=ILogger)

# Peer services -- same mechanism
@requires(email=IEmailService, search=ISearchService)

# Aggregate injection
@requires(dicts=list[IDictionary])

# Optional dependencies
@requires(cache=ICache, optional=True)

Calling methods on injected services

Once you have a service via @requires, call its methods directly:

@component("my-app")
@requires(config=IConfig, email=IEmailService, search=ISearchService)
class MyApp:

    @computed
    def url(self):
        return self.rt.config.get("my-app.url", "http://localhost")

    async def submit(self, params):
        await self.rt.email.send(to=params.to, body=params.body)

    async def find(self, query):
        return await self.rt.search.query(q=query)

No string-based dispatch. No bus routing. Direct, type-safe method calls.

Optional dependencies for loose coupling

If a peer service might not be deployed, use optional=True:

@requires(email=IEmailService, optional=True)
class MyApp:
    async def submit(self, params):
        if self.rt.email:
            await self.rt.email.send(to=params.to, body=params.body)
        else:
            self.rt.logger.warning("Email service not available")

Transport adapters route external requests

Transport adapters (REST, MCP, CLI) are the bridge between the outside world and your component graph. They discover @runnable schemas via kernel.runnables() and call the handler directly:

HTTP POST /api/v1/orders/place   --> RESTTransport --> schema.handler(validated)
MCP tool call "search.query"     --> MCPTransport  --> schema.handler(validated)
CLI: myapp orders place          --> CLITransport  --> schema.handler(validated)

The component never knows what called it. The transport calls the runnable handler directly – no bus dispatch.

Concrete example

# Infrastructure deps + peer deps, all through @requires
@component("order-service")
@requires(config=IConfig, logger=ILogger, email=IEmailService)
class OrderService:
    @computed
    def url(self):
        # Reactive! Auto-recomputes when config changes.
        return self.rt.config.get("orders.url", "http://localhost")

    @runnable("place", params=OrderParams, description="Place an order")
    async def place_order(self, params):
        order = await self._save_order(params)
        # Direct call to peer service -- type-safe, IDE-friendly
        await self.rt.email.send(
            to=params.customer_email,
            body=f"Order {order.id} confirmed"
        )
        return {"order_id": order.id}

---

Platform Layer: Custom Routes

Most of the time, you use RESTTransport or mount_rest() and routes are auto-generated from @runnable schemas. But for custom routes (file uploads, SSE streams, WebSocket), you write them by hand.

How to set up the platform layer

# platform/boot.py -- the boot script

from signalpy.kernel import Kernel
from signalpy.providers.config import ConfigProvider
from signalpy.providers.logging_provider import LoggingProvider
from signalpy.adapters.rest import RESTTransport

async def boot():
    kernel = Kernel()
    kernel.discover([
        ConfigProvider, LoggingProvider,
        RESTTransport,
        # ... your app components
    ])
    await kernel.boot()
    return kernel

Custom route handlers

For routes beyond what RESTTransport auto-generates, use kernel.runnables() to find schemas and call handlers directly:

# platform/routes.py -- custom route handlers

from fastapi import APIRouter, Request

router = APIRouter()

@router.post("/api/tasks/upload")
async def upload_task(request: Request):
    # Custom route for file upload -- beyond auto-generated routes
    file = await request.form()
    # Call the component's runnable handler directly
    schema = request.app.state.kernel.runnables()["task-service.process_upload"]
    result = await schema.handler(ProcessUploadParams(file=file))
    return {"ok": True, "data": result}

When to write custom routes

• Simple case: Use RESTTransport or mount_rest(). No hand-written routes needed.
• Custom routes: For routes that do something beyond standard POST-to-runnable (e.g., file uploads, SSE streams, WebSocket), write them by hand using schema handlers.

---

The @subscribe Decorator – Events (Tell, Don’t Ask)

There’s a second communication pattern beyond @requires: events via @subscribe.

@component("audit-logger")
@requires(logger=ILogger)
class AuditLogger:
    @subscribe("orders.*", description="Log all order events")
    async def on_order_event(self, event_type, data):
        self.rt.logger.info(f"Audit: {event_type}", data=data)

How it differs from @requires + direct calls:

	@requires + method call	@subscribe
Direction	Request -> Response (ask)	Fire -> Forget (tell)
Coupling	Caller declares dependency	Publisher doesn’t know subscribers
Return value	Yes (result from method)	No (void)
Failure	Caller sees error	Publisher never knows
Pattern	Command / Query	Event / Notification
Analogy	“Hey search, find me X”	“FYI: an order was placed”

When to use events:

• You want to announce something happened without caring who listens
• Multiple components might react to the same event
• The publisher shouldn’t depend on the subscriber (reverse coupling is wrong)

# Inside a component that processes orders:
@runnable("place-order", params=OrderParams, description="Place an order")
async def place_order(self, params):
    order = await self._save_order(params)
    # Tell the world. Don't care who's listening.
    self.rt.publish("orders.placed", {"order_id": order.id, "total": order.total})
    return {"order_id": order.id}

Now any component can @subscribe("orders.placed") — audit logger, analytics, notification sender — without the order service knowing about any of them.

---

Summary: Three Orthogonal Mechanisms

    @requires + direct calls            @runnable + transports
    (components talk to each other)     (outside world reaches components)
            |                                     |
            |   structural, reactive,             |   schema-only declaration,
            |   boot-time, guaranteed,            |   transport adapters discover
            |   type-safe                         |   and call schema.handler
            |                                     |
    --------+-------------------------------------+--------
    |                   Component                          |
    --------+----------------------------------------------+
            |
     self.rt.publish / @subscribe
     (components react to events)
            |
      fire-and-forget, loose coupling, many listeners

Three mechanisms, three purposes, orthogonal:

1. @requires + direct calls – dependency injection. “I need this service, I call its methods.” Components talk to each other.
2. @runnable + transports – schema declarations. “This operation exists, here’s its shape.” The outside world reaches components.
3. @subscribe + publish – event notification. “Something happened, FYI.” Components react to events without coupling.

These three don’t overlap — components don’t go through the bus to call each other, runnables don’t use @requires to find their consumers, and the bus doesn’t dispatch to runnables.

---

Quick Reference: “I want to X, which do I use?”

I want to…	Use
Read config reactively	@requires(config=IConfig) -> self.rt.config.get(...)
Log something	@requires(logger=ILogger) -> self.rt.logger.info(...)
Read credentials	@requires(creds=ICredentials) -> self.rt.creds.get(...)
Store/retrieve data	@requires(storage=IStorage) -> await self.rt.storage.put(...)
Call another component’s action	@requires(svc=IService) -> await self.rt.svc.action(...)
Announce an event	self.rt.publish("event.type", {...})
React to another component’s event	@subscribe("event.type") on a method
React when a dependency changes	@effect – reads are auto-tracked
Cache a derived value	@computed – recomputes only when deps change
Expose an action to the outside world	@runnable("name", params=Model)
Make a runnable accessible via REST/MCP/CLI	rest={...} / mcp={...} on @component
Wire up the system at startup	Platform layer: Kernel(), .discover(), .boot()
Get a service reference in a boot script	kernel.registry.require("IContract") (boot only!)