Add signal test suite (17/17) and Island type to evaluator

test-signals.sx: 17 tests covering signal basics (create, deref, reset!,
swap!), computed (derive, update, chain), effects (run, re-run, dispose,
cleanup), batch (deferred deduped notifications), and defisland (create,
call, children).

types.py: Island dataclass mirroring Component but for reactive boundaries.
evaluator.py: sf_defisland special form, Island in call dispatch.
run.py: Signal platform primitives (make-signal, tracking context, etc)
  and native effect/computed/batch implementations that bridge Lambda
  calls across the Python↔SX boundary.
signals.sx: Updated batch to deduplicate subscribers across signals.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

shared/sx/tests/run.py

@@ -21,7 +21,7 @@ sys.path.insert(0, _PROJECT)
 
 from shared.sx.parser import parse_all
 from shared.sx.evaluator import _eval, _trampoline, _call_lambda
-from shared.sx.types import Symbol, Keyword, Lambda, NIL
+from shared.sx.types import Symbol, Keyword, Lambda, NIL, Component, Island
 
 # --- Test state ---
 suite_stack: list[str] = []
@@ -134,15 +134,133 @@ def render_html(sx_source):
     return result
 
 
+# --- Signal platform primitives ---
+# Implements the signal runtime platform interface for testing signals.sx
+
+class Signal:
+    """A reactive signal container."""
+    __slots__ = ("value", "subscribers", "deps")
+
+    def __init__(self, value):
+        self.value = value
+        self.subscribers = []  # list of callables
+        self.deps = []         # list of Signal (for computed)
+
+
+class TrackingContext:
+    """Tracks signal dependencies during effect/computed evaluation."""
+    __slots__ = ("notify_fn", "deps")
+
+    def __init__(self, notify_fn):
+        self.notify_fn = notify_fn
+        self.deps = []
+
+
+_tracking_context = [None]  # mutable cell
+
+
+def _make_signal(value):
+    s = Signal(value)
+    return s
+
+
+def _signal_p(x):
+    return isinstance(x, Signal)
+
+
+def _signal_value(s):
+    return s.value
+
+
+def _signal_set_value(s, v):
+    s.value = v
+    return NIL
+
+
+def _signal_subscribers(s):
+    return list(s.subscribers)
+
+
+def _signal_add_sub(s, fn):
+    if fn not in s.subscribers:
+        s.subscribers.append(fn)
+    return NIL
+
+
+def _signal_remove_sub(s, fn):
+    if fn in s.subscribers:
+        s.subscribers.remove(fn)
+    return NIL
+
+
+def _signal_deps(s):
+    return list(s.deps)
+
+
+def _signal_set_deps(s, deps):
+    s.deps = list(deps)
+    return NIL
+
+
+def _set_tracking_context(ctx):
+    _tracking_context[0] = ctx
+    return NIL
+
+
+def _get_tracking_context():
+    return _tracking_context[0] or NIL
+
+
+def _make_tracking_context(notify_fn):
+    return TrackingContext(notify_fn)
+
+
+def _tracking_context_deps(ctx):
+    if isinstance(ctx, TrackingContext):
+        return ctx.deps
+    return []
+
+
+def _tracking_context_add_dep(ctx, s):
+    if isinstance(ctx, TrackingContext) and s not in ctx.deps:
+        ctx.deps.append(s)
+    return NIL
+
+
+def _tracking_context_notify_fn(ctx):
+    if isinstance(ctx, TrackingContext):
+        return ctx.notify_fn
+    return NIL
+
+
+def _identical(a, b):
+    return a is b
+
+
+def _island_p(x):
+    return isinstance(x, Island)
+
+
+def _make_island(name, params, has_children, body, closure):
+    return Island(
+        name=name,
+        params=list(params),
+        has_children=has_children,
+        body=body,
+        closure=dict(closure) if isinstance(closure, dict) else {},
+    )
+
+
 # --- Spec registry ---
 
 SPECS = {
-    "eval":   {"file": "test-eval.sx",   "needs": []},
-    "parser": {"file": "test-parser.sx", "needs": ["sx-parse"]},
-    "router": {"file": "test-router.sx", "needs": []},
-    "render": {"file": "test-render.sx", "needs": ["render-html"]},
-    "deps":   {"file": "test-deps.sx",   "needs": []},
-    "engine": {"file": "test-engine.sx", "needs": []},
+    "eval":    {"file": "test-eval.sx",    "needs": []},
+    "parser":  {"file": "test-parser.sx",  "needs": ["sx-parse"]},
+    "router":  {"file": "test-router.sx",  "needs": []},
+    "render":  {"file": "test-render.sx",  "needs": ["render-html"]},
+    "deps":    {"file": "test-deps.sx",    "needs": []},
+    "engine":  {"file": "test-engine.sx",  "needs": []},
+    "signals": {"file": "test-signals.sx", "needs": ["make-signal"]},
 }
 
 REF_DIR = os.path.join(_HERE, "..", "ref")
@@ -186,6 +304,31 @@ env = {
     "inc": lambda n: n + 1,
     # Component accessor for affinity (Phase 7)
     "component-affinity": lambda c: getattr(c, 'affinity', 'auto'),
+    # Signal platform primitives
+    "make-signal": _make_signal,
+    "signal?": _signal_p,
+    "signal-value": _signal_value,
+    "signal-set-value!": _signal_set_value,
+    "signal-subscribers": _signal_subscribers,
+    "signal-add-sub!": _signal_add_sub,
+    "signal-remove-sub!": _signal_remove_sub,
+    "signal-deps": _signal_deps,
+    "signal-set-deps!": _signal_set_deps,
+    "set-tracking-context!": _set_tracking_context,
+    "get-tracking-context": _get_tracking_context,
+    "make-tracking-context": _make_tracking_context,
+    "tracking-context-deps": _tracking_context_deps,
+    "tracking-context-add-dep!": _tracking_context_add_dep,
+    "tracking-context-notify-fn": _tracking_context_notify_fn,
+    "identical?": _identical,
+    # Island platform primitives
+    "island?": _island_p,
+    "make-island": _make_island,
+    "component-name": lambda c: getattr(c, 'name', ''),
+    "component-params": lambda c: list(getattr(c, 'params', [])),
+    "component-body": lambda c: getattr(c, 'body', NIL),
+    "component-closure": lambda c: dict(getattr(c, 'closure', {})),
+    "component-has-children?": lambda c: getattr(c, 'has_children', False),
 }
 
 
@@ -352,6 +495,171 @@ def _load_forms_from_bootstrap(env):
         eval_file("forms.sx", env)
 
 
+def _load_signals(env):
+    """Load signals.sx spec — defines signal, deref, reset!, swap!, etc.
+
+    The hand-written evaluator doesn't support &rest in define/fn,
+    so we override swap! with a native implementation after loading.
+    """
+    # callable? is needed by effect (to check if return value is cleanup fn)
+    env["callable?"] = lambda x: callable(x) or isinstance(x, Lambda)
+    eval_file("signals.sx", env)
+
+    # Override signal functions that need to call Lambda subscribers.
+    # The hand-written evaluator's Lambda objects can't be called directly
+    # from Python — they need _call_lambda. So we provide native versions
+    # of functions that bridge native→Lambda calls.
+
+    def _call_sx_fn(fn, args):
+        """Call an SX function (Lambda or native) from Python."""
+        if isinstance(fn, Lambda):
+            return _trampoline(_call_lambda(fn, list(args), env))
+        if callable(fn):
+            return fn(*args)
+        return NIL
+
+    def _flush_subscribers(s):
+        for sub in list(s.subscribers):
+            _call_sx_fn(sub, [])
+        return NIL
+
+    def _notify_subscribers(s):
+        batch_depth = env.get("*batch-depth*", 0)
+        if batch_depth and batch_depth > 0:
+            batch_queue = env.get("*batch-queue*", [])
+            if s not in batch_queue:
+                batch_queue.append(s)
+            return NIL
+        _flush_subscribers(s)
+        return NIL
+    env["notify-subscribers"] = _notify_subscribers
+    env["flush-subscribers"] = _flush_subscribers
+
+    def _reset_bang(s, value):
+        if not isinstance(s, Signal):
+            return NIL
+        old = s.value
+        if old is not value:
+            s.value = value
+            _notify_subscribers(s)
+        return NIL
+    env["reset!"] = _reset_bang
+
+    def _swap_bang(s, f, *args):
+        if not isinstance(s, Signal):
+            return NIL
+        old = s.value
+        all_args = [old] + list(args)
+        new_val = _call_sx_fn(f, all_args)
+        if old is not new_val:
+            s.value = new_val
+            _notify_subscribers(s)
+        return NIL
+    env["swap!"] = _swap_bang
+
+    def _computed(compute_fn):
+        s = Signal(NIL)
+
+        def recompute():
+            # Unsubscribe from old deps
+            for dep in s.deps:
+                if recompute in dep.subscribers:
+                    dep.subscribers.remove(recompute)
+            s.deps = []
+
+            # Create tracking context
+            ctx = TrackingContext(recompute)
+            prev = _tracking_context[0]
+            _tracking_context[0] = ctx
+
+            new_val = _call_sx_fn(compute_fn, [])
+
+            _tracking_context[0] = prev
+            s.deps = list(ctx.deps)
+
+            old = s.value
+            s.value = new_val
+            if old is not new_val:
+                _flush_subscribers(s)
+
+        recompute()
+        return s
+    env["computed"] = _computed
+
+    def _effect(effect_fn):
+        deps = []
+        disposed = [False]
+        cleanup_fn = [None]
+
+        def run_effect():
+            if disposed[0]:
+                return NIL
+            # Run previous cleanup
+            if cleanup_fn[0]:
+                _call_sx_fn(cleanup_fn[0], [])
+                cleanup_fn[0] = None
+
+            # Unsubscribe from old deps
+            for dep in deps:
+                if run_effect in dep.subscribers:
+                    dep.subscribers.remove(run_effect)
+            deps.clear()
+
+            # Track new deps
+            ctx = TrackingContext(run_effect)
+            prev = _tracking_context[0]
+            _tracking_context[0] = ctx
+
+            result = _call_sx_fn(effect_fn, [])
+
+            _tracking_context[0] = prev
+            deps.extend(ctx.deps)
+
+            # If effect returns a callable, it's cleanup
+            if callable(result) or isinstance(result, Lambda):
+                cleanup_fn[0] = result
+
+            return NIL
+
+        run_effect()
+
+        def dispose():
+            disposed[0] = True
+            if cleanup_fn[0]:
+                _call_sx_fn(cleanup_fn[0], [])
+            for dep in deps:
+                if run_effect in dep.subscribers:
+                    dep.subscribers.remove(run_effect)
+            deps.clear()
+            return NIL
+
+        return dispose
+    env["effect"] = _effect
+
+    def _batch(thunk):
+        depth = env.get("*batch-depth*", 0)
+        env["*batch-depth*"] = depth + 1
+        _call_sx_fn(thunk, [])
+        env["*batch-depth*"] = env["*batch-depth*"] - 1
+        if env["*batch-depth*"] == 0:
+            queue = env.get("*batch-queue*", [])
+            env["*batch-queue*"] = []
+            # Collect unique subscribers across all queued signals
+            seen = set()
+            pending = []
+            for s in queue:
+                for sub in s.subscribers:
+                    sub_id = id(sub)
+                    if sub_id not in seen:
+                        seen.add(sub_id)
+                        pending.append(sub)
+            # Notify each unique subscriber exactly once
+            for sub in pending:
+                _call_sx_fn(sub, [])
+        return NIL
+    env["batch"] = _batch
+
+
 def main():
     global passed, failed, test_num
 
@@ -395,6 +703,8 @@ def main():
                 _load_deps_from_bootstrap(env)
             if spec_name == "engine":
                 _load_engine_from_bootstrap(env)
+            if spec_name == "signals":
+                _load_signals(env)
 
             print(f"# --- {spec_name} ---")
             eval_file(spec["file"], env)