# AUTO-GENERATED by scripts/codegen_kernel_abi.py — DO NOT EDIT
"""Thin gRPC `Call` dispatch for kernel syscalls.

Source of truth: PyKernel ``#[pymethods]`` in
``rust/kernel/src/generated_kernel_abi_pyo3.rs``.  The codegen
scans that file for syscall methods (sys_read, sys_write,
sys_setattr, …) and emits this module so the gRPC servicer can
route calls to the Python ``NexusFS`` wrapper without any
@rpc_expose / dispatch.py / handlers/filesystem.py / params-dataclass
chain.

The dispatcher uses ``inspect.signature`` to filter kwargs against
the matching NexusFS method, so adding a new kernel syscall is just
``_KERNEL_SYSCALL_ALLOWLIST`` + re-run the codegen.  Methods that
take ``**attrs`` (sys_setattr) get arbitrary extras forwarded
unchanged.

Mutation syscalls fire pub/sub events through the gRPC servicer's
``subscription_manager`` (file_write / file_delete / file_rename /
dir_create / dir_delete) — see ``_EVENTS`` for the wire-name →
event-metadata mapping.  Tier 2 wire-shape adapters
(``{"deleted": True}`` / ``{"renamed": True}`` / ``bytes_written``
merge / ``{"acquired", "lock_id"}``) live in
``_apply_result_adapter`` and preserve the legacy ``handle_*``
wrapper response shapes.

Re-generate: python scripts/codegen_kernel_abi.py
"""

from __future__ import annotations

import asyncio
import inspect
import logging
from typing import Any

from nexus.contracts.constants import ROOT_ZONE_ID

logger = logging.getLogger(__name__)


# Wire-form RPC method names this module handles. The gRPC servicer
# tests membership BEFORE invoking the legacy ``dispatch_method`` path.
KERNEL_SYSCALL_NAMES: frozenset[str] = frozenset(
    {
        "access",
        "close",
        "delete",
        "exists",
        "flush_write_buffer",
        "fsync",
        "is_directory",
        "list",
        "lock_acquire",
        "mkdir",
        "read",
        "rename",
        "rmdir",
        "sync",
        "sys_copy",
        "sys_lock",
        "sys_mkdir",
        "sys_read",
        "sys_readdir",
        "sys_rename",
        "sys_rmdir",
        "sys_setattr",
        "sys_stat",
        "sys_unlink",
        "sys_unlock",
        "sys_write",
        "write",
    }
)


# Wire name → canonical NexusFS method name (see codegen for the why).
_ALIASES: dict[str, str] = {
    "delete": "sys_unlink",
    "exists": "access",
    "flush_write_buffer": "flush_write_buffer",
    "fsync": "fsync",
    "list": "sys_readdir",
    "lock_acquire": "sys_lock",
    "mkdir": "mkdir",
    "read": "read",
    "rename": "sys_rename",
    "rmdir": "rmdir",
    "sync": "sync",
    "sys_mkdir": "mkdir",
    "sys_readdir": "sys_readdir",
    "sys_rmdir": "rmdir",
    "sys_write": "write",
    "write": "write",
}


# Wire name → event metadata for pub/sub firing on mutation syscalls.
# Mirrors the legacy ``DispatchEntry.event_*`` fields.
_EVENTS: dict[str, dict[str, str]] = {
    "delete": {"event_type": "file_delete", "path_attr": "path"},
    "mkdir": {"event_type": "dir_create", "path_attr": "path"},
    "rename": {"event_type": "file_rename", "path_attr": "new_path", "old_path_attr": "old_path"},
    "rmdir": {"event_type": "dir_delete", "path_attr": "path"},
    "sys_mkdir": {"event_type": "dir_create", "path_attr": "path"},
    "sys_rename": {
        "event_type": "file_rename",
        "path_attr": "new_path",
        "old_path_attr": "old_path",
    },
    "sys_rmdir": {"event_type": "dir_delete", "path_attr": "path"},
    "sys_unlink": {"event_type": "file_delete", "path_attr": "path"},
    "sys_write": {"event_type": "file_write", "path_attr": "path", "size_key": "bytes_written"},
    "write": {"event_type": "file_write", "path_attr": "path", "size_key": "bytes_written"},
}


def _resolve_method_name(method: str) -> str:
    """Resolve a wire-form RPC name to the NexusFS method to invoke."""
    return _ALIASES.get(method, method)


def _build_kwargs(
    func: Any,
    params: dict[str, Any],
    context: Any,
) -> dict[str, Any]:
    """Filter ``params`` to ``func``'s signature; forward extras to **kwargs.

    Methods like ``sys_setattr(path, *, context, **attrs)`` collect
    arbitrary extras through their VAR_KEYWORD param — without explicit
    forwarding, ``inspect.signature`` would drop them.  This is how
    DT_MOUNT's ``entry_type`` / ``zone_id`` / ``backend_name`` reach
    the kernel without each needing a named slot in NexusFS's signature.
    """
    sig = inspect.signature(func)
    has_var_keyword = any(p.kind == inspect.Parameter.VAR_KEYWORD for p in sig.parameters.values())

    kwargs: dict[str, Any] = {}
    matched: set[str] = set()
    for name, param in sig.parameters.items():
        if name == "self" or param.kind == inspect.Parameter.VAR_KEYWORD:
            continue
        if name in ("context", "_context"):
            kwargs[name] = context
            matched.add(name)
        elif name in params:
            kwargs[name] = params[name]
            matched.add(name)

    if has_var_keyword:
        for k, v in params.items():
            if k not in matched and k not in ("context", "_context"):
                kwargs[k] = v

    return kwargs


def _apply_result_adapter(
    method: str,
    raw_result: Any,
    params: dict[str, Any],
) -> Any:
    """Re-shape NexusFS results into the wire format Tier 2 callers expect.

    Mirrors the response shapes the legacy ``handle_*`` wrappers
    produced so deleting the legacy chain is a no-op on the wire.

    Path-bearing dicts (sys_stat metadata, sys_readdir entries) get
    their internal ``/zone/<id>/...`` prefix stripped via
    ``unscope_internal_dict`` so RPC callers see user-facing paths.

    ``params`` is the wire-form input dict — required for write
    syscalls so we can compute ``bytes_written`` from the request
    payload (``content`` / ``buf``) when the kernel returns just
    ``None`` or the new metadata.
    """
    from nexus.server.path_utils import (
        unscope_internal_dict,
        unscope_internal_path,
    )

    if method in ("delete", "sys_unlink"):
        return {"deleted": True}
    if method in ("rename", "sys_rename"):
        return {"renamed": True}
    if method in ("mkdir", "sys_mkdir"):
        return {"created": True}
    if method in ("rmdir", "sys_rmdir"):
        return {"removed": True}
    if method in ("write", "sys_write"):
        content = params.get("content") or params.get("buf") or b""
        content_len = len(content.encode("utf-8")) if isinstance(content, str) else len(content)
        wire: dict[str, Any] = {"bytes_written": content_len}
        if isinstance(raw_result, dict):
            wire.update(raw_result)
        return wire
    if method == "sys_stat":
        # `Kernel::sys_stat` returns the StatResult dict directly
        # (or `None` for missing files).  Match the kernel ABI shape
        # so consumers can test `result is None` for not-found.  The
        # legacy `{"metadata": ...}` wrap was for a Tier 2 caller
        # that has been deleted; the `unscope_internal_dict` call is
        # still applied to strip internal `path` keys from the dict.
        if isinstance(raw_result, dict):
            return unscope_internal_dict(raw_result, ["path"])
        return raw_result
    if method in ("access", "exists"):
        return {"exists": bool(raw_result)}
    if method == "is_directory":
        return {"is_directory": bool(raw_result)}
    if method == "sys_unlock":
        return {"released": bool(raw_result)}
    if method == "sys_lock":
        # Tier 1 contract: bare `lock_id` string when granted, `None`
        # on contention.  Matches the kernel ABI's `Option<String>`
        # exactly; consumers test `result is None` for contention.
        # The legacy Tier 2 dict wrapper was removed alongside
        # `lock_acquire` in commit 231620c3c.
        return raw_result
    if method == "lock_acquire":
        return {"acquired": raw_result is not None, "lock_id": raw_result}
    if method in ("sys_readdir", "list"):
        # PaginatedResult (from limit kwarg) → wire dict; bare list →
        # wire dict with has_more=False / next_cursor=None.
        if hasattr(raw_result, "to_dict"):
            paginated = raw_result.to_dict()
            items = [
                unscope_internal_dict(f, ["path", "virtual_path"])
                if isinstance(f, dict)
                else unscope_internal_path(f)
                for f in paginated["items"]
            ]
            return {
                "files": items,
                "next_cursor": paginated["next_cursor"],
                "has_more": paginated["has_more"],
                "total_count": paginated.get("total_count"),
            }
        raw_entries = raw_result if isinstance(raw_result, list) else []
        entries = [
            unscope_internal_dict(f, ["path", "virtual_path"])
            if isinstance(f, dict)
            else unscope_internal_path(f)
            for f in raw_entries
        ]
        return {"files": entries, "has_more": False, "next_cursor": None}
    return raw_result


async def _fire_event(
    subscription_manager: Any,
    method: str,
    params: dict[str, Any],
    result: Any,
    context: Any,
) -> None:
    """Fire pub/sub event for a mutation syscall (best-effort, non-blocking).

    No-op when the wire method has no entry in ``_EVENTS`` or
    ``subscription_manager`` is None.  Pulls path / old_path / size
    from the params dict + result via ``_EVENTS`` metadata.
    """
    meta = _EVENTS.get(method)
    if meta is None or subscription_manager is None:
        return

    try:
        zone_id = getattr(context, "zone_id", None) or ROOT_ZONE_ID
        path = params.get(meta.get("path_attr", "path"))
        if not path:
            return
        data: dict[str, Any] = {"file_path": path}
        old_path_attr = meta.get("old_path_attr")
        if old_path_attr:
            old_path = params.get(old_path_attr)
            if old_path:
                data["old_path"] = old_path
        size_key = meta.get("size_key")
        if size_key and isinstance(result, dict) and size_key in result:
            data["size"] = result[size_key]
        await subscription_manager.broadcast(meta["event_type"], data, zone_id)
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "[kernel-syscall-dispatch] event firing failed for %s on %s: %s",
            meta.get("event_type"),
            method,
            exc,
        )


async def _occ_write_dispatch(
    nexus_fs: Any,
    params: dict[str, Any],
    context: Any,
) -> Any:
    """Special-case write with OCC (``if_match`` / ``if_none_match``).

    Mirrors what the legacy ``handle_write`` wrapper did — when the
    RPC carries OCC kwargs and ``force`` is not set, route through
    ``nexus.lib.occ.occ_write`` so the CAS check happens at the RPC
    layer.  Otherwise fall through to plain ``NexusFS.write``.

    Returned bytes_written count is always derived from the content
    payload (see ``_apply_result_adapter``).
    """
    from nexus.lib.occ import occ_write

    content = params.get("content") or params.get("buf") or b""
    if isinstance(content, str):
        content = content.encode("utf-8")
    offset = int(params.get("offset", 0) or 0)
    if_match = params.get("if_match") or None
    if_none_match = bool(params.get("if_none_match"))
    force = bool(params.get("force"))

    if (if_match or if_none_match) and not force:
        return await occ_write(
            nexus_fs,
            params["path"],
            content,
            context=context,
            if_match=if_match,
            if_none_match=if_none_match,
            offset=offset,
        )
    # #4005 round-2: NexusFS.write is sync — offload like dispatch_kernel_syscall.
    return await asyncio.to_thread(
        nexus_fs.write, params["path"], content, context=context, offset=offset
    )


def _apply_pre_call_defaults(method: str, params: dict[str, Any]) -> dict[str, Any]:
    """Apply legacy-conservative defaults to wire params (Codex review of #3701).

    ``NexusFS.mkdir`` defaults to ``parents=True, exist_ok=True``
    (mkdir -p) and ``NexusFS.rmdir`` defaults to ``recursive=True``
    (rm -rf), but the legacy ``mkdir`` / ``rmdir`` / ``sys_mkdir``
    / ``sys_rmdir`` RPC aliases have always been conservative so
    legacy clients sending only ``{"path": "/foo"}`` get a plain
    mkdir that errors on missing parents / existing paths and a
    non-recursive rmdir.  Dropping the override would silently turn
    safe legacy calls into destructive recursive deletes / silent
    mkdir-p — a real behavioral regression flagged by Codex.

    Returns a (possibly defaulted) shallow copy of params; the
    caller passes the result to ``_build_kwargs``.  Non-syscall
    methods get back the same dict.
    """
    if method in ("mkdir", "sys_mkdir"):
        out = dict(params)
        out.setdefault("parents", False)
        out.setdefault("exist_ok", False)
        return out
    if method in ("rmdir", "sys_rmdir"):
        out = dict(params)
        out.setdefault("recursive", False)
        return out
    return params


async def dispatch_kernel_syscall(
    nexus_fs: Any,
    method: str,
    params: dict[str, Any],
    context: Any,
    *,
    subscription_manager: Any = None,
) -> Any:
    """Dispatch a kernel syscall RPC to the matching ``NexusFS`` method.

    * Resolves wire alias to canonical NexusFS method via ``_ALIASES``.
    * Filters params to the method's signature with ``_build_kwargs``;
      ``**attrs`` methods (sys_setattr) get extras forwarded.
    * Special-cases ``write`` / ``sys_write`` for OCC (if_match /
      if_none_match) — routes through ``occ_write`` when those kwargs
      are present.
    * Applies legacy-conservative ``mkdir`` / ``rmdir`` defaults via
      ``_apply_pre_call_defaults`` (Codex review of #3701).
    * Reshapes the result via ``_apply_result_adapter`` so legacy
      Tier 2 wire shapes are preserved (``{"deleted": True}`` etc.).
    * Fires the matching pub/sub event via ``_fire_event`` when the
      caller passes a ``subscription_manager``.

    Returns the (possibly adapted) result; gRPC servicer encodes it.
    """
    # ``close`` is a client-side lifecycle method that some REMOTE service
    # proxies erroneously dispatch through the RPC transport during teardown.
    # Calling ``NexusFS.close`` server-side would tear down the live server
    # filesystem — short-circuit to a no-op so the wire call succeeds without
    # affecting server state.
    if method == "close":
        return {}

    params = _apply_pre_call_defaults(method, params)

    if method in ("write", "sys_write"):
        raw_result = await _occ_write_dispatch(nexus_fs, params, context)
    else:
        canonical = _resolve_method_name(method)
        func = getattr(nexus_fs, canonical, None)
        if func is None:
            raise AttributeError(
                f"NexusFS has no method {canonical!r} (from RPC method {method!r})"
            )

        kwargs = _build_kwargs(func, params, context)

        if asyncio.iscoroutinefunction(func):
            raw_result = await func(**kwargs)
        else:
            # #4005 round-2: NexusFS sys_* methods are sync and may do
            # blocking I/O (DB hits, large reads). Calling them inline
            # would park the asyncio loop — same DoS class that justified
            # excluding sys_watch. Offload to the default thread pool.
            raw_result = await asyncio.to_thread(func, **kwargs)

    result = _apply_result_adapter(method, raw_result, params)

    await _fire_event(subscription_manager, method, params, result, context)

    return result
