"""
Async s-expression evaluator and HTML renderer.

Mirrors the sync evaluator (evaluator.py) and HTML renderer (html.py) but
every step is ``async`` so I/O primitives can be ``await``ed inline.

This is the execution engine for ``defhandler`` — handlers contain I/O
calls (``query``, ``service``, ``request-arg``, etc.) interleaved with
control flow (``if``, ``let``, ``map``, ``when``).  The sync
collect-then-substitute resolver can't handle data dependencies between
I/O results and control flow, so handlers need inline async evaluation.

Evaluation modes
~~~~~~~~~~~~~~~~

The same component AST can be evaluated in different modes depending on
where the rendering boundary is drawn (server vs client).  Five modes
exist across the codebase:

  Function              Expands components?  Output          Used for
  --------------------  -------------------  --------------  ----------------------------
  _eval (sync)          Yes                  Python values   register_components, Jinja sx()
  _arender (async)      Yes                  HTML            render_to_html
  _aser (async)         No — serializes      SX wire format  render_to_sx
  _aser_component       Yes, one level       SX wire format  render_to_sx_with_env (layouts)
  sx.js renderDOM       Yes                  DOM nodes       Client-side

_aser deliberately does NOT expand ~component calls — it serializes them
as SX wire format so the client can render them.  But layout components
(used by render_to_sx_with_env) need server-side expansion because they
depend on Python context (auth state, fragments, etc.).  That's what
_aser_component / async_eval_slot_to_sx provides: expand the top-level
component body server-side, then serialize its children for the client.

Usage::

    from shared.sx.async_eval import async_render

    html = await async_render(handler_def.body, env, ctx)
"""

from __future__ import annotations

import contextvars
import inspect
from typing import Any

from .types import Component, Island, Keyword, Lambda, Macro, NIL, Symbol

# When True, _aser expands known components server-side instead of serializing
# them for client rendering. Set during page slot evaluation so Python-only
# helpers (e.g. highlight) in component bodies execute on the server.
_expand_components: contextvars.ContextVar[bool] = contextvars.ContextVar(
    "_expand_components", default=False
)
# sx_ref.py removed — stub so module loads. OCaml bridge handles macro expansion.
def _expand_macro(*a, **kw):
    raise RuntimeError("sx_ref.py has been removed — use SX_USE_OCAML=1")
from .types import EvalError
from .primitives import _PRIMITIVES
from .primitives_io import IO_PRIMITIVES, RequestContext, execute_io
from .parser import SxExpr, serialize
from .html import (
    HTML_TAGS, VOID_ELEMENTS, BOOLEAN_ATTRS,
    escape_text, escape_attr, _RawHTML, css_class_collector, _svg_context,
)


# ---------------------------------------------------------------------------
# Async TCO — thunk + trampoline
# ---------------------------------------------------------------------------

class _AsyncThunk:
    """Deferred (expr, env, ctx) for tail-call optimization."""
    __slots__ = ("expr", "env", "ctx")
    def __init__(self, expr: Any, env: dict[str, Any], ctx: RequestContext) -> None:
        self.expr = expr
        self.env = env
        self.ctx = ctx


async def _async_trampoline(val: Any) -> Any:
    """Iteratively resolve thunks from tail positions."""
    while isinstance(val, _AsyncThunk):
        val = await _async_eval(val.expr, val.env, val.ctx)
    return val


# ---------------------------------------------------------------------------
# Async evaluate
# ---------------------------------------------------------------------------

async def async_eval(expr: Any, env: dict[str, Any], ctx: RequestContext) -> Any:
    """Public entry — evaluates and trampolines thunks."""
    result = await _async_eval(expr, env, ctx)
    while isinstance(result, _AsyncThunk):
        result = await _async_eval(result.expr, result.env, result.ctx)
    return result


async def _async_eval(expr: Any, env: dict[str, Any], ctx: RequestContext) -> Any:
    """Internal evaluator — may return _AsyncThunk for tail positions."""
    # --- literals ---
    if isinstance(expr, (int, float, str, bool)):
        return expr
    if expr is None or expr is NIL:
        return NIL

    # --- symbol lookup ---
    if isinstance(expr, Symbol):
        name = expr.name
        if name in env:
            return env[name]
        if name in _PRIMITIVES:
            return _PRIMITIVES[name]
        if name == "true":
            return True
        if name == "false":
            return False
        if name == "nil":
            return NIL
        raise EvalError(f"Undefined symbol: {name}")

    # --- keyword ---
    if isinstance(expr, Keyword):
        return expr.name

    # --- dict literal ---
    if isinstance(expr, dict):
        return {k: await _async_trampoline(await _async_eval(v, env, ctx)) for k, v in expr.items()}

    # --- list ---
    if not isinstance(expr, list):
        return expr
    if not expr:
        return []

    head = expr[0]

    if not isinstance(head, (Symbol, Lambda, list)):
        return [await _async_trampoline(await _async_eval(x, env, ctx)) for x in expr]

    if isinstance(head, Symbol):
        name = head.name

        # I/O primitives — await inline
        if name in IO_PRIMITIVES:
            args, kwargs = await _parse_io_args(expr[1:], env, ctx)
            return await execute_io(name, args, kwargs, ctx)

        # Special forms
        sf = _ASYNC_SPECIAL_FORMS.get(name)
        if sf is not None:
            return await sf(expr, env, ctx)

        ho = _ASYNC_HO_FORMS.get(name)
        if ho is not None:
            return await ho(expr, env, ctx)

        # Macro expansion — tail position
        if name in env:
            val = env[name]
            if isinstance(val, Macro):
                expanded = _expand_macro(val, expr[1:], env)
                return _AsyncThunk(expanded, env, ctx)

        # Render forms in eval position — delegate to renderer and return
        # as _RawHTML so it won't be double-escaped when used in render
        # context later.  Allows (let ((x (<> ...))) ...) etc.
        if name in ("<>", "raw!") or name in HTML_TAGS:
            html = await _arender(expr, env, ctx)
            return _RawHTML(html)

    # --- function / lambda call ---
    fn = await _async_trampoline(await _async_eval(head, env, ctx))
    args = [await _async_trampoline(await _async_eval(a, env, ctx)) for a in expr[1:]]

    if callable(fn) and not isinstance(fn, (Lambda, Component)):
        result = fn(*args)
        if inspect.iscoroutine(result):
            return await result
        return result
    if isinstance(fn, Lambda):
        return await _async_call_lambda(fn, args, env, ctx)
    if isinstance(fn, Component):
        return await _async_call_component(fn, expr[1:], env, ctx)
    raise EvalError(f"Not callable: {fn!r}")


async def _parse_io_args(
    exprs: list[Any], env: dict[str, Any], ctx: RequestContext,
) -> tuple[list[Any], dict[str, Any]]:
    """Parse and evaluate I/O node args."""
    args: list[Any] = []
    kwargs: dict[str, Any] = {}
    i = 0
    while i < len(exprs):
        item = exprs[i]
        if isinstance(item, Keyword) and i + 1 < len(exprs):
            kwargs[item.name] = await async_eval(exprs[i + 1], env, ctx)
            i += 2
        else:
            args.append(await async_eval(item, env, ctx))
            i += 1
    return args, kwargs


async def _async_call_lambda(
    fn: Lambda, args: list[Any], caller_env: dict[str, Any], ctx: RequestContext,
) -> Any:
    # Too many args is an error; too few pads with nil
    if len(args) > len(fn.params):
        raise EvalError(f"{fn!r} expects {len(fn.params)} args, got {len(args)}")
    local = dict(fn.closure)
    local.update(caller_env)
    for p, v in zip(fn.params, args):
        local[p] = v
    # Pad missing params with nil
    for p in fn.params[len(args):]:
        local[p] = None
    return _AsyncThunk(fn.body, local, ctx)


async def _async_call_component(
    comp: Component, raw_args: list[Any], env: dict[str, Any], ctx: RequestContext,
) -> Any:
    kwargs: dict[str, Any] = {}
    children: list[Any] = []
    i = 0
    while i < len(raw_args):
        arg = raw_args[i]
        if isinstance(arg, Keyword) and i + 1 < len(raw_args):
            kwargs[arg.name] = await async_eval(raw_args[i + 1], env, ctx)
            i += 2
        else:
            children.append(await async_eval(arg, env, ctx))
            i += 1
    local = dict(comp.closure)
    local.update(env)
    for p in comp.params:
        local[p] = kwargs.get(p, NIL)
    if comp.has_children:
        local["children"] = children
    return _AsyncThunk(comp.body, local, ctx)


# ---------------------------------------------------------------------------
# Async special forms
# ---------------------------------------------------------------------------

async def _asf_if(expr, env, ctx):
    cond = await _async_trampoline(await _async_eval(expr[1], env, ctx))
    if cond and cond is not NIL:
        return _AsyncThunk(expr[2], env, ctx)
    if len(expr) > 3:
        return _AsyncThunk(expr[3], env, ctx)
    return NIL


async def _asf_when(expr, env, ctx):
    cond = await _async_trampoline(await _async_eval(expr[1], env, ctx))
    if cond and cond is not NIL:
        for body_expr in expr[2:-1]:
            await _async_trampoline(await _async_eval(body_expr, env, ctx))
        if len(expr) > 2:
            return _AsyncThunk(expr[-1], env, ctx)
    return NIL


async def _asf_and(expr, env, ctx):
    result: Any = True
    for arg in expr[1:]:
        result = await _async_trampoline(await _async_eval(arg, env, ctx))
        if not result:
            return result
    return result


async def _asf_or(expr, env, ctx):
    result: Any = False
    for arg in expr[1:]:
        result = await _async_trampoline(await _async_eval(arg, env, ctx))
        if result:
            return result
    return result


async def _asf_let(expr, env, ctx):
    # Named let: (let name ((x 0) ...) body)
    if isinstance(expr[1], Symbol):
        return await _asf_named_let(expr, env, ctx)

    bindings = expr[1]
    local = dict(env)
    if isinstance(bindings, list):
        if bindings and isinstance(bindings[0], list):
            for binding in bindings:
                var = binding[0]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await _async_trampoline(await _async_eval(binding[1], local, ctx))
        elif len(bindings) % 2 == 0:
            for i in range(0, len(bindings), 2):
                var = bindings[i]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await _async_trampoline(await _async_eval(bindings[i + 1], local, ctx))
    for body_expr in expr[2:-1]:
        await _async_trampoline(await _async_eval(body_expr, local, ctx))
    if len(expr) > 2:
        return _AsyncThunk(expr[-1], local, ctx)
    return NIL


async def _asf_named_let(expr, env, ctx):
    """Async named let: (let name ((x 0) ...) body)"""
    loop_name = expr[1].name
    bindings = expr[2]
    body = expr[3:]

    params: list[str] = []
    inits: list = []

    if isinstance(bindings, list):
        if bindings and isinstance(bindings[0], list):
            for binding in bindings:
                var = binding[0]
                params.append(var.name if isinstance(var, Symbol) else var)
                inits.append(binding[1])
        elif len(bindings) % 2 == 0:
            for i in range(0, len(bindings), 2):
                var = bindings[i]
                params.append(var.name if isinstance(var, Symbol) else var)
                inits.append(bindings[i + 1])

    loop_body = body[0] if len(body) == 1 else [Symbol("begin")] + list(body)
    loop_fn = Lambda(params, loop_body, dict(env), name=loop_name)
    loop_fn.closure[loop_name] = loop_fn

    init_vals = [await _async_trampoline(await _async_eval(init, env, ctx)) for init in inits]
    return await _async_call_lambda(loop_fn, init_vals, env, ctx)


async def _asf_letrec(expr, env, ctx):
    """Async letrec: (letrec ((name1 val1) ...) body)"""
    bindings = expr[1]
    local = dict(env)

    names: list[str] = []
    val_exprs: list = []

    if isinstance(bindings, list):
        if bindings and isinstance(bindings[0], list):
            for binding in bindings:
                var = binding[0]
                vname = var.name if isinstance(var, Symbol) else var
                names.append(vname)
                val_exprs.append(binding[1])
                local[vname] = NIL
        elif len(bindings) % 2 == 0:
            for i in range(0, len(bindings), 2):
                var = bindings[i]
                vname = var.name if isinstance(var, Symbol) else var
                names.append(vname)
                val_exprs.append(bindings[i + 1])
                local[vname] = NIL

    values = [await _async_trampoline(await _async_eval(ve, local, ctx)) for ve in val_exprs]
    for name, val in zip(names, values):
        local[name] = val
    for val in values:
        if isinstance(val, Lambda):
            for name in names:
                val.closure[name] = local[name]

    for body_expr in expr[2:-1]:
        await _async_trampoline(await _async_eval(body_expr, local, ctx))
    if len(expr) > 2:
        return _AsyncThunk(expr[-1], local, ctx)
    return NIL


async def _asf_dynamic_wind(expr, env, ctx):
    """Async dynamic-wind: (dynamic-wind before body after)"""
    before = await _async_trampoline(await _async_eval(expr[1], env, ctx))
    body_fn = await _async_trampoline(await _async_eval(expr[2], env, ctx))
    after = await _async_trampoline(await _async_eval(expr[3], env, ctx))

    async def _call_thunk(fn):
        if isinstance(fn, Lambda):
            return await _async_trampoline(await _async_call_lambda(fn, [], env, ctx))
        if callable(fn):
            r = fn()
            if inspect.iscoroutine(r):
                return await r
            return r
        raise EvalError(f"dynamic-wind: expected thunk, got {type(fn).__name__}")

    await _call_thunk(before)
    try:
        result = await _call_thunk(body_fn)
    finally:
        await _call_thunk(after)
    return result


async def _asf_lambda(expr, env, ctx):
    params_expr = expr[1]
    param_names = []
    for p in params_expr:
        if isinstance(p, Symbol):
            param_names.append(p.name)
        elif isinstance(p, str):
            param_names.append(p)
    return Lambda(param_names, expr[2], dict(env))


async def _asf_define(expr, env, ctx):
    name_sym = expr[1]
    value = await _async_trampoline(await _async_eval(expr[2], env, ctx))
    if isinstance(value, Lambda) and value.name is None:
        value.name = name_sym.name
    env[name_sym.name] = value
    return value


async def _asf_defcomp(expr, env, ctx):
    # Component definitions are handled by OCaml kernel at load time.
    # Python-side: just store a minimal Component in env for reference.
    from .types import Component
    name_sym = expr[1]
    name = name_sym.name if hasattr(name_sym, 'name') else str(name_sym)
    env[name] = Component(name=name.lstrip("~"), params=[], has_children=False,
                          body=expr[-1], closure={})
    return NIL


async def _asf_defstyle(expr, env, ctx):
    # Style definitions handled by OCaml kernel.
    return NIL


async def _asf_defmacro(expr, env, ctx):
    # Macro definitions handled by OCaml kernel.
    from .types import Macro
    name_sym = expr[1]
    name = name_sym.name if hasattr(name_sym, 'name') else str(name_sym)
    params_form = expr[2] if len(expr) > 3 else []
    param_names = [p.name for p in params_form if isinstance(p, Symbol) and not p.name.startswith("&")]
    rest_param = None
    for i, p in enumerate(params_form):
        if isinstance(p, Symbol) and p.name == "&rest" and i + 1 < len(params_form):
            rest_param = params_form[i + 1].name if isinstance(params_form[i + 1], Symbol) else None
    env[name] = Macro(name=name, params=param_names, rest_param=rest_param, body=expr[-1])
    return NIL


async def _asf_defhandler(expr, env, ctx):
    # Handler definitions handled by OCaml kernel.
    return NIL


async def _asf_begin(expr, env, ctx):
    for sub in expr[1:-1]:
        await _async_trampoline(await _async_eval(sub, env, ctx))
    if len(expr) > 1:
        return _AsyncThunk(expr[-1], env, ctx)
    return NIL


async def _asf_quote(expr, env, ctx):
    return expr[1] if len(expr) > 1 else NIL


async def _asf_quasiquote(expr, env, ctx):
    return await _async_qq_expand(expr[1], env, ctx)


async def _async_qq_expand(template, env, ctx):
    if not isinstance(template, list):
        return template
    if not template:
        return []
    head = template[0]
    if isinstance(head, Symbol):
        if head.name == "unquote":
            return await async_eval(template[1], env, ctx)
        if head.name == "splice-unquote":
            raise EvalError("splice-unquote not inside a list")
    result: list[Any] = []
    for item in template:
        if (isinstance(item, list) and len(item) == 2
                and isinstance(item[0], Symbol) and item[0].name == "splice-unquote"):
            spliced = await async_eval(item[1], env, ctx)
            if isinstance(spliced, list):
                result.extend(spliced)
            elif spliced is not None and spliced is not NIL:
                result.append(spliced)
        else:
            result.append(await _async_qq_expand(item, env, ctx))
    return result


async def _asf_cond(expr, env, ctx):
    clauses = expr[1:]
    if not clauses:
        return NIL
    if (isinstance(clauses[0], list) and len(clauses[0]) == 2
            and not (isinstance(clauses[0][0], Symbol) and clauses[0][0].name in (
                "=", "<", ">", "<=", ">=", "!=", "and", "or"))):
        for clause in clauses:
            test = clause[0]
            if isinstance(test, Symbol) and test.name in ("else", ":else"):
                return _AsyncThunk(clause[1], env, ctx)
            if isinstance(test, Keyword) and test.name == "else":
                return _AsyncThunk(clause[1], env, ctx)
            if await _async_trampoline(await _async_eval(test, env, ctx)):
                return _AsyncThunk(clause[1], env, ctx)
    else:
        i = 0
        while i < len(clauses) - 1:
            test = clauses[i]
            result = clauses[i + 1]
            if isinstance(test, Keyword) and test.name == "else":
                return _AsyncThunk(result, env, ctx)
            if isinstance(test, Symbol) and test.name in (":else", "else"):
                return _AsyncThunk(result, env, ctx)
            if await _async_trampoline(await _async_eval(test, env, ctx)):
                return _AsyncThunk(result, env, ctx)
            i += 2
    return NIL


async def _asf_case(expr, env, ctx):
    match_val = await _async_trampoline(await _async_eval(expr[1], env, ctx))
    clauses = expr[2:]
    i = 0
    while i < len(clauses) - 1:
        test = clauses[i]
        result = clauses[i + 1]
        if isinstance(test, Keyword) and test.name == "else":
            return _AsyncThunk(result, env, ctx)
        if isinstance(test, Symbol) and test.name in (":else", "else"):
            return _AsyncThunk(result, env, ctx)
        if match_val == await _async_trampoline(await _async_eval(test, env, ctx)):
            return _AsyncThunk(result, env, ctx)
        i += 2
    return NIL


async def _asf_thread_first(expr, env, ctx):
    result = await _async_trampoline(await _async_eval(expr[1], env, ctx))
    for form in expr[2:]:
        if isinstance(form, list):
            fn = await _async_trampoline(await _async_eval(form[0], env, ctx))
            args = [result] + [await _async_trampoline(await _async_eval(a, env, ctx)) for a in form[1:]]
        else:
            fn = await _async_trampoline(await _async_eval(form, env, ctx))
            args = [result]
        if callable(fn) and not isinstance(fn, (Lambda, Component)):
            result = fn(*args)
            if inspect.iscoroutine(result):
                result = await result
        elif isinstance(fn, Lambda):
            result = await _async_trampoline(await _async_call_lambda(fn, args, env, ctx))
        else:
            raise EvalError(f"-> form not callable: {fn!r}")
    return result


async def _asf_set_bang(expr, env, ctx):
    value = await _async_trampoline(await _async_eval(expr[2], env, ctx))
    env[expr[1].name] = value
    return value


_ASYNC_SPECIAL_FORMS: dict[str, Any] = {
    "if": _asf_if,
    "when": _asf_when,
    "cond": _asf_cond,
    "case": _asf_case,
    "and": _asf_and,
    "or": _asf_or,
    "let": _asf_let,
    "let*": _asf_let,
    "letrec": _asf_letrec,
    "lambda": _asf_lambda,
    "fn": _asf_lambda,
    "define": _asf_define,
    "defstyle": _asf_defstyle,
    "defcomp": _asf_defcomp,
    "defmacro": _asf_defmacro,
    "defhandler": _asf_defhandler,
    "begin": _asf_begin,
    "do": _asf_begin,
    "quote": _asf_quote,
    "quasiquote": _asf_quasiquote,
    "->": _asf_thread_first,
    "set!": _asf_set_bang,
    "dynamic-wind": _asf_dynamic_wind,
}


# ---------------------------------------------------------------------------
# Async delimited continuations — shift / reset
# ---------------------------------------------------------------------------

_ASYNC_RESET_RESUME: list = []


async def _asf_reset(expr, env, ctx):
    """(reset body) — async version."""
    from .types import Continuation, _ShiftSignal
    body = expr[1]
    try:
        return await async_eval(body, env, ctx)
    except _ShiftSignal as sig:
        def cont_fn(value=None):
            from .types import NIL
            _ASYNC_RESET_RESUME.append(value if value is not None else NIL)
            try:
                # Continuations are handled by OCaml kernel.
                # Python-side cont_fn should not be called in normal operation.
                raise RuntimeError(
                    "Python-side continuation invocation not supported — "
                    "use OCaml bridge for shift/reset"
                )
            finally:
                _ASYNC_RESET_RESUME.pop()
        k = Continuation(cont_fn)
        sig_env = dict(sig.env)
        sig_env[sig.k_name] = k
        return await async_eval(sig.body, sig_env, ctx)


async def _asf_shift(expr, env, ctx):
    """(shift k body) — async version."""
    from .types import _ShiftSignal
    if _ASYNC_RESET_RESUME:
        return _ASYNC_RESET_RESUME[-1]
    k_name = expr[1].name
    body = expr[2]
    raise _ShiftSignal(k_name, body, env)

_ASYNC_SPECIAL_FORMS["reset"] = _asf_reset
_ASYNC_SPECIAL_FORMS["shift"] = _asf_shift


# ---------------------------------------------------------------------------
# Async higher-order forms
# ---------------------------------------------------------------------------

async def _aho_map(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for item in coll:
        if isinstance(fn, Lambda):
            results.append(await _async_trampoline(await _async_call_lambda(fn, [item], env, ctx)))
        elif callable(fn):
            r = fn(item)
            results.append(await r if inspect.iscoroutine(r) else r)
        else:
            raise EvalError(f"map requires callable, got {type(fn).__name__}")
    return results


async def _aho_map_indexed(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for i, item in enumerate(coll):
        if isinstance(fn, Lambda):
            results.append(await _async_trampoline(await _async_call_lambda(fn, [i, item], env, ctx)))
        elif callable(fn):
            r = fn(i, item)
            results.append(await r if inspect.iscoroutine(r) else r)
        else:
            raise EvalError(f"map-indexed requires callable, got {type(fn).__name__}")
    return results


async def _aho_filter(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for item in coll:
        if isinstance(fn, Lambda):
            val = await _async_trampoline(await _async_call_lambda(fn, [item], env, ctx))
        elif callable(fn):
            val = fn(item)
            if inspect.iscoroutine(val):
                val = await val
        else:
            raise EvalError(f"filter requires callable, got {type(fn).__name__}")
        if val:
            results.append(item)
    return results


async def _aho_reduce(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    acc = await async_eval(expr[2], env, ctx)
    coll = await async_eval(expr[3], env, ctx)
    for item in coll:
        if isinstance(fn, Lambda):
            acc = await _async_trampoline(await _async_call_lambda(fn, [acc, item], env, ctx))
        else:
            acc = fn(acc, item)
            if inspect.iscoroutine(acc):
                acc = await acc
    return acc


async def _aho_some(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    for item in coll:
        if isinstance(fn, Lambda):
            result = await _async_trampoline(await _async_call_lambda(fn, [item], env, ctx))
        else:
            result = fn(item)
            if inspect.iscoroutine(result):
                result = await result
        if result:
            return result
    return NIL


async def _aho_every(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    for item in coll:
        if isinstance(fn, Lambda):
            val = await _async_trampoline(await _async_call_lambda(fn, [item], env, ctx))
        else:
            val = fn(item)
            if inspect.iscoroutine(val):
                val = await val
        if not val:
            return False
    return True


async def _aho_for_each(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    for item in coll:
        if isinstance(fn, Lambda):
            await _async_trampoline(await _async_call_lambda(fn, [item], env, ctx))
        elif callable(fn):
            r = fn(item)
            if inspect.iscoroutine(r):
                await r
    return NIL


_ASYNC_HO_FORMS: dict[str, Any] = {
    "map": _aho_map,
    "map-indexed": _aho_map_indexed,
    "filter": _aho_filter,
    "reduce": _aho_reduce,
    "some": _aho_some,
    "every?": _aho_every,
    "for-each": _aho_for_each,
}


# ---------------------------------------------------------------------------
# Async HTML renderer
# ---------------------------------------------------------------------------

async def async_render(
    expr: Any,
    env: dict[str, Any],
    ctx: RequestContext | None = None,
) -> str:
    """Render an s-expression to HTML, awaiting I/O primitives inline."""
    if ctx is None:
        ctx = RequestContext()
    return await _arender(expr, env, ctx)


async def _arender(expr: Any, env: dict[str, Any], ctx: RequestContext) -> str:
    if expr is None or expr is NIL or expr is False or expr is True:
        return ""
    if isinstance(expr, _RawHTML):
        return expr.html
    if isinstance(expr, str):
        return escape_text(expr)
    if isinstance(expr, (int, float)):
        return escape_text(str(expr))
    if isinstance(expr, Symbol):
        val = await async_eval(expr, env, ctx)
        return await _arender(val, env, ctx)
    if isinstance(expr, Keyword):
        return escape_text(expr.name)
    if isinstance(expr, list):
        if not expr:
            return ""
        return await _arender_list(expr, env, ctx)
    if isinstance(expr, dict):
        return ""
    return escape_text(str(expr))


async def _arender_list(expr: list, env: dict[str, Any], ctx: RequestContext) -> str:
    head = expr[0]

    if isinstance(head, Symbol):
        name = head.name

        # I/O primitive — await, then render result
        if name in IO_PRIMITIVES:
            result = await async_eval(expr, env, ctx)
            return await _arender(result, env, ctx)

        # raw!
        if name == "raw!":
            parts = []
            for arg in expr[1:]:
                val = await async_eval(arg, env, ctx)
                if isinstance(val, _RawHTML):
                    parts.append(val.html)
                elif isinstance(val, str):
                    parts.append(val)
                elif val is not None and val is not NIL:
                    parts.append(str(val))
            return "".join(parts)

        # <>
        if name == "<>":
            parts = []
            for child in expr[1:]:
                parts.append(await _arender(child, env, ctx))
            return "".join(parts)

        # html: prefix → force tag rendering
        if name.startswith("html:"):
            return await _arender_element(name[5:], expr[1:], env, ctx)

        # Render-aware special forms
        # If name is also an HTML tag and (keyword arg or SVG context) → tag call
        arsf = _ASYNC_RENDER_FORMS.get(name)
        if arsf is not None:
            if name in HTML_TAGS and (
                (len(expr) > 1 and isinstance(expr[1], Keyword))
                or _svg_context.get(False)
            ):
                return await _arender_element(name, expr[1:], env, ctx)
            return await arsf(expr, env, ctx)

        # Macro expansion
        if name in env:
            val = env[name]
            if isinstance(val, Macro):
                expanded = _expand_macro(val, expr[1:], env)
                return await _arender(expanded, env, ctx)

        # HTML tag
        if name in HTML_TAGS:
            return await _arender_element(name, expr[1:], env, ctx)

        # Component
        if name.startswith("~"):
            val = env.get(name)
            if isinstance(val, Component):
                return await _arender_component(val, expr[1:], env, ctx)

        # Custom element (hyphenated name with keyword attrs) → tag
        if "-" in name and len(expr) > 1 and isinstance(expr[1], Keyword):
            return await _arender_element(name, expr[1:], env, ctx)

        # SVG/MathML context → unknown names are child elements
        if _svg_context.get(False):
            return await _arender_element(name, expr[1:], env, ctx)

        # Fallback — evaluate then render
        result = await async_eval(expr, env, ctx)
        return await _arender(result, env, ctx)

    if isinstance(head, (Lambda, list)):
        result = await async_eval(expr, env, ctx)
        return await _arender(result, env, ctx)

    # Data list
    parts = []
    for item in expr:
        parts.append(await _arender(item, env, ctx))
    return "".join(parts)


async def _arender_element(
    tag: str, args: list, env: dict[str, Any], ctx: RequestContext,
) -> str:
    attrs: dict[str, Any] = {}
    children: list[Any] = []
    i = 0
    while i < len(args):
        arg = args[i]
        if isinstance(arg, Keyword) and i + 1 < len(args):
            attr_val = await async_eval(args[i + 1], env, ctx)
            attrs[arg.name] = attr_val
            i += 2
        else:
            children.append(arg)
            i += 1

    class_val = attrs.get("class")
    if class_val is not None and class_val is not NIL and class_val is not False:
        collector = css_class_collector.get(None)
        if collector is not None:
            collector.update(str(class_val).split())

    parts = [f"<{tag}"]
    for attr_name, attr_val in attrs.items():
        if attr_val is None or attr_val is NIL or attr_val is False:
            continue
        if attr_name in BOOLEAN_ATTRS:
            if attr_val:
                parts.append(f" {attr_name}")
        elif attr_val is True:
            parts.append(f" {attr_name}")
        else:
            parts.append(f' {attr_name}="{escape_attr(str(attr_val))}"')
    parts.append(">")
    opening = "".join(parts)

    if tag in VOID_ELEMENTS:
        return opening

    # SVG/MathML namespace auto-detection: set context for children
    token = None
    if tag in ("svg", "math"):
        token = _svg_context.set(True)

    try:
        child_parts = []
        for child in children:
            child_parts.append(await _arender(child, env, ctx))
    finally:
        if token is not None:
            _svg_context.reset(token)

    return f"{opening}{''.join(child_parts)}</{tag}>"


async def _arender_component(
    comp: Component, args: list, env: dict[str, Any], ctx: RequestContext,
) -> str:
    kwargs: dict[str, Any] = {}
    children: list[Any] = []
    i = 0
    while i < len(args):
        arg = args[i]
        if isinstance(arg, Keyword) and i + 1 < len(args):
            kwargs[arg.name] = await async_eval(args[i + 1], env, ctx)
            i += 2
        else:
            children.append(arg)
            i += 1
    local = dict(comp.closure)
    local.update(env)
    for p in comp.params:
        local[p] = kwargs.get(p, NIL)
    if comp.has_children:
        child_html = []
        for c in children:
            child_html.append(await _arender(c, env, ctx))
        local["children"] = _RawHTML("".join(child_html))
    return await _arender(comp.body, local, ctx)


async def _arender_lambda(
    fn: Lambda, args: tuple, env: dict[str, Any], ctx: RequestContext,
) -> str:
    local = dict(fn.closure)
    local.update(env)
    for p, v in zip(fn.params, args):
        local[p] = v
    return await _arender(fn.body, local, ctx)


# ---------------------------------------------------------------------------
# Async render-aware special forms
# ---------------------------------------------------------------------------

async def _arsf_if(expr, env, ctx):
    cond = await async_eval(expr[1], env, ctx)
    if cond and cond is not NIL:
        return await _arender(expr[2], env, ctx)
    if len(expr) > 3:
        return await _arender(expr[3], env, ctx)
    return ""


async def _arsf_when(expr, env, ctx):
    cond = await async_eval(expr[1], env, ctx)
    if cond and cond is not NIL:
        parts = []
        for body_expr in expr[2:]:
            parts.append(await _arender(body_expr, env, ctx))
        return "".join(parts)
    return ""


async def _arsf_cond(expr, env, ctx):
    clauses = expr[1:]
    if not clauses:
        return ""
    if isinstance(clauses[0], list) and len(clauses[0]) == 2:
        for clause in clauses:
            test = clause[0]
            if isinstance(test, Symbol) and test.name in ("else", ":else"):
                return await _arender(clause[1], env, ctx)
            if isinstance(test, Keyword) and test.name == "else":
                return await _arender(clause[1], env, ctx)
            if await async_eval(test, env, ctx):
                return await _arender(clause[1], env, ctx)
    else:
        i = 0
        while i < len(clauses) - 1:
            test = clauses[i]
            result = clauses[i + 1]
            if isinstance(test, Keyword) and test.name == "else":
                return await _arender(result, env, ctx)
            if isinstance(test, Symbol) and test.name in (":else", "else"):
                return await _arender(result, env, ctx)
            if await async_eval(test, env, ctx):
                return await _arender(result, env, ctx)
            i += 2
    return ""


async def _arsf_let(expr, env, ctx):
    bindings = expr[1]
    local = dict(env)
    if isinstance(bindings, list):
        if bindings and isinstance(bindings[0], list):
            for binding in bindings:
                var = binding[0]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await async_eval(binding[1], local, ctx)
        elif len(bindings) % 2 == 0:
            for i in range(0, len(bindings), 2):
                var = bindings[i]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await async_eval(bindings[i + 1], local, ctx)
    parts = []
    for body_expr in expr[2:]:
        parts.append(await _arender(body_expr, local, ctx))
    return "".join(parts)


async def _arsf_begin(expr, env, ctx):
    parts = []
    for sub in expr[1:]:
        parts.append(await _arender(sub, env, ctx))
    return "".join(parts)


async def _arsf_define(expr, env, ctx):
    await async_eval(expr, env, ctx)
    return ""


async def _arsf_map(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    parts = []
    for item in coll:
        if isinstance(fn, Lambda):
            parts.append(await _arender_lambda(fn, (item,), env, ctx))
        elif callable(fn):
            r = fn(item)
            if inspect.iscoroutine(r):
                r = await r
            parts.append(await _arender(r, env, ctx))
        else:
            parts.append(await _arender(item, env, ctx))
    return "".join(parts)


async def _arsf_map_indexed(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    parts = []
    for i, item in enumerate(coll):
        if isinstance(fn, Lambda):
            parts.append(await _arender_lambda(fn, (i, item), env, ctx))
        elif callable(fn):
            r = fn(i, item)
            if inspect.iscoroutine(r):
                r = await r
            parts.append(await _arender(r, env, ctx))
        else:
            parts.append(await _arender(item, env, ctx))
    return "".join(parts)


async def _arsf_filter(expr, env, ctx):
    result = await async_eval(expr, env, ctx)
    return await _arender(result, env, ctx)


async def _arsf_for_each(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    parts = []
    for item in coll:
        if isinstance(fn, Lambda):
            parts.append(await _arender_lambda(fn, (item,), env, ctx))
        elif callable(fn):
            r = fn(item)
            if inspect.iscoroutine(r):
                r = await r
            parts.append(await _arender(r, env, ctx))
        else:
            parts.append(await _arender(item, env, ctx))
    return "".join(parts)


_ASYNC_RENDER_FORMS: dict[str, Any] = {
    "if": _arsf_if,
    "when": _arsf_when,
    "cond": _arsf_cond,
    "let": _arsf_let,
    "let*": _arsf_let,
    "begin": _arsf_begin,
    "do": _arsf_begin,
    "define": _arsf_define,
    "defstyle": _arsf_define,
    "defcomp": _arsf_define,
    "defmacro": _arsf_define,
    "defhandler": _arsf_define,
    "map": _arsf_map,
    "map-indexed": _arsf_map_indexed,
    "filter": _arsf_filter,
    "for-each": _arsf_for_each,
}


# ---------------------------------------------------------------------------
# Async serialize — evaluate I/O/control flow, produce sx source (not HTML)
# ---------------------------------------------------------------------------
# Used by defhandler execution.  Fragment providers need to return sx wire
# format (s-expression source) that consuming apps parse and render client-
# side.  This mirrors the old Python handlers that used sx_call() to build
# sx strings.
#
# _aser ("async serialize") works like _arender but instead of producing
# HTML for components/tags/<>, it serializes them back to sx source with
# their arguments evaluated.


async def async_eval_to_sx(
    expr: Any,
    env: dict[str, Any],
    ctx: RequestContext | None = None,
) -> str:
    """Evaluate *expr* (resolving I/O inline) and produce sx source string.

    Unlike ``async_render`` (which produces HTML), this produces sx wire
    format suitable for fragment responses that clients render themselves.
    """
    if ctx is None:
        ctx = RequestContext()
    result = await _aser(expr, env, ctx)
    if isinstance(result, SxExpr):
        return result
    if result is None or result is NIL:
        return SxExpr("")
    if isinstance(result, str):
        return SxExpr(result)
    return SxExpr(serialize(result))


async def _maybe_expand_component_result(
    result: Any,
    env: dict[str, Any],
    ctx: RequestContext,
) -> Any:
    """If *result* is a component call (SxExpr or string starting with
    ``(~``), re-parse and expand it server-side.

    This ensures Python-only helpers (e.g. ``highlight``) inside the
    component body are evaluated on the server rather than being
    serialized for the client where they don't exist.
    """
    raw = None
    if isinstance(result, SxExpr):
        raw = str(result).strip()
    elif isinstance(result, str):
        raw = result.strip()
    if raw and raw.startswith("(~"):
        from .parser import parse_all
        parsed = parse_all(raw)
        if parsed:
            return await async_eval_slot_to_sx(parsed[0], env, ctx)
    return result


async def async_eval_slot_to_sx(
    expr: Any,
    env: dict[str, Any],
    ctx: RequestContext | None = None,
) -> str:
    """Like async_eval_to_sx but expands component calls.

    Used by defpage slot evaluation where the content expression is
    typically a component call like ``(~dashboard-content)``.  Normal
    ``async_eval_to_sx`` serializes component calls without expanding;
    this variant expands one level so IO primitives in the body execute,
    then serializes the result as SX wire format.
    """
    if ctx is None:
        ctx = RequestContext()

    # Enable server-side component expansion for this slot evaluation.
    # This lets _aser expand known components (so Python-only helpers
    # like highlight execute server-side) instead of serializing them
    # for client rendering.
    token = _expand_components.set(True)
    try:
        return await _eval_slot_inner(expr, env, ctx)
    finally:
        _expand_components.reset(token)


async def _eval_slot_inner(
    expr: Any,
    env: dict[str, Any],
    ctx: RequestContext,
) -> str:
    """Inner implementation — runs with _expand_components=True."""
    # If expr is a component call, expand it through _aser
    if isinstance(expr, list) and expr:
        head = expr[0]
        if isinstance(head, Symbol) and head.name.startswith("~"):
            comp = env.get(head.name)
            if isinstance(comp, Component):
                result = await _aser_component(comp, expr[1:], env, ctx)
                if isinstance(result, SxExpr):
                    return result
                if result is None or result is NIL:
                    return SxExpr("")
                if isinstance(result, str):
                    return SxExpr(result)
                return SxExpr(serialize(result))
            elif isinstance(comp, Island):
                pass  # Islands serialize as SX for client-side rendering
            else:
                import logging
                logging.getLogger("sx.eval").error(
                    "async_eval_slot_to_sx: component %s not found in env "
                    "(will fall through to _aser and serialize unexpanded — "
                    "client will see 'Unknown component'). "
                    "Check that the .sx file is loaded and the service's sx/ "
                    "directory is bind-mounted in docker-compose.dev.yml.",
                    head.name,
                )
    # Fall back to normal async_eval_to_sx
    result = await _aser(expr, env, ctx)
    # If the result is a component call (from case/if/let branches or
    # page helpers returning strings), re-parse and expand it server-side
    # so that Python-only helpers like ``highlight`` in the component body
    # get evaluated here, not on the client.
    result = await _maybe_expand_component_result(result, env, ctx)
    if isinstance(result, SxExpr):
        return result
    if result is None or result is NIL:
        return SxExpr("")
    if isinstance(result, str):
        return SxExpr(result)
    return SxExpr(serialize(result))


async def _aser(expr: Any, env: dict[str, Any], ctx: RequestContext) -> Any:
    """Evaluate *expr*, producing SxExpr for rendering forms, raw values
    for everything else."""
    if isinstance(expr, (int, float, bool)):
        return expr
    if isinstance(expr, SxExpr):
        return expr
    if isinstance(expr, str):
        return expr
    if expr is None or expr is NIL:
        return NIL

    if isinstance(expr, Symbol):
        name = expr.name
        if name in env:
            return env[name]
        if name in _PRIMITIVES:
            return _PRIMITIVES[name]
        if name == "true":
            return True
        if name == "false":
            return False
        if name == "nil":
            return NIL
        raise EvalError(f"Undefined symbol: {name}")

    if isinstance(expr, Keyword):
        return expr.name

    if isinstance(expr, dict):
        return {k: await _aser(v, env, ctx) for k, v in expr.items()}

    if not isinstance(expr, list):
        return expr
    if not expr:
        return []

    head = expr[0]
    if not isinstance(head, (Symbol, Lambda, list)):
        return [await _aser(x, env, ctx) for x in expr]

    if isinstance(head, Symbol):
        name = head.name

        # I/O primitives — await, return actual data
        if name in IO_PRIMITIVES:
            args, kwargs = await _parse_io_args(expr[1:], env, ctx)
            return await execute_io(name, args, kwargs, ctx)

        # <> — serialize children as sx fragment
        if name == "<>":
            return await _aser_fragment(expr[1:], env, ctx)

        # raw! — serialize
        if name == "raw!":
            return await _aser_call("raw!", expr[1:], env, ctx)

        # html: prefix → force tag serialization
        if name.startswith("html:"):
            return await _aser_call(name[5:], expr[1:], env, ctx)

        # Component call — expand macros, expand known components (in slot
        # eval context only), serialize unknown
        if name.startswith("~"):
            val = env.get(name)
            if isinstance(val, Macro):
                expanded = _expand_macro(val, expr[1:], env)
                return await _aser(expanded, env, ctx)
            if isinstance(val, Component):
                if _expand_components.get() or val.render_target == "server":
                    return await _aser_component(val, expr[1:], env, ctx)
            return await _aser_call(name, expr[1:], env, ctx)

        # Serialize-mode special/HO forms (checked BEFORE HTML_TAGS
        # because some names like "map" are both HTML tags and sx forms).
        # If name is also an HTML tag and (keyword arg or SVG context) → tag call.
        sf = _ASER_FORMS.get(name)
        if sf is not None:
            if name in HTML_TAGS and (
                (len(expr) > 1 and isinstance(expr[1], Keyword))
                or _svg_context.get(False)
            ):
                return await _aser_call(name, expr[1:], env, ctx)
            return await sf(expr, env, ctx)

        # Lake — serialize (server-morphable slot within island)
        if name == "lake":
            return await _aser_call(name, expr[1:], env, ctx)

        # Marsh — serialize (reactive server-morphable slot within island)
        if name == "marsh":
            return await _aser_call(name, expr[1:], env, ctx)

        # HTML tag — serialize (don't render to HTML)
        if name in HTML_TAGS:
            return await _aser_call(name, expr[1:], env, ctx)

        # Macro expansion
        if name in env:
            val = env[name]
            if isinstance(val, Macro):
                expanded = _expand_macro(val, expr[1:], env)
                return await _aser(expanded, env, ctx)

        # Custom element (hyphenated name with keyword attrs) → serialize as tag
        if "-" in name and len(expr) > 1 and isinstance(expr[1], Keyword):
            return await _aser_call(name, expr[1:], env, ctx)

        # SVG/MathML context → unknown names are child elements
        if _svg_context.get(False):
            return await _aser_call(name, expr[1:], env, ctx)

    # Function / lambda call — evaluate (produces data, not rendering)
    fn = await async_eval(head, env, ctx)
    args = [await async_eval(a, env, ctx) for a in expr[1:]]

    if callable(fn) and not isinstance(fn, (Lambda, Component)):
        result = fn(*args)
        if inspect.iscoroutine(result):
            return await result
        return result
    if isinstance(fn, Lambda):
        return await _async_trampoline(await _async_call_lambda(fn, args, env, ctx))
    if isinstance(fn, Component):
        # Component invoked as function — serialize the call
        return await _aser_call(f"~{fn.name}", expr[1:], env, ctx)
    raise EvalError(f"Not callable: {fn!r}")


async def _aser_fragment(children: list, env: dict, ctx: RequestContext) -> SxExpr:
    """Serialize ``(<> child1 child2 ...)`` to sx source."""
    parts: list[str] = []
    for child in children:
        result = await _aser(child, env, ctx)
        if isinstance(result, list):
            # Flatten lists (e.g. from map)
            for item in result:
                if item is not NIL and item is not None:
                    parts.append(serialize(item))
        elif result is not NIL and result is not None:
            parts.append(serialize(result))
    if not parts:
        return SxExpr("")
    return SxExpr("(<> " + " ".join(parts) + ")")


async def _aser_component(
    comp: Component, args: list, env: dict, ctx: RequestContext,
) -> Any:
    """Expand a component body through _aser — produces SX, not HTML."""
    kwargs: dict[str, Any] = {}
    children: list[Any] = []
    i = 0
    while i < len(args):
        arg = args[i]
        if isinstance(arg, Keyword) and i + 1 < len(args):
            kwargs[arg.name] = await _aser(args[i + 1], env, ctx)
            i += 2
        else:
            children.append(arg)
            i += 1
    local = dict(comp.closure)
    local.update(env)
    for p in comp.params:
        local[p] = kwargs.get(p, NIL)
    if comp.has_children:
        child_parts = []
        for c in children:
            child_parts.append(serialize(await _aser(c, env, ctx)))
        local["children"] = SxExpr("(<> " + " ".join(child_parts) + ")")
    return await _aser(comp.body, local, ctx)


async def _aser_call(
    name: str, args: list, env: dict, ctx: RequestContext,
) -> SxExpr:
    """Serialize ``(name :key val child ...)`` — evaluate args but keep
    as sx source instead of rendering to HTML."""
    # SVG/MathML namespace auto-detection for serializer
    token = None
    if name in ("svg", "math"):
        token = _svg_context.set(True)

    try:
        parts = [name]
        extra_class: str | None = None
        i = 0
        while i < len(args):
            arg = args[i]
            if isinstance(arg, Keyword) and i + 1 < len(args):
                val = await _aser(args[i + 1], env, ctx)
                if val is not NIL and val is not None:
                    parts.append(f":{arg.name}")
                    # Plain list → serialize for the client.
                    # Rendered items (SxExpr) → wrap in (<> ...) fragment.
                    # Data items (dicts, strings, numbers) → (list ...)
                    # so the client gets an iterable array, not a
                    # DocumentFragment that breaks map/filter.
                    if isinstance(val, list):
                        live = [v for v in val
                                if v is not NIL and v is not None]
                        items = [serialize(v) for v in live]
                        if not items:
                            parts.append("nil")
                        elif any(isinstance(v, SxExpr) for v in live):
                            parts.append(
                                "(<> " + " ".join(items) + ")"
                            )
                        else:
                            parts.append(
                                "(list " + " ".join(items) + ")"
                            )
                    else:
                        parts.append(serialize(val))
                i += 2
            else:
                result = await _aser(arg, env, ctx)
                if result is not NIL and result is not None:
                    # Flatten list results (e.g. from map) into individual
                    # children, matching _aser_fragment behaviour
                    if isinstance(result, list):
                        for item in result:
                            if item is not NIL and item is not None:
                                parts.append(serialize(item))
                    else:
                        parts.append(serialize(result))
                i += 1
        # If we converted a :style to a class, merge into existing :class or add it
        if extra_class:
            _merge_class_into_parts(parts, extra_class)
        return SxExpr("(" + " ".join(parts) + ")")
    finally:
        if token is not None:
            _svg_context.reset(token)


def _merge_class_into_parts(parts: list[str], class_name: str) -> None:
    """Merge an extra class name into the serialized parts list.

    If :class already exists, append to it. Otherwise add :class.
    """
    for i, p in enumerate(parts):
        if p == ":class" and i + 1 < len(parts):
            # Existing :class — append our class
            existing = parts[i + 1]
            if existing.startswith('"') and existing.endswith('"'):
                # Quoted string — insert before closing quote
                parts[i + 1] = existing[:-1] + " " + class_name + '"'
            else:
                # Expression — wrap in (str ...)
                parts[i + 1] = f'(str {existing} " {class_name}")'
            return
    # No existing :class — add one
    parts.insert(1, f'"{class_name}"')
    parts.insert(1, ":class")


# ---------------------------------------------------------------------------
# Serialize-mode special forms
# ---------------------------------------------------------------------------

async def _assf_if(expr, env, ctx):
    cond = await async_eval(expr[1], env, ctx)
    if cond and cond is not NIL:
        return await _aser(expr[2], env, ctx)
    if len(expr) > 3:
        return await _aser(expr[3], env, ctx)
    return NIL


async def _assf_when(expr, env, ctx):
    cond = await async_eval(expr[1], env, ctx)
    if cond and cond is not NIL:
        result: Any = NIL
        for body_expr in expr[2:]:
            result = await _aser(body_expr, env, ctx)
        return result
    return NIL


async def _assf_let(expr, env, ctx):
    bindings = expr[1]
    local = dict(env)
    if isinstance(bindings, list):
        if bindings and isinstance(bindings[0], list):
            for binding in bindings:
                var = binding[0]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await _aser(binding[1], local, ctx)
        elif len(bindings) % 2 == 0:
            for i in range(0, len(bindings), 2):
                var = bindings[i]
                vname = var.name if isinstance(var, Symbol) else var
                local[vname] = await _aser(bindings[i + 1], local, ctx)
    result: Any = NIL
    for body_expr in expr[2:]:
        result = await _aser(body_expr, local, ctx)
    return result


async def _assf_cond(expr, env, ctx):
    clauses = expr[1:]
    if not clauses:
        return NIL
    if (isinstance(clauses[0], list) and len(clauses[0]) == 2
            and not (isinstance(clauses[0][0], Symbol) and clauses[0][0].name in (
                "=", "<", ">", "<=", ">=", "!=", "and", "or"))):
        for clause in clauses:
            test = clause[0]
            if isinstance(test, Symbol) and test.name in ("else", ":else"):
                return await _aser(clause[1], env, ctx)
            if isinstance(test, Keyword) and test.name == "else":
                return await _aser(clause[1], env, ctx)
            if await async_eval(test, env, ctx):
                return await _aser(clause[1], env, ctx)
    else:
        i = 0
        while i < len(clauses) - 1:
            test = clauses[i]
            result = clauses[i + 1]
            if isinstance(test, Keyword) and test.name == "else":
                return await _aser(result, env, ctx)
            if isinstance(test, Symbol) and test.name in (":else", "else"):
                return await _aser(result, env, ctx)
            if await async_eval(test, env, ctx):
                return await _aser(result, env, ctx)
            i += 2
    return NIL


async def _assf_case(expr, env, ctx):
    match_val = await async_eval(expr[1], env, ctx)
    clauses = expr[2:]
    i = 0
    while i < len(clauses) - 1:
        test = clauses[i]
        result = clauses[i + 1]
        if isinstance(test, Keyword) and test.name == "else":
            return await _aser(result, env, ctx)
        if isinstance(test, Symbol) and test.name in (":else", "else"):
            return await _aser(result, env, ctx)
        if match_val == await async_eval(test, env, ctx):
            return await _aser(result, env, ctx)
        i += 2
    return NIL


async def _assf_begin(expr, env, ctx):
    result: Any = NIL
    for sub in expr[1:]:
        result = await _aser(sub, env, ctx)
    return result


async def _assf_define(expr, env, ctx):
    await async_eval(expr, env, ctx)
    return NIL


async def _assf_defisland(expr, env, ctx):
    """Evaluate defisland AND serialize it — client needs the definition."""
    import logging
    logging.getLogger("sx.eval").info("_assf_defisland called for: %s", expr[1] if len(expr) > 1 else expr)
    await async_eval(expr, env, ctx)
    return serialize(expr)


async def _assf_lambda(expr, env, ctx):
    return await _asf_lambda(expr, env, ctx)


async def _assf_and(expr, env, ctx):
    return await _asf_and(expr, env, ctx)


async def _assf_or(expr, env, ctx):
    return await _asf_or(expr, env, ctx)


async def _assf_quote(expr, env, ctx):
    return expr[1] if len(expr) > 1 else NIL


async def _assf_quasiquote(expr, env, ctx):
    return await _async_qq_expand(expr[1], env, ctx)


async def _assf_thread_first(expr, env, ctx):
    return await _asf_thread_first(expr, env, ctx)


async def _assf_set_bang(expr, env, ctx):
    return await _asf_set_bang(expr, env, ctx)


# ---------------------------------------------------------------------------
# Serialize-mode higher-order forms
# ---------------------------------------------------------------------------

async def _asho_ser_map(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for item in coll:
        if isinstance(fn, Lambda):
            local = dict(fn.closure)
            local.update(env)
            for p, v in zip(fn.params, [item]):
                local[p] = v
            results.append(await _aser(fn.body, local, ctx))
        elif callable(fn):
            r = fn(item)
            results.append(await r if inspect.iscoroutine(r) else r)
        else:
            raise EvalError(f"map requires callable, got {type(fn).__name__}")
    return results


async def _asho_ser_map_indexed(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for i, item in enumerate(coll):
        if isinstance(fn, Lambda):
            local = dict(fn.closure)
            local.update(env)
            local[fn.params[0]] = i
            local[fn.params[1]] = item
            results.append(await _aser(fn.body, local, ctx))
        elif callable(fn):
            r = fn(i, item)
            results.append(await r if inspect.iscoroutine(r) else r)
        else:
            raise EvalError(f"map-indexed requires callable, got {type(fn).__name__}")
    return results


async def _asho_ser_filter(expr, env, ctx):
    # filter is value-producing, delegate to eval
    return await async_eval(expr, env, ctx)


async def _asho_ser_for_each(expr, env, ctx):
    fn = await async_eval(expr[1], env, ctx)
    coll = await async_eval(expr[2], env, ctx)
    results = []
    for item in coll:
        if isinstance(fn, Lambda):
            local = dict(fn.closure)
            local.update(env)
            local[fn.params[0]] = item
            results.append(await _aser(fn.body, local, ctx))
        elif callable(fn):
            r = fn(item)
            results.append(await r if inspect.iscoroutine(r) else r)
    return results


_ASER_FORMS: dict[str, Any] = {
    "if": _assf_if,
    "when": _assf_when,
    "cond": _assf_cond,
    "case": _assf_case,
    "and": _assf_and,
    "or": _assf_or,
    "let": _assf_let,
    "let*": _assf_let,
    "lambda": _assf_lambda,
    "fn": _assf_lambda,
    "define": _assf_define,
    "defstyle": _assf_define,
    "defcomp": _assf_define,
    "defmacro": _assf_define,
    "defhandler": _assf_define,
    "defisland": _assf_defisland,
    "begin": _assf_begin,
    "do": _assf_begin,
    "quote": _assf_quote,
    "quasiquote": _assf_quasiquote,
    "->": _assf_thread_first,
    "set!": _assf_set_bang,
    "map": _asho_ser_map,
    "map-indexed": _asho_ser_map_indexed,
    "filter": _asho_ser_filter,
    "for-each": _asho_ser_for_each,
}
