Runtime server and kernel

The simulation server keeps one binding and its mutable simulation state in one process. POST /api/step integrates one configured timestep; reset restores the binding's initial state. Observing state, health or the WebSocket stream does not advance the simulation.

Foreground startup

From a source checkout:

uv sync --locked --extra server --extra rust
.venv/bin/python tools/install_spo_kernel.py --check-only --json
.venv/bin/spo serve domainpacks/minimal_domain/binding_spec.yaml

The default address is 127.0.0.1:8000. Ctrl-C stops the process. The command starts no persistent service and configures no automatic restart. On Windows, use the corresponding executables under .venv/Scripts/.

The default --require-kernel admission first requires the selected binding's simulation engine to dispatch to Rust. A NumPy selection raises RuntimeError before the listener opens. Admission then executes actual native phase and amplitude integration with analytical solutions. A missing or unusable kernel refuses startup.

Install or update the kernel before starting the consumer process. A process that loaded SPO without the kernel can retain its earlier NumPy selection; required-native verification refuses it until the process is restarted. The verification report also requires an identifiable native library file so its hash can be recorded; an embedded module without that file is refused clearly.

An intentionally Python-only environment can install [server] and pass --allow-python. This allows NumPy fallback; an installed usable kernel can still be selected. The base Python package retains its documented numerical fallbacks.

HTTP contracts

Endpoint	Behaviour
GET /	Simulation dashboard.
GET /api/state	Current step, layer/global coherence, regime and amplitude summary.
POST /api/step	Advance one timestep and return the resulting snapshot.
POST /api/reset	Restore the initial simulation state.
GET /api/config	Binding name/dimensions/periods, amplitude mode, selected backend and kernel admission.
GET /api/health	Check simulation snapshot/coherence/regime without advancing it.
GET /api/metrics	Current Prometheus metrics.
GET /api/studio-feed	Current STUDIO feed envelope.
WS /ws/stream	Read-only snapshot stream.

Configuration includes backend ("rust" or "numpy"), kernel_required, and kernel. Required-native admission fills kernel with the verified distribution version and native library sha256; optional admission leaves that verification report null.

With SPO_ENV=production, SPO_API_KEY is mandatory. Mutable endpoints require the matching X-API-Key header; production rate limits apply. Use one worker for the shared in-process simulation. See Production Deployment.

Public Python API

create_app(spec_path, require_kernel=False) returns a FastAPI application. Direct Python callers explicitly select require_kernel=True when native computation is part of their deployment contract. The CLI selects it by default.

verify_kernel() runs public UPDEEngine.step and StuartLandauEngine.step against uncoupled analytical trajectories and returns KernelVerification(version, extension, sha256). It creates its own engines and does not advance an existing simulation. The report binds successful numerical verification to the loaded native library.

kernel

Verify the installed native kernel through public numerical consumers.

Classes

KernelVerification dataclass

KernelVerification(
    version: str, extension: str, sha256: str
)

Identity of a native binary that passed public numerical checks.

Attributes

version : str Installed spo-kernel distribution version. extension : str Absolute path of the loaded extension. sha256 : str SHA-256 of the loaded native library.

Functions:

verify_kernel

verify_kernel() -> KernelVerification

Execute native phase and amplitude steps with analytical oracles.

Both engines integrate uncoupled oscillators: phase advances by omega * dt and a Stuart-Landau amplitude at its unit equilibrium stays one. The public engines must select Rust before either result can qualify native operation. This check does not advance a caller's simulation.

Returns

KernelVerification Installed binary identity after both real numerical operations pass.

Raises

ImportError The native extension cannot be loaded or has no identifiable library file. RuntimeError A public engine selected the NumPy fallback. AssertionError A native result disagrees with its analytical solution.

Source code in src/scpn_phase_orchestrator/runtime/kernel.py

def verify_kernel() -> KernelVerification:
    """Execute native phase and amplitude steps with analytical oracles.

    Both engines integrate uncoupled oscillators: phase advances by omega * dt
    and a Stuart-Landau amplitude at its unit equilibrium stays one. The public
    engines must select Rust before either result can qualify native operation.
    This check does not advance a caller's simulation.

    Returns
    -------
    KernelVerification
        Installed binary identity after both real numerical operations pass.

    Raises
    ------
    ImportError
        The native extension cannot be loaded or has no identifiable library file.
    RuntimeError
        A public engine selected the NumPy fallback.
    AssertionError
        A native result disagrees with its analytical solution.
    """
    extension = importlib.import_module("spo_kernel.spo_kernel")
    phase = UPDEEngine(2, dt=0.01, method="rk4")
    amplitude = StuartLandauEngine(2, dt=0.01, method="rk4")
    if phase.backend != "rust" or amplitude.backend != "rust":
        raise RuntimeError("spo-kernel required: a numerical engine selected NumPy")

    initial = np.array([0.2, 1.1], dtype=np.float64)
    omega = np.array([1.0, -0.5], dtype=np.float64)
    zero = np.zeros((2, 2), dtype=np.float64)
    expected = initial + 0.01 * omega
    result = phase.step(initial, omega, zero, alpha=zero)
    np.testing.assert_allclose(result, expected, rtol=0.0, atol=1e-13)
    result_amplitude = amplitude.step(
        np.concatenate((initial, np.ones(2))),
        omega,
        np.ones(2),
        zero,
        zero,
        0.0,
        0.0,
        alpha=zero,
    )
    np.testing.assert_allclose(
        result_amplitude,
        np.concatenate((expected, np.ones(2))),
        rtol=0.0,
        atol=1e-13,
    )
    binary_path = getattr(extension, "__file__", None)
    if binary_path is None:
        raise ImportError("spo-kernel has no native library path")
    binary = Path(binary_path).resolve()
    return KernelVerification(
        version=version("spo-kernel"),
        extension=str(binary),
        sha256=hashlib.sha256(binary.read_bytes()).hexdigest(),
    )

create_app

create_app(
    spec_path: str | Path, *, require_kernel: bool = False
) -> FastAPI

Create FastAPI app for the given binding spec.

Parameters

spec_path : str | Path Filesystem path to the binding-spec file. require_kernel : bool, default False Require native phase and amplitude verification and native dispatch in the configured simulation before accepting requests.

Returns

fastapi.FastAPI The configured FastAPI application.

Raises

RuntimeError If the configured simulation selects NumPy when native execution is required, or native verification fails. ImportError If a required optional dependency is not installed. HTTPException If the request is invalid.

Source code in src/scpn_phase_orchestrator/runtime/server.py

def create_app(spec_path: str | Path, *, require_kernel: bool = False) -> FastAPI:
    """Create FastAPI app for the given binding spec.

    Parameters
    ----------
    spec_path : str | Path
        Filesystem path to the binding-spec file.
    require_kernel : bool, default False
        Require native phase and amplitude verification and native dispatch
        in the configured simulation before accepting requests.

    Returns
    -------
    fastapi.FastAPI
        The configured FastAPI application.

    Raises
    ------
    RuntimeError
        If the configured simulation selects NumPy when native execution is
        required, or native verification fails.
    ImportError
        If a required optional dependency is not installed.
    HTTPException
        If the request is invalid.
    """
    try:
        from collections.abc import AsyncIterator
        from contextlib import asynccontextmanager

        from fastapi import Depends, FastAPI, Header, HTTPException
        from fastapi.responses import HTMLResponse
    except ImportError as exc:
        msg = "fastapi not installed. pip install fastapi uvicorn"
        raise ImportError(msg) from exc

    from scpn_phase_orchestrator.runtime.network_security import (
        FixedWindowRateLimiter,
        env_int,
        is_production_mode,
    )

    spec = load_binding_spec(spec_path)
    sim = SimulationState(spec)
    kernel_report = None
    if require_kernel:
        selected_engine = sim.sl_engine if sim.amplitude_mode else sim.engine
        if selected_engine is None or selected_engine.backend != "rust":
            raise RuntimeError("spo-kernel required: simulation selected NumPy")
        kernel_report = verify_kernel()

    @asynccontextmanager
    async def _lifespan(_app: FastAPI) -> AsyncIterator[None]:
        """Release engine resources when the process shuts down.

        The simulation state itself is held in-process (numpy arrays), but
        future integrations (gRPC channels, database handles, external
        adapters) register cleanup here so a graceful shutdown never
        leaks a descriptor.
        """
        logger.info(
            "spo server startup: spec=%s n_osc=%d amplitude_mode=%s",
            spec.name,
            sim.n_osc,
            sim.amplitude_mode,
        )
        try:
            yield
        finally:
            logger.info("spo server shutdown: spec=%s", spec.name)
            with sim._lock:
                sim.event_bus.clear()

    app = FastAPI(title="SPO Dashboard", version=__version__, lifespan=_lifespan)

    @app.middleware("http")
    async def _log_http_request(
        request: FastAPIRequest,
        call_next: Callable[[FastAPIRequest], Awaitable[Response]],
    ) -> Response:
        """Log an HTTP request to the audit stream."""
        start = time.perf_counter()
        status_code = 500
        try:
            response = await call_next(request)
            status_code = response.status_code
            return response
        finally:
            duration_ms = (time.perf_counter() - start) * 1000.0
            path = request.url.path
            logger.info(
                "http.request: method=%s path=%s status_code=%d duration_ms=%.3f",
                request.method,
                path,
                status_code,
                duration_ms,
                extra={
                    "http_method": request.method,
                    "http_path": path,
                    "status_code": status_code,
                    "duration_ms": duration_ms,
                },
            )

    _api_key = os.environ.get("SPO_API_KEY")
    _production = is_production_mode("SPO")
    if _production and not _api_key:
        raise RuntimeError("SPO_API_KEY is required when SPO_ENV=production")
    _rate_limit = env_int("SPO_RATE_LIMIT_PER_MINUTE", 120 if _production else 0)
    _limiter = FixedWindowRateLimiter(_rate_limit) if _rate_limit > 0 else None

    async def _require_auth(
        request: FastAPIRequest,
        x_api_key: str | None = Header(None),
    ) -> None:
        """Authorise an HTTP request, raising on failure."""
        if _api_key is None:
            identity = request.client.host if request.client is not None else "local"
        elif x_api_key is None or not hmac.compare_digest(x_api_key, _api_key):
            raise HTTPException(status_code=401, detail="Invalid or missing X-API-Key")
        else:
            identity = x_api_key
        if _limiter is not None and not _limiter.allow(identity):
            raise HTTPException(status_code=429, detail="Rate limit exceeded")

    @app.get("/", response_class=HTMLResponse)
    async def dashboard() -> str:
        """Handle GET / — serve the HTML dashboard."""
        return DASHBOARD_HTML

    @app.get("/api/state")
    async def get_state() -> dict[str, Any]:
        """Handle GET /api/state — return current simulation snapshot."""
        with sim._lock:
            return sim.snapshot()

    @app.get("/api/studio-feed")
    async def get_studio_feed() -> dict[str, object]:
        """Handle GET /api/studio-feed — return live STUDIO feed envelope."""
        with sim._lock:
            return sim.studio_feed()

    @app.post("/api/step", dependencies=[Depends(_require_auth)])
    async def post_step() -> dict[str, Any]:
        """Handle POST /api/step — advance simulation one tick."""
        with sim._lock:
            snap = sim.step()
        logger.debug(
            "api.step: step=%d R_global=%.4f regime=%s",
            snap.get("step", -1),
            snap.get("R_global", float("nan")),
            snap.get("regime", ""),
        )
        return snap

    @app.post("/api/reset", dependencies=[Depends(_require_auth)])
    async def post_reset() -> dict[str, Any]:
        """Handle POST /api/reset — reset simulation to initial state."""
        with sim._lock:
            snap = sim.reset()
        logger.info(
            "api.reset: step=%d regime=%s",
            snap.get("step", 0),
            snap.get("regime", ""),
        )
        return snap

    @app.get("/api/config")
    async def get_config() -> dict[str, Any]:
        """Handle GET /api/config — return engine configuration."""
        return {
            "name": spec.name,
            "n_oscillators": sim.n_osc,
            "n_layers": len(spec.layers),
            "amplitude_mode": sim.amplitude_mode,
            "sample_period_s": spec.sample_period_s,
            "control_period_s": spec.control_period_s,
            "backend": (
                sim.sl_engine.backend
                if sim.sl_engine is not None
                else sim.engine.backend
            ),
            "kernel_required": require_kernel,
            "kernel": (
                {
                    "version": kernel_report.version,
                    "sha256": kernel_report.sha256,
                }
                if kernel_report is not None
                else None
            ),
        }

    @app.get("/api/metrics")
    async def get_metrics() -> Response:
        """Handle GET /api/metrics — export Prometheus-format metrics."""
        from fastapi.responses import PlainTextResponse

        with sim._lock:
            snap = sim.snapshot()
        upde_state = UPDEState(
            layers=[
                LayerState(R=ly["R"], psi=ly.get("psi", 0.0)) for ly in snap["layers"]
            ],
            cross_layer_alignment=np.eye(len(snap["layers"])),
            stability_proxy=snap["R_global"],
            regime_id=snap["regime"],
        )
        observability = RuntimeObservability()
        text = observability.prometheus_text(
            RuntimeMetricSnapshot(
                upde_state=upde_state,
                regime=snap["regime"],
                latency_ms=0.0,
                step_idx=snap.get("step"),
            )
        )
        return PlainTextResponse(text, media_type="text/plain")

    @app.get("/api/health")
    async def health() -> dict[str, Any]:
        """Deep health check — verifies engine, monitor, and regime subsystems."""
        checks: dict[str, str] = {}
        try:
            with sim._lock:
                snap = sim.snapshot()
            checks["engine"] = "ok" if snap.get("step", -1) >= 0 else "degraded"
            r_val = snap.get("R_global", float("nan"))
            checks["R_finite"] = "ok" if np.isfinite(r_val) else "error"
            checks["regime"] = "ok" if snap.get("regime") else "unknown"
        except HEALTH_CHECK_EXCEPTIONS as exc:
            # Log the detail server-side; do not leak the exception text into the
            # health response (CodeQL py/stack-trace-exposure — information
            # exposure through an exception).
            logger.warning("health-check engine probe failed: %s", exc)
            checks["engine"] = "error"

        healthy = all(v == "ok" for v in checks.values())
        return {"status": "healthy" if healthy else "degraded", "checks": checks}

    @app.websocket("/ws/stream")
    async def ws_stream(websocket: WebSocket) -> None:
        """Read-only observer: streams snapshots without advancing simulation."""
        await websocket.accept()
        try:
            while True:
                with sim._lock:
                    state = sim.snapshot()
                await websocket.send_text(json.dumps(state))
                await asyncio.sleep(spec.sample_period_s)
        except WebSocketDisconnect:
            pass

    return app

Command module

spo serve is implemented in scpn_phase_orchestrator.runtime.cli.serve.

serve

Run a simulation API in the foreground with explicit native admission.

Functions:

serve

serve(
    spec_path: Path,
    host: str,
    port: int,
    require_kernel: bool,
) -> None

Serve a binding's simulation until interrupted or terminated.

Parameters

spec_path : pathlib.Path Binding specification for the in-process simulation. host : str Listening address; defaults to host loopback. port : int TCP port in the range 1 through 65535. require_kernel : bool Refuse unavailable native computation unless Python is explicitly allowed.

Raises

click.ClickException A server dependency, binding or required native kernel is unusable.

Source code in src/scpn_phase_orchestrator/runtime/cli/serve.py

@main.command(help="Serve a binding's simulation in the foreground; stop with Ctrl-C.")
@click.argument(
    "spec_path", type=click.Path(exists=True, dir_okay=False, path_type=Path)
)
@click.option("--host", default="127.0.0.1", show_default=True)
@click.option("--port", type=click.IntRange(1, 65535), default=8000, show_default=True)
@click.option(
    "--require-kernel/--allow-python",
    default=True,
    show_default=True,
    help="Require verified native computation before serving requests.",
)
def serve(spec_path: Path, host: str, port: int, require_kernel: bool) -> None:
    """Serve a binding's simulation until interrupted or terminated.

    Parameters
    ----------
    spec_path : pathlib.Path
        Binding specification for the in-process simulation.
    host : str
        Listening address; defaults to host loopback.
    port : int
        TCP port in the range 1 through 65535.
    require_kernel : bool
        Refuse unavailable native computation unless Python is explicitly allowed.

    Raises
    ------
    click.ClickException
        A server dependency, binding or required native kernel is unusable.
    """
    try:
        import uvicorn

        from scpn_phase_orchestrator.runtime.server import create_app

        app = create_app(spec_path, require_kernel=require_kernel)
    except (ImportError, RuntimeError, ValueError, AssertionError) as exc:
        raise click.ClickException(str(exc)) from exc
    uvicorn.run(app, host=host, port=port, access_log=True)