runtime.py

"""Mirrors @berthos/sdk's runtime.ts boot sequence: load manifest -> import the
app module -> assert exports match manifest -> run hooks -> serve RPC. The
orchestration itself is idiomatic Python (importlib, not a port of Node's
dynamic import()) — only the wire protocols (manifest shape, RPC framing)
are shared with the TypeScript SDK, not this glue code.
"""

from __future__ import annotations

import importlib.util
import os
import sys
from pathlib import Path

from .app import AppContext, BerthApp
from .context_bus import create_unix_socket_context_bus
from .local_context_bus import LocalContextBus
from .manifest import load_manifest
from .rpc import serve_stdio_forever, start_rpc_server


def _create_context_bus():
    """Real context-bus daemon if reachable (entrypoint.sh starts it before
    this runtime — same file/env contract @berthos/sdk's runtime.ts uses);
    falls back to a local no-op otherwise, so an app never hard-fails just
    because it's running outside a sandbox with the daemon (e.g. a bare
    `python3 -m berth_sdk.runtime` during a quick script)."""
    socket_path = os.environ.get("BERTH_CONTEXT_BUS_SOCKET", "/tmp/berth-context-bus.sock")
    try:
        return create_unix_socket_context_bus(socket_path)
    except Exception as err:  # noqa: BLE001 - genuinely any connection failure should fall back, not crash boot
        print(
            f"[berth:runtime] context-bus daemon not reachable at {socket_path} ({err}) — falling back to local no-op",
            file=sys.stderr,
        )
        return LocalContextBus()


def _assert_exports_match_manifest(app: BerthApp, declared_exports: list[str]) -> None:
    code_exports = set(app.exports.keys())
    manifest_exports = set(declared_exports)

    missing_in_code = sorted(manifest_exports - code_exports)
    missing_in_manifest = sorted(code_exports - manifest_exports)

    if missing_in_code or missing_in_manifest:
        problems = []
        if missing_in_code:
            problems.append(f"declared in berth.yml but not implemented: {', '.join(missing_in_code)}")
        if missing_in_manifest:
            problems.append(f"implemented in code but not declared in berth.yml: {', '.join(missing_in_manifest)}")
        raise RuntimeError(f"exports mismatch between berth.yml and app code — {'; '.join(problems)}")


def default_paths() -> tuple[str, str]:
    """(manifest path, app entry), from the same env vars runtime.ts reads."""
    app_root = Path(os.getcwd())
    manifest_path = os.environ.get("BERTH_MANIFEST_PATH", str(app_root / "berth.yml"))
    app_entry = os.environ.get("BERTH_APP_ENTRY", str(app_root / "src" / "app.py"))
    return manifest_path, app_entry


def load_app(app_entry: str) -> BerthApp:
    """Imports the app module and returns its module-level `app`."""
    spec = importlib.util.spec_from_file_location("berth_app_entry", app_entry)
    if spec is None or spec.loader is None:
        raise RuntimeError(f"could not load a Python module from {app_entry}")
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    app = getattr(module, "app", None)
    if not isinstance(app, BerthApp):
        raise RuntimeError(f"{app_entry} must define a module-level `app = define_app(...)`")
    return app


def main() -> None:
    manifest_path, app_entry = default_paths()

    print(f"[berth:runtime] loading manifest from {manifest_path}", file=sys.stderr)
    manifest = load_manifest(manifest_path)

    print(f"[berth:runtime] loading app entry {app_entry}", file=sys.stderr)
    app = load_app(app_entry)

    _assert_exports_match_manifest(app, [e.name for e in manifest.exports])

    for hook in app.on_install_hooks:
        hook()

    context_bus = _create_context_bus()
    ctx = AppContext(manifest, context_bus)
    for hook in app.on_agent_ready_hooks:
        hook(ctx)

    socket_thread = start_rpc_server(app, socket_path=os.environ.get("BERTH_RPC_SOCKET"))
    print(f'[berth:runtime] "{manifest.name}" ready', file=sys.stderr)

    # Reads stdio RPC requests until stdin closes.
    serve_stdio_forever(app)

    # In a multi-app sandbox every app's stdin is /dev/null and it is reached
    # only through its socket, so stdin ends at once. The socket server runs on
    # a daemon thread, which died with the process: the app logged "ready",
    # then exited, and every call to it was refused. Keep serving the socket,
    # which is what the Node runtime does (its open server keeps the event
    # loop alive).
    if socket_thread is not None:
        socket_thread.join()


if __name__ == "__main__":
    main()