sx-host plan steps 1-2: defhelper + SX config + SXTP spec + nav tools

Step 1 — defhelper: SX-defined page data helpers replace Python helpers.
(defhelper name (params) body) in .sx files, using existing IO primitives
(query, action, service). Loaded into OCaml kernel as pure SX defines.

Step 2 — SX config: app-config.sx replaces app-config.yaml with (defconfig)
form. (env-get "VAR") resolves secrets from environment. Kebab-to-underscore
aliasing ensures backward compatibility with all 174 config consumers.

Also: SXTP protocol spec (applications/sxtp/spec.sx), docs article,
sx_nav move/delete modes, reactive-runtime moved to geography.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

shared/config.py

@@ -1,4 +1,4 @@
-# suma_browser/config.py
+# shared/config.py — SX-first config loader with YAML fallback
 from __future__ import annotations
 
 import asyncio
@@ -6,13 +6,16 @@ import os
 from types import MappingProxyType
 from typing import Any, Optional
 import copy
-import yaml
 
-# Default config path (override with APP_CONFIG_FILE)
-_DEFAULT_CONFIG_PATH = os.environ.get(
+# Default config paths (override with APP_CONFIG_FILE)
+_DEFAULT_YAML_PATH = os.environ.get(
     "APP_CONFIG_FILE",
     os.path.join(os.getcwd(), "config/app-config.yaml"),
 )
+_DEFAULT_SX_PATH = os.environ.get(
+    "APP_CONFIG_SX_FILE",
+    os.path.join(os.getcwd(), "config/app-config.sx"),
+)
 
 # Module state
 _init_lock = asyncio.Lock()
@@ -23,7 +26,6 @@ _data_plain: Any = None    # plain builtins for pretty-print / logging
 def _freeze(obj: Any) -> Any:
     """Deep-freeze containers to read-only equivalents."""
     if isinstance(obj, dict):
-        # freeze children first, then wrap dict in mappingproxy
         return MappingProxyType({k: _freeze(v) for k, v in obj.items()})
     if isinstance(obj, list):
         return tuple(_freeze(v) for v in obj)
@@ -33,10 +35,88 @@ def _freeze(obj: Any) -> Any:
         return tuple(_freeze(v) for v in obj)
     return obj
 
+
+def _sx_to_dict(expr: Any) -> Any:
+    """Convert parsed SX config values to plain Python dicts/lists.
+
+    - Keyword keys become strings (kebab-case preserved, also aliased to
+      underscore form for backward compatibility with YAML consumers).
+    - (env-get "VAR") calls are resolved to os.environ.
+    - Lists become plain Python lists.
+    - Everything else passes through as-is.
+    """
+    from shared.sx.types import Keyword, Symbol
+
+    # (env-get "VAR") → os.environ.get("VAR")
+    if isinstance(expr, list) and len(expr) == 2:
+        head = expr[0]
+        if isinstance(head, Symbol) and head.name == "env-get":
+            var_name = str(expr[1])
+            return os.environ.get(var_name)
+
+    # dict with keyword keys
+    if isinstance(expr, dict):
+        result: dict[str, Any] = {}
+        for k, v in expr.items():
+            key = k if isinstance(k, str) else str(k)
+            val = _sx_to_dict(v)
+            result[key] = val
+            # Alias kebab-case → underscore for backward compat
+            underscore = key.replace("-", "_")
+            if underscore != key:
+                result[underscore] = val
+        return result
+
+    if isinstance(expr, list):
+        # Check for (env-get ...) first (already handled above for len==2)
+        return [_sx_to_dict(item) for item in expr]
+
+    if isinstance(expr, tuple):
+        return [_sx_to_dict(item) for item in expr]
+
+    if isinstance(expr, Keyword):
+        return str(expr)
+
+    if isinstance(expr, Symbol):
+        name = expr.name
+        if name == "true":
+            return True
+        if name == "false":
+            return False
+        if name == "nil":
+            return None
+        return name
+
+    return expr
+
+
+def _load_sx_config(path: str) -> dict:
+    """Load an SX config file and return a plain dict.
+
+    Expects a single (defconfig name body) form.
+    """
+    from shared.sx.parser import parse_all
+
+    with open(path, "r", encoding="utf-8") as f:
+        source = f.read()
+
+    exprs = parse_all(source)
+    for expr in exprs:
+        if (isinstance(expr, list) and len(expr) >= 3
+                and hasattr(expr[0], 'name') and expr[0].name == "defconfig"):
+            # (defconfig name {body})
+            body = expr[2]
+            return _sx_to_dict(body)
+
+    raise ValueError(f"No (defconfig ...) form found in {path}")
+
+
 # ---------------- API ----------------
 async def init_config(path: Optional[str] = None, *, force: bool = False) -> None:
     """
-    Load YAML exactly as-is and cache both a frozen (read-only) and a plain copy.
+    Load config and cache both a frozen (read-only) and a plain copy.
+
+    Prefers SX config (app-config.sx) when available, falls back to YAML.
     Idempotent; pass force=True to reload.
     """
     global _data_frozen, _data_plain
@@ -48,14 +128,20 @@ async def init_config(path: Optional[str] = None, *, force: bool = False) -> Non
         if _data_frozen is not None and not force:
             return
 
-        cfg_path = path or _DEFAULT_CONFIG_PATH
-        if not os.path.exists(cfg_path):
-            raise FileNotFoundError(f"Config file not found: {cfg_path}")
+        # Try SX first, then YAML
+        sx_path = path if (path and path.endswith(".sx")) else _DEFAULT_SX_PATH
+        yaml_path = path if (path and not path.endswith(".sx")) else _DEFAULT_YAML_PATH
 
-        with open(cfg_path, "r", encoding="utf-8") as f:
-            raw = yaml.safe_load(f)  # whatever the YAML root is
+        if os.path.exists(sx_path):
+            raw = _load_sx_config(sx_path)
+        elif os.path.exists(yaml_path):
+            import yaml
+            with open(yaml_path, "r", encoding="utf-8") as f:
+                raw = yaml.safe_load(f)
+        else:
+            raise FileNotFoundError(
+                f"No config found: tried {sx_path} and {yaml_path}")
 
-        # store plain as loaded; store frozen for normal use
         _data_plain = raw
         _data_frozen = _freeze(raw)
 
@@ -77,8 +163,13 @@ def as_plain() -> Any:
 
 def pretty() -> str:
     """
-    YAML pretty string without mappingproxy noise.
+    Pretty string for logging. Uses YAML if available, else pprint.
     """
     if _data_plain is None:
         raise RuntimeError("init_config() has not been awaited yet.")
-    return yaml.safe_dump(_data_plain, sort_keys=False, allow_unicode=True)
+    try:
+        import yaml
+        return yaml.safe_dump(_data_plain, sort_keys=False, allow_unicode=True)
+    except ImportError:
+        import pprint
+        return pprint.pformat(_data_plain)

shared/sx/ocaml_bridge.py

@@ -308,14 +308,30 @@ class OcamlBridge:
             return
         self._helpers_injected = True
         try:
-            from .pages import get_page_helpers
+            from .pages import get_page_helpers, get_sx_helpers
             import inspect
-            helpers = get_page_helpers("sx")
-            if not helpers:
-                self._helpers_injected = False
-                return
             count = 0
+
+            # 1. Inject SX-defined helpers (defhelper) — pure SX, no Python bridge
+            #    Load from all services since they're pure SX defines.
+            sx_helpers: dict[str, str] = {}
+            from .pages import _SX_HELPERS
+            for svc_helpers in _SX_HELPERS.values():
+                sx_helpers.update(svc_helpers)
+            for name, sx_source in sx_helpers.items():
+                try:
+                    await self._send_command(f'(load-source "{_escape(sx_source)}")')
+                    await self._read_until_ok(ctx=None)
+                    count += 1
+                except OcamlBridgeError:
+                    _logger.warning("Failed to inject SX helper: %s", name)
+
+            # 2. Inject Python helpers — wrapped as (helper "name" ...) IO bridge calls
+            helpers = get_page_helpers("sx")
             for name, fn in helpers.items():
+                # Skip if already defined by defhelper (SX takes priority)
+                if name in sx_helpers:
+                    continue
                 if callable(fn) and not name.startswith("~"):
                     try:
                         sig = inspect.signature(fn)
@@ -333,7 +349,12 @@ class OcamlBridge:
                         count += 1
                     except OcamlBridgeError:
                         pass
-            _logger.info("Injected %d page helpers into OCaml kernel", count)
+
+            if not count and not helpers and not sx_helpers:
+                self._helpers_injected = False
+                return
+            _logger.info("Injected %d page helpers into OCaml kernel (%d SX, %d Python)",
+                         count, len(sx_helpers), count - len(sx_helpers))
         except Exception as e:
             _logger.warning("Helper injection failed: %s", e)
             self._helpers_injected = False

shared/sx/pages.py

@@ -51,6 +51,7 @@ def _eval_error_sx(e: EvalError, context: str) -> str:
 
 _PAGE_REGISTRY: dict[str, dict[str, PageDef]] = {}
 _PAGE_HELPERS: dict[str, dict[str, Any]] = {}  # service → name → callable
+_SX_HELPERS: dict[str, dict[str, str]] = {}    # service → name → SX source
 
 
 def register_page(service: str, page_def: PageDef) -> None:
@@ -137,6 +138,19 @@ def get_page_helpers(service: str) -> dict[str, Any]:
     return dict(_PAGE_HELPERS.get(service, {}))
 
 
+def register_sx_helper(service: str, name: str, source: str) -> None:
+    """Register an SX-defined helper (from defhelper) for a service."""
+    if service not in _SX_HELPERS:
+        _SX_HELPERS[service] = {}
+    _SX_HELPERS[service][name] = source
+    logger.debug("Registered SX helper %s:%s", service, name)
+
+
+def get_sx_helpers(service: str) -> dict[str, str]:
+    """Return SX-defined helpers for a service (name → SX source)."""
+    return dict(_SX_HELPERS.get(service, {}))
+
+
 # ---------------------------------------------------------------------------
 # Loading — parse .sx files and collect PageDef instances
 # ---------------------------------------------------------------------------
@@ -179,8 +193,8 @@ def _parse_defpage(expr: list) -> PageDef | None:
 
 
 def load_page_file(filepath: str, service_name: str) -> list[PageDef]:
-    """Parse an .sx file and register any defpage definitions."""
-    from .parser import parse_all
+    """Parse an .sx file and register any defpage/defhelper definitions."""
+    from .parser import parse_all, serialize
 
     with open(filepath, encoding="utf-8") as f:
         source = f.read()
@@ -189,16 +203,42 @@ def load_page_file(filepath: str, service_name: str) -> list[PageDef]:
     pages: list[PageDef] = []
 
     for expr in exprs:
-        if (isinstance(expr, list) and expr
-                and hasattr(expr[0], 'name') and expr[0].name == "defpage"):
+        if not isinstance(expr, list) or not expr:
+            continue
+        head = getattr(expr[0], 'name', None)
+        if head == "defpage":
             pd = _parse_defpage(expr)
             if pd:
                 register_page(service_name, pd)
                 pages.append(pd)
+        elif head == "defhelper":
+            _parse_defhelper(expr, service_name, serialize)
 
     return pages
 
 
+def _parse_defhelper(expr: list, service_name: str, serialize) -> None:
+    """Parse (defhelper name (params...) body...) and register as SX helper.
+
+    Translates to a (define name (fn (params...) body...)) SX source string
+    that will be loaded into the OCaml kernel at render time.
+    """
+    if len(expr) < 4:
+        logger.warning("defhelper: too few forms: %s", expr[:2])
+        return
+    name = expr[1].name if hasattr(expr[1], 'name') else str(expr[1])
+    params = expr[2]
+    body = expr[3:]
+
+    # Build the equivalent define/fn form
+    body_sx = " ".join(serialize(b) for b in body)
+    if len(body) > 1:
+        body_sx = f"(do {body_sx})"
+    params_sx = serialize(params)
+    sx_source = f'(define {name} (fn {params_sx} {body_sx}))'
+    register_sx_helper(service_name, name, sx_source)
+
+
 def load_page_dir(directory: str, service_name: str) -> list[PageDef]:
     """Load all .sx files from a directory and register pages."""
     import glob as glob_mod