Files
rose-ash/shared/sx/types.py
giles 4c4806c8dd
All checks were successful
Build and Deploy / build-and-deploy (push) Successful in 4m11s
Fix all 9 spec test failures: Env scope chain, IO detection, offline mutation
- env.py: Add MergedEnv with dual-parent lookup (primary for set!,
  secondary for reads), add dict-compat methods to Env
- platform_py.py: make_lambda stores env reference (no copy), env_merge
  uses MergedEnv for proper set! propagation, ancestor detection prevents
  unbounded chains in TCO recursion, sf_set_bang walks scope chain
- types.py: Component/Island io_refs defaults to None (not computed)
  instead of empty set, so component-pure? falls through to scan
- run.py: Test env uses Env class, mock execute-action calls SX lambdas
  via _call_sx instead of direct Python call

Spec tests: 320/320 (was 311/320)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-11 09:42:04 +00:00

384 lines
12 KiB
Python

"""
Core types for the s-expression language.
Symbol — unquoted identifier (e.g. div, ~card, map)
Keyword — colon-prefixed key (e.g. :class, :id)
Lambda — callable closure created by (lambda ...) or (fn ...)
Nil — singleton null value
"""
from __future__ import annotations
from dataclasses import dataclass, field
from typing import Any
# ---------------------------------------------------------------------------
# Nil
# ---------------------------------------------------------------------------
class _Nil:
"""Singleton nil value — falsy, serialises as 'nil'."""
_instance: _Nil | None = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
def __bool__(self):
return False
def __repr__(self):
return "nil"
def __eq__(self, other):
return other is None or isinstance(other, _Nil)
def __hash__(self):
return hash(None)
NIL = _Nil()
# ---------------------------------------------------------------------------
# Symbol
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Symbol:
"""An unquoted symbol/identifier."""
name: str
def __repr__(self):
return f"Symbol({self.name!r})"
def __eq__(self, other):
if isinstance(other, Symbol):
return self.name == other.name
if isinstance(other, str):
return self.name == other
return False
def __hash__(self):
return hash(self.name)
@property
def is_component(self) -> bool:
"""True if this symbol names a component (~prefix)."""
return self.name.startswith("~")
# ---------------------------------------------------------------------------
# Keyword
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class Keyword:
"""A keyword starting with colon (e.g. :class, :id)."""
name: str
def __repr__(self):
return f"Keyword({self.name!r})"
def __eq__(self, other):
if isinstance(other, Keyword):
return self.name == other.name
return False
def __hash__(self):
return hash((":", self.name))
# ---------------------------------------------------------------------------
# Lambda
# ---------------------------------------------------------------------------
@dataclass
class Lambda:
"""A callable closure.
Created by ``(lambda (x) body)`` or ``(fn (x) body)``.
Captures the defining environment so free variables resolve correctly.
"""
params: list[str]
body: Any
closure: dict[str, Any] = field(default_factory=dict)
name: str | None = None # optional, set by (define name (fn ...))
def __repr__(self):
tag = self.name or "lambda"
return f"<{tag}({', '.join(self.params)})>"
def __call__(self, *args: Any, evaluator: Any = None, caller_env: dict | None = None) -> Any:
"""Invoke the lambda. Requires *evaluator* — the evaluate() function."""
if evaluator is None:
raise RuntimeError("Lambda requires evaluator to be called")
if len(args) != len(self.params):
raise RuntimeError(
f"{self!r} expects {len(self.params)} args, got {len(args)}"
)
local = dict(self.closure)
if caller_env:
local.update(caller_env)
for p, v in zip(self.params, args):
local[p] = v
return evaluator(self.body, local)
# ---------------------------------------------------------------------------
# Macro
# ---------------------------------------------------------------------------
@dataclass
class Macro:
"""A macro — an AST-transforming function.
Created by ``(defmacro name (params... &rest rest) body)``.
Receives unevaluated arguments, evaluates its body to produce a new
s-expression, which is then evaluated in the caller's environment.
"""
params: list[str]
rest_param: str | None # &rest parameter name
body: Any # unevaluated — returns an s-expression to eval
closure: dict[str, Any] = field(default_factory=dict)
name: str | None = None
def __repr__(self):
tag = self.name or "macro"
return f"<{tag}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# Component
# ---------------------------------------------------------------------------
@dataclass
class Component:
"""A reusable UI component defined via ``(defcomp ~name (&key ...) body)``.
Components are like lambdas but accept keyword arguments and support
a ``children`` rest parameter.
"""
name: str
params: list[str] # keyword parameter names (without &key prefix)
has_children: bool # True if &rest children declared
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
css_classes: set[str] = field(default_factory=set) # pre-scanned :class values
deps: set[str] = field(default_factory=set) # transitive component deps (~names)
io_refs: set[str] | None = None # transitive IO primitive refs (None = not computed)
affinity: str = "auto" # "auto" | "client" | "server"
@property
def is_pure(self) -> bool:
"""True if this component has no transitive IO dependencies."""
return not self.io_refs
@property
def render_target(self) -> str:
"""Where this component should render: 'server' or 'client'."""
if self.affinity == "server":
return "server"
if self.affinity == "client":
return "client"
return "server" if self.io_refs else "client"
def __repr__(self):
return f"<Component ~{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# Island
# ---------------------------------------------------------------------------
@dataclass
class Island:
"""A reactive UI component defined via ``(defisland ~name (&key ...) body)``.
Islands are like components but create a reactive boundary. Inside an
island, signals are tracked — deref subscribes DOM nodes to signals.
On the server, islands render as static HTML with hydration attributes.
"""
name: str
params: list[str]
has_children: bool
body: Any
closure: dict[str, Any] = field(default_factory=dict)
css_classes: set[str] = field(default_factory=set)
deps: set[str] = field(default_factory=set)
io_refs: set[str] | None = None
def __repr__(self):
return f"<Island ~{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# HandlerDef
# ---------------------------------------------------------------------------
@dataclass
class HandlerDef:
"""A declarative fragment handler defined in an .sx file.
Created by ``(defhandler name (&key param...) body)``.
The body is evaluated in a sandboxed environment with only
s-expression primitives available.
"""
name: str
params: list[str] # keyword parameter names
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<handler:{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# RelationDef
# ---------------------------------------------------------------------------
@dataclass(frozen=True)
class RelationDef:
"""A declared relation between two entity types.
Created by ``(defrelation :name ...)`` s-expressions.
"""
name: str # "page->market"
from_type: str # "page"
to_type: str # "market"
cardinality: str # "one-to-one" | "one-to-many" | "many-to-many"
inverse: str | None # "market->page"
nav: str # "submenu" | "tab" | "badge" | "inline" | "hidden"
nav_icon: str | None # "fa fa-shopping-bag"
nav_label: str | None # "markets"
# ---------------------------------------------------------------------------
# PageDef
# ---------------------------------------------------------------------------
@dataclass
class PageDef:
"""A declarative GET page defined in an .sx file.
Created by ``(defpage name :path "/..." :auth :public :content expr)``.
Slots are stored as unevaluated AST and resolved at request time.
"""
name: str
path: str
auth: str | list # "public", "login", "admin", or ["rights", ...]
layout: Any # layout name/config (unevaluated)
cache: dict | None
data_expr: Any # unevaluated AST
content_expr: Any # unevaluated AST
filter_expr: Any
aside_expr: Any
menu_expr: Any
stream: bool = False # enable streaming response
fallback_expr: Any = None # fallback content while streaming
shell_expr: Any = None # immediate shell content (wraps suspense)
closure: dict[str, Any] = field(default_factory=dict)
render_plan: dict[str, Any] | None = field(default=None, repr=False)
_FIELD_MAP = {
"name": "name", "path": "path", "auth": "auth",
"layout": "layout", "cache": "cache",
"data": "data_expr", "content": "content_expr",
"filter": "filter_expr", "aside": "aside_expr",
"menu": "menu_expr", "stream": "stream",
"fallback": "fallback_expr", "shell": "shell_expr",
}
def get(self, key, default=None):
attr = self._FIELD_MAP.get(key)
if attr is not None:
return getattr(self, attr)
return default
def __repr__(self):
return f"<page:{self.name} path={self.path!r}>"
# ---------------------------------------------------------------------------
# QueryDef / ActionDef
# ---------------------------------------------------------------------------
@dataclass
class QueryDef:
"""A declarative data query defined in an .sx file.
Created by ``(defquery name (&key param...) "docstring" body)``.
The body is evaluated with async I/O primitives to produce JSON data.
"""
name: str
params: list[str] # keyword parameter names
doc: str # docstring
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<query:{self.name}({', '.join(self.params)})>"
@dataclass
class ActionDef:
"""A declarative action defined in an .sx file.
Created by ``(defaction name (&key param...) "docstring" body)``.
The body is evaluated with async I/O primitives to produce JSON data.
"""
name: str
params: list[str] # keyword parameter names
doc: str # docstring
body: Any # unevaluated s-expression body
closure: dict[str, Any] = field(default_factory=dict)
def __repr__(self):
return f"<action:{self.name}({', '.join(self.params)})>"
# ---------------------------------------------------------------------------
# Continuation
# ---------------------------------------------------------------------------
class Continuation:
"""A captured delimited continuation (shift/reset).
Callable with one argument — provides the value that the shift
expression "returns" within the delimited context.
"""
__slots__ = ("fn",)
def __init__(self, fn):
self.fn = fn
def __call__(self, value=NIL):
return self.fn(value)
def __repr__(self):
return "<continuation>"
class _ShiftSignal(BaseException):
"""Raised by shift to unwind to the nearest reset.
Inherits from BaseException (not Exception) to avoid being caught
by generic except clauses in user code.
"""
__slots__ = ("k_name", "body", "env")
def __init__(self, k_name, body, env):
self.k_name = k_name
self.body = body
self.env = env
# ---------------------------------------------------------------------------
# Type alias
# ---------------------------------------------------------------------------
# An s-expression value after evaluation
SExp = int | float | str | bool | Symbol | Keyword | Lambda | Macro | Component | Island | Continuation | HandlerDef | RelationDef | PageDef | QueryDef | ActionDef | list | dict | _Nil | None