"""Unified filesystem implementation for Nexus."""
# Kernel interface unification — see KERNEL-ARCHITECTURE.md §4.5

import builtins
import contextlib
import inspect
import logging
from collections.abc import Callable
from pathlib import Path
from typing import Any

from nexus.contracts.cache_store import CacheStoreABC, NullCacheStore
from nexus.contracts.constants import ROOT_ZONE_ID
from nexus.contracts.types import OperationContext
from nexus.core.config import (
    CacheConfig,
    DistributedConfig,
    MemoryConfig,
    ParseConfig,
    PermissionConfig,
)
from nexus.core.nexus_fs_content import ContentMixin
from nexus.core.nexus_fs_dispatch import DispatchMixin
from nexus.core.nexus_fs_internal import InternalMixin
from nexus.core.nexus_fs_metadata import MetadataMixin
from nexus.core.nexus_fs_watch import WatchMixin
from nexus.storage.record_store import RecordStoreABC

logger = logging.getLogger(__name__)


def _declares_hook_spec(instance: Any) -> bool:
    """Return True only when ``hook_spec`` is a real attribute on the object."""
    try:
        attr = inspect.getattr_static(instance, "hook_spec")
    except AttributeError:
        return False
    return callable(attr)


def _register_hooks_for_spec(dispatch: Any, spec: Any) -> None:
    """Register all hooks from a HookSpec onto the dispatch (NexusFS)."""
    if spec is None or spec.is_empty:
        return
    for h in spec.resolvers:
        dispatch.register_resolver(h)
    for h in spec.read_hooks:
        dispatch.register_intercept_read(h)
    for h in spec.write_hooks:
        dispatch.register_intercept_write(h)
    for h in spec.write_batch_hooks:
        dispatch.register_intercept_write_batch(h)
    for h in spec.delete_hooks:
        dispatch.register_intercept_delete(h)
    for h in spec.rename_hooks:
        dispatch.register_intercept_rename(h)
    for h in spec.copy_hooks:
        dispatch.register_intercept_copy(h)
    for h in spec.mkdir_hooks:
        dispatch.register_intercept_mkdir(h)
    for h in spec.rmdir_hooks:
        dispatch.register_intercept_rmdir(h)
    for h in spec.stat_hooks:
        dispatch.register_intercept_stat(h)
    for h in spec.access_hooks:
        dispatch.register_intercept_access(h)


def _unregister_hooks_for_spec(dispatch: Any, spec: Any) -> None:
    """Unregister all hooks from a HookSpec."""
    if spec is None or spec.is_empty:
        return
    for h in spec.resolvers:
        dispatch.unregister_resolver(h)
    for h in spec.read_hooks:
        dispatch.unregister_intercept_read(h)
    for h in spec.write_hooks:
        dispatch.unregister_intercept_write(h)
    for h in spec.write_batch_hooks:
        dispatch.unregister_intercept_write_batch(h)
    for h in spec.delete_hooks:
        dispatch.unregister_intercept_delete(h)
    for h in spec.rename_hooks:
        dispatch.unregister_intercept_rename(h)
    for h in spec.copy_hooks:
        dispatch.unregister_intercept_copy(h)
    for h in spec.mkdir_hooks:
        dispatch.unregister_intercept_mkdir(h)
    for h in spec.rmdir_hooks:
        dispatch.unregister_intercept_rmdir(h)
    for h in spec.stat_hooks:
        dispatch.unregister_intercept_stat(h)
    for h in spec.access_hooks:
        dispatch.unregister_intercept_access(h)


class NexusFS(  # type: ignore[misc]
    ContentMixin,
    MetadataMixin,
    InternalMixin,
    DispatchMixin,
    WatchMixin,
):
    """
    Unified filesystem for Nexus.

    Provides file operations (read, write, delete) with metadata tracking
    using content-addressable storage (CAS) for automatic deduplication.

    Works with any backend (local, GCS, S3, etc.) that implements the Backend interface.

    All backends use CAS by default for:
    - Automatic deduplication (same content stored once)
    - Content integrity (hash verification)
    - Efficient storage
    """

    def __init__(
        self,
        metadata_store: "Any | str | Path | None",
        record_store: RecordStoreABC | None = None,
        cache_store: CacheStoreABC | None = None,
        *,
        cache: CacheConfig | None = None,
        permissions: PermissionConfig | None = None,
        distributed: DistributedConfig | None = None,
        memory: MemoryConfig | None = None,
        parsing: ParseConfig | None = None,
        init_cred: OperationContext | None = None,
    ):
        """Initialize NexusFS kernel.

        Backends are mounted via ``DriverLifecycleCoordinator.mount()``
        (which writes to the Rust kernel's MountTable) — like Linux VFS,
        no global backend.

        Args:
            metadata_store: Metastore wiring. Four accepted shapes:

                - ``PyKernel``: a pre-built kernel (typically from
                  ``nexus.connect()``). Reused as-is; ``set_metastore_path``
                  is assumed to have been called on it already.
                - ``str | Path``: redb file path. NexusFS constructs a fresh
                  ``PyKernel`` and calls ``set_metastore_path(str(path))``.
                  This is the default for tests and quickstarts — the Rust
                  kernel is the SSOT, no parallel Python store opens the file.
                - ``None``: tempfile fallback — a fresh kernel + a tempfile
                  redb path.
                - Anything else exposing a ``_rust_kernel`` attribute (the
                  legacy ``RustMetastoreProxy`` shim, kept until the proxy
                  class is fully deleted): unwrapped to its kernel.

            init_cred: Kernel process credential — like Linux ``init_task.cred``.
                Used as fallback identity for internal operations (audit pipe
                writes, service bootstrap mkdir). Immutable after construction.
                Pass ``None`` only for bare-kernel tests that never call syscalls.
        """
        # Normalize the four input shapes into a single shape that the
        # kernel-boot branch below understands:
        #   * (kernel, path) — kernel ready to wire, path opened by us.
        #   * (kernel, None) — kernel already wired (or in-memory).
        #   * (None, path) — build a fresh kernel and wire it.
        #   * (None, None) — build a fresh kernel + tempfile redb.
        _metastore_path: str | None = None
        _provided_kernel: Any = None
        if isinstance(metadata_store, Path):
            _metastore_path = str(metadata_store)
        elif isinstance(metadata_store, str):
            _metastore_path = metadata_store
        elif metadata_store is None:
            pass
        elif hasattr(metadata_store, "_rust_kernel"):
            # Legacy ``RustMetastoreProxy`` shim — unwrap.
            _provided_kernel = metadata_store._rust_kernel
        else:
            # Bare ``PyKernel`` (post-W3b shape).
            _provided_kernel = metadata_store
        # ``metadata_store`` is no longer read past this point.
        metadata_store = None
        # Config defaults
        cache = cache or CacheConfig()
        permissions = permissions or PermissionConfig()
        distributed = distributed or DistributedConfig()
        memory = memory or MemoryConfig()
        parsing = parsing or ParseConfig()

        # Kernel zone identity — analogous to Linux sb->s_dev.
        # Standalone: always ROOT_ZONE_ID. Federation: set at link time.
        self._zone_id: str = ROOT_ZONE_ID

        self._cache_config = cache
        self._perm_config = permissions
        self._distributed_config = distributed
        self._memory_config_obj = memory
        self._parse_config = parsing
        self._config: Any | None = None

        # Map config fields to flat attributes
        self._enable_memory_paging = memory.enable_paging
        self._memory_main_capacity = memory.main_capacity
        self._memory_recall_max_age_hours = memory.recall_max_age_hours
        # _enforce_permissions removed — permission enforcement is fully delegated
        # to KernelDispatch INTERCEPT hooks. RebacPermissionCheckHook holds the flag
        # internally. No hook = no check = zero overhead (~20ns set lookup).
        self._enforce_zone_isolation = permissions.enforce_zone_isolation
        self.allow_admin_bypass = permissions.allow_admin_bypass

        # Three pillars: metadata (kernel-owned), record store, cache store.
        # No self.backend — all I/O goes through router.route().backend.
        # ``self.metadata`` is gone post-W3b: callers reach the metastore
        # via ``self._kernel.metastore_*`` (kernel-internal infrastructure)
        # or through ``nexus.kernel_helpers`` (the five non-trivial
        # wrappers around ``kernel.metastore_*``). The kernel handle is
        # the only Python-side metastore reference.
        self._record_store = record_store
        self._sql_engine: Any = None
        self._db_session_factory: Any = None
        self.SessionLocal: Any = None
        if record_store is not None:
            self._sql_engine = record_store.engine
            self._db_session_factory = record_store.session_factory
            self.SessionLocal = self._db_session_factory

        # Initialize cache store (Task #22: Fourth Pillar)
        self.cache_store: CacheStoreABC = (
            cache_store if cache_store is not None else NullCacheStore()
        )

        # Issue #1801: kernel process credential — like Linux init_task.cred.
        # Immutable after construction. Used as fallback identity for internal
        # operations. External callers should pass explicit context= to syscalls.
        self._init_cred: OperationContext | None = init_cred

        # ── Kernel-owned primitives (always present, created here) ──────
        # See KERNEL-ARCHITECTURE.md §1 DI patterns table.

        # Advisory locks handled by Rust kernel LockManager (sys_lock / sys_unlock).

        self._init_dispatch()

        from nexus.core.dispatch import get_global_registry

        self._ops_registry = get_global_registry()
        self._mounted_backend_instances: dict[str, Any] = {}

        import os as _os_ipc

        _ipc_self_addr = _os_ipc.environ.get("NEXUS_ADVERTISE_ADDR")

        self._transport_pool = None
        if _ipc_self_addr:
            from nexus.remote.rpc_transport import RPCTransportPool as _RPCTransportPool

            self._transport_pool = _RPCTransportPool()

        # ── Kernel (Issue #1817 — single-FFI sys_read/sys_write) ──
        # Constructed BEFORE DriverLifecycleCoordinator so DLC sees the
        # kernel from birth (VFSRouter is the Rust SSOT for routing).
        from nexus._rust_compat import RUST_AVAILABLE

        self._kernel = None
        if RUST_AVAILABLE:
            try:
                if _provided_kernel is not None:
                    # Caller already owns a kernel (typically from
                    # ``nexus.connect()`` or a previous NexusFS) — reuse
                    # it rather than constructing a parallel one. We
                    # assume ``set_metastore_path`` is already in place
                    # if ``_metastore_path`` is None.
                    self._kernel = _provided_kernel
                    if _metastore_path is not None:
                        self._kernel.set_metastore_path(str(_metastore_path))
                else:
                    from nexus_runtime import PyKernel as _Kernel

                    self._kernel = _Kernel()
                    # Start write-buffer background flusher (production only).
                    # Test kernels skip this — see make_test_nexus.
                    self._kernel.start_write_buffer_flusher(250)
                    # Phase 4 (full): drain federation's blob-fetcher
                    # slot + install real `PeerBlobClient` (idempotent).
                    # Log on failure so a stale wheel surfaces in the
                    # logs rather than silently disabling federation.
                    # Skip federation wiring during pytest — parallel xdist
                    # workers compete for the same gRPC port causing 8+ min
                    # hangs. Federation is an integration-level concern; unit
                    # tests pass a pre-built kernel via _provided_kernel instead.
                    if "pytest" not in __import__("sys").modules:
                        try:
                            import nexus_runtime as _nk

                            _nk.install_federation_wiring(self._kernel)
                            _nk.install_transport_wiring(self._kernel)
                        except Exception as _wiring_exc:
                            logger.warning(
                                "install_transport_wiring/install_federation_wiring "
                                "failed (federation peer-blob fetch will fall back "
                                "to Noop): %s",
                                _wiring_exc,
                            )
                    # Wire redb metastore — ALL reads and writes go through
                    # Rust redb. When the caller handed us a path, open
                    # against it. When ``None`` was passed, fall back to a
                    # tempfile redb (suitable for fixtures that don't care
                    # about on-disk persistence).
                    if _metastore_path is None:
                        import os
                        import tempfile

                        fd, tmp_path = tempfile.mkstemp(suffix=".redb")
                        os.close(fd)
                        _metastore_path = tmp_path

                    self._kernel.set_metastore_path(str(_metastore_path))
                    # All kernel wiring that depends on other NexusFS attributes
                    # (_mount_table, _vfs_lock_manager) is deferred to after
                    # __init__ completes — see "Deferred kernel wiring" block below.
            except Exception as exc:
                import logging as _logging

                _logging.getLogger(__name__).warning(
                    "Kernel init failed — falling back to Python path: %s", exc
                )
                self._kernel = None

        from nexus.core.driver_lifecycle_coordinator import DriverLifecycleCoordinator

        self._driver_coordinator: DriverLifecycleCoordinator = DriverLifecycleCoordinator(
            self,
            kernel=self._kernel,
        )

        # Deferred kernel wiring: bind mount table + VFS lock after all attributes exist.
        if self._kernel is not None:
            _mt = getattr(self._driver_coordinator, "_mount_table", None)
            if _mt is not None:
                _mt.bind_kernel(self._kernel)
            _vfs_rust = getattr(getattr(self, "_vfs_lock_manager", None), "_rust", None)
            if _vfs_rust is not None:
                self._kernel.set_vfs_lock(_vfs_rust)

        logger.info(
            "IPC primitives initialized: DriverCoordinator (self_address=%s)",
            _ipc_self_addr or "none/single-node",
        )

        # ── Kernel-knows (sentinel None, injected by factory) ───────────
        # See KERNEL-ARCHITECTURE.md §1 DI patterns table.
        # None = graceful degrade (like Linux LSM: no module loaded = no check).

        self._event_bus: Any = None
        # Hook specs for enlisted services (duck-typed hook_spec() capture)
        self._hook_specs: dict[str, Any] = {}
        # Lifecycle state — set by link() / initialize() / bootstrap()
        self._linked: bool = False
        self._initialized: bool = False
        self._bootstrapped: bool = False
        self._close_callbacks: list[
            Callable[[], None]
        ] = []  # Issue #1793: factory-registered service close
        self._runtime_closeables: list[Any] = []

    # =====================================================================
    # Lifecycle methods: link() → initialize() → bootstrap()
    #
    # Linearized in PR #3371 Phase 2: create_nexus_fs() calls
    # _wire_services() / _initialize_services() directly and sets the
    # flags. These methods are now flag-only no-ops for backward compat.
    # =====================================================================

    def link(
        self,
        *,
        enabled_bricks: "frozenset[str] | None" = None,
        parsing: Any = None,
        workflow_engine: Any = None,
    ) -> None:
        """Phase 1: Wire service topology — flag-only no-op.

        Actual wiring is done by factory._lifecycle._wire_services(),
        called directly from create_nexus_fs(). This method exists for
        backward compatibility (tests, manual construction).
        """
        if self._linked:
            return
        self._linked = True

    def initialize(self) -> None:
        """Phase 2: One-time side effects — flag-only no-op.

        Actual initialization is done by factory._lifecycle._initialize_services(),
        called directly from create_nexus_fs(). This method exists for
        backward compatibility (tests, manual construction).
        """
        if self._initialized:
            return
        if not self._linked:
            self.link()
        self._initialized = True

    def bootstrap(self) -> None:
        """Phase 3: Start background services.  Server/Worker only.

        Auto-starts all BackgroundService instances via Rust kernel
        service_start_all().

        Idempotent — guarded by ``_bootstrapped`` flag.
        """
        if self._bootstrapped:
            return
        if not self._initialized:
            self.initialize()
        # Auto-lifecycle: start BackgroundService instances (Issue #1580)
        if self._kernel is not None:
            self._kernel.service_start_all(30000)
            self._kernel.service_mark_bootstrapped()
        self._bootstrapped = True

    def _register_runtime_closeable(self, resource: Any) -> None:
        """Register a process-local resource to close with the filesystem.

        Used for runtime-owned handles that are not persisted in metadata
        and are not discoverable through the normal service graph, such as
        the REMOTE client's shared RPC transport.
        """
        self._runtime_closeables.append(resource)

    # -- Service registry accessors (Issue #1452) ---------------------------

    def service(self, name: str) -> Any | None:
        """Look up a registered service by canonical name.

        Returns the service instance, or ``None`` if not registered.
        Delegates to Rust kernel ServiceRegistry.
        """
        if self._kernel is None:
            return None
        return self._kernel.service_lookup(name)

    @property
    def service_registry(self) -> Any:
        """Read-only access for factory / diagnostics. Returns self (NexusFS)."""
        return self

    # Services accessed via self.service("name") → Rust kernel (Issue #1452).
    # Registered by factory via enlist_services() at link().

    @property
    def service_coordinator(self) -> Any:
        """Lifecycle coordinator. Returns self — NexusFS delegates to Rust kernel."""
        return self

    def swap_service(self, name: str, new_instance: Any, **kwargs: Any) -> None:
        """Hot-swap a service — drain + replace + rehook (#1452)."""
        exports = kwargs.get("exports", ())
        hook_spec = kwargs.get("hook_spec")
        timeout = kwargs.get("drain_timeout", 10.0)
        timeout_ms = int(timeout * 1000)

        # Unregister old hooks
        old_spec = self._hook_specs.get(name)
        if old_spec is None:
            old_instance = self._kernel.service_lookup(name)
            if old_instance is not None and _declares_hook_spec(old_instance):
                old_spec = old_instance.hook_spec()
        if old_spec is not None:
            _unregister_hooks_for_spec(self, old_spec)

        # Rust side: drain + replace
        self._kernel.service_swap(name, new_instance, list(exports), timeout_ms)

        # Register new hooks
        new_spec = hook_spec
        if new_spec is None and _declares_hook_spec(new_instance):
            new_spec = new_instance.hook_spec()
        if new_spec is not None and not new_spec.is_empty:
            self._hook_specs[name] = new_spec
        elif name in self._hook_specs:
            del self._hook_specs[name]
        _register_hooks_for_spec(self, self._hook_specs.get(name))

    @property
    def namespace_manager(self) -> Any | None:
        """Public accessor for the NamespaceManager (via ServiceRegistry)."""
        _pe = self.service("permission_enforcer")
        if _pe is not None:
            return getattr(_pe, "namespace_manager", None)
        return None

    @property
    def config(self) -> Any | None:
        """Public accessor for the runtime configuration object."""
        return self._config

    # _resolve_cred, _build_rust_ctx, _get_context_identity, _validate_path,
    # _parse_context, _ensure_context_ttl,
    # _dispatch_write_events — all moved to InternalMixin (nexus_fs_internal.py)

    # @rpc_expose removed — kernel syscall, served by the thin dispatcher
    # in `nexus.server._kernel_syscall_dispatch`.
    def sys_lock(
        self,
        path: str,
        mode: str = "exclusive",
        ttl: float = 30.0,
        max_holders: int = 1,
        lock_id: str | None = None,
        *,
        context: OperationContext | None = None,
    ) -> str | None:
        """Acquire or extend advisory lock (POSIX fcntl(F_SETLK)).

        When lock_id is None: try-acquire a new lock.
        When lock_id is provided: extend TTL of an existing lock (heartbeat).

        Returns lock_id on success, None on failure.
        """
        path = self._validate_path(path)
        self._enforce_lock_write_permission(path, context)
        return self._kernel.sys_lock(
            path,
            lock_id=lock_id or "",
            mode=mode,
            max_holders=max_holders,
            ttl_secs=int(ttl),
        )

    # @rpc_expose removed — kernel syscall, served by the thin dispatcher
    # in `nexus.server._kernel_syscall_dispatch`.
    def sys_unlock(
        self,
        path: str,
        lock_id: str | None = None,
        force: bool = False,
        *,
        context: OperationContext | None = None,
    ) -> bool:
        """Release advisory lock.

        When force=False (default): release lock by lock_id (requires lock_id).
        When force=True: force-release ALL holders (admin operation, ignores lock_id).

        Returns True if released.
        """
        path = self._validate_path(path)
        if not force and not lock_id:
            raise ValueError("lock_id is required for non-force release")
        self._enforce_lock_write_permission(path, context)
        return self._kernel.sys_unlock(path, lock_id=lock_id or "", force=force)

    def _enforce_lock_write_permission(
        self,
        path: str,
        context: OperationContext | None,
    ) -> None:
        """Require WRITE on ``path`` before touching the Rust lock manager.

        Issue #3786 / Codex Round 6 finding #2: NexusFS.sys_lock previously
        ignored its ``context`` kwarg and the RPC-edge handler enforced WRITE.
        With the legacy handler chain deleted, the enforcement gate now lives
        on the NexusFS method directly so any caller (RPC or in-process)
        gets the policy.  System contexts (server-internal callers) skip.
        """
        if context is None:
            return
        if getattr(context, "is_system", False):
            return
        enforcer = self.service("permission_enforcer") if hasattr(self, "service") else None
        if enforcer is None:
            return

        from nexus.contracts.exceptions import PermissionDeniedError
        from nexus.contracts.types import Permission

        allowed = enforcer.check(path, Permission.WRITE, context)
        if not allowed:
            raise PermissionDeniedError(
                f"lock denied: WRITE not granted for {path!r}",
                path=path,
            )

    def _acquire_lock_sync(
        self,
        path: str,
        timeout: float,
        context: OperationContext | None = None,  # noqa: ARG002
    ) -> str | None:
        """Acquire advisory lock synchronously via kernel sys_lock."""
        from nexus.contracts.exceptions import LockTimeout

        lock_id = self.sys_lock(path, ttl=timeout)
        if lock_id is None:
            raise LockTimeout(path=path, timeout=timeout)
        return lock_id

    def _release_lock_sync(
        self,
        lock_id: str,
        path: str,
        context: OperationContext | None = None,  # noqa: ARG002
    ) -> None:
        """Release advisory lock synchronously via kernel sys_unlock."""
        if not lock_id:
            return
        try:
            self.sys_unlock(path, lock_id=lock_id)
        except Exception as e:
            logger.error(f"Failed to release lock {lock_id} for {path}: {e}")

    # sys_watch is in nexus_fs_watch.py (WatchMixin)

    def workspace_snapshot(
        self,
        workspace_path: str | None = None,
        description: str | None = None,
        tags: builtins.list[str] | None = None,
    ) -> dict[str, Any]:
        return self._workspace_rpc_service.workspace_snapshot(
            workspace_path=workspace_path,
            description=description,
            tags=tags,
        )

    def workspace_restore(
        self,
        snapshot_number: int,
        workspace_path: str | None = None,
    ) -> dict[str, Any]:
        return self._workspace_rpc_service.workspace_restore(
            snapshot_number=snapshot_number,
            workspace_path=workspace_path,
        )

    def workspace_log(
        self,
        workspace_path: str | None = None,
        limit: int = 100,
    ) -> builtins.list[dict[str, Any]]:
        return self._workspace_rpc_service.workspace_log(
            workspace_path=workspace_path,
            limit=limit,
        )

    def workspace_diff(
        self,
        snapshot_1: int,
        snapshot_2: int,
        workspace_path: str | None = None,
    ) -> dict[str, Any]:
        return self._workspace_rpc_service.workspace_diff(
            snapshot_1=snapshot_1,
            snapshot_2=snapshot_2,
            workspace_path=workspace_path,
        )

    # --- Workspace Registry (→ _workspace_rpc_service) ---

    def register_workspace(
        self,
        path: str,
        name: str | None = None,
        description: str | None = None,
        created_by: str | None = None,
        tags: builtins.list[str] | None = None,
        metadata: dict[str, Any] | None = None,
        session_id: str | None = None,
        ttl: Any | None = None,
    ) -> dict[str, Any]:
        return self._workspace_rpc_service.register_workspace(
            path=path,
            name=name,
            description=description,
            created_by=created_by,
            tags=tags,
            metadata=metadata,
            session_id=session_id,
            ttl=ttl,
        )

    def unregister_workspace(self, path: str) -> bool:
        return self._workspace_rpc_service.unregister_workspace(path=path)

    def list_workspaces(self, context: Any | None = None) -> builtins.list[dict]:
        return self._workspace_rpc_service.list_workspaces(context=context)

    def get_workspace_info(self, path: str) -> dict | None:
        return self._workspace_rpc_service.get_workspace_info(path=path)

    # --- Sandbox Operations (→ _sandbox_rpc_service) ---

    def sandbox_create(
        self,
        name: str,
        ttl_minutes: int = 10,
        provider: str | None = "e2b",
        template_id: str | None = None,
        context: dict | None = None,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_create(
            name=name,
            ttl_minutes=ttl_minutes,
            provider=provider,
            template_id=template_id,
            context=context,
        )

    def sandbox_get_or_create(
        self,
        name: str,
        ttl_minutes: int = 10,
        provider: str | None = None,
        template_id: str | None = None,
        verify_status: bool = True,
        context: dict | None = None,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_get_or_create(
            name=name,
            ttl_minutes=ttl_minutes,
            provider=provider,
            template_id=template_id,
            verify_status=verify_status,
            context=context,
        )

    def sandbox_run(
        self,
        sandbox_id: str,
        language: str,
        code: str,
        timeout: int = 300,
        nexus_url: str | None = None,
        nexus_api_key: str | None = None,
        context: dict | None = None,
        as_script: bool = False,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_run(
            sandbox_id=sandbox_id,
            language=language,
            code=code,
            timeout=timeout,
            nexus_url=nexus_url,
            nexus_api_key=nexus_api_key,
            context=context,
            as_script=as_script,
        )

    def sandbox_pause(self, sandbox_id: str, context: dict | None = None) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_pause(sandbox_id=sandbox_id, context=context)

    def sandbox_resume(self, sandbox_id: str, context: dict | None = None) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_resume(sandbox_id=sandbox_id, context=context)

    def sandbox_stop(self, sandbox_id: str, context: dict | None = None) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_stop(sandbox_id=sandbox_id, context=context)

    def sandbox_list(
        self,
        context: dict | None = None,
        verify_status: bool = False,
        user_id: str | None = None,
        zone_id: str | None = None,
        agent_id: str | None = None,
        status: str | None = None,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_list(
            context=context,
            verify_status=verify_status,
            user_id=user_id,
            zone_id=zone_id,
            agent_id=agent_id,
            status=status,
        )

    def sandbox_status(self, sandbox_id: str, context: dict | None = None) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_status(sandbox_id=sandbox_id, context=context)

    def sandbox_connect(
        self,
        sandbox_id: str,
        provider: str = "e2b",
        sandbox_api_key: str | None = None,
        mount_path: str = "/mnt/nexus",
        nexus_url: str | None = None,
        nexus_api_key: str | None = None,
        agent_id: str | None = None,
        context: dict | None = None,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_connect(
            sandbox_id=sandbox_id,
            provider=provider,
            sandbox_api_key=sandbox_api_key,
            mount_path=mount_path,
            nexus_url=nexus_url,
            nexus_api_key=nexus_api_key,
            agent_id=agent_id,
            context=context,
        )

    def sandbox_disconnect(
        self,
        sandbox_id: str,
        provider: str = "e2b",
        sandbox_api_key: str | None = None,
        context: dict | None = None,
    ) -> dict[Any, Any]:
        return self._sandbox_rpc_service.sandbox_disconnect(
            sandbox_id=sandbox_id,
            provider=provider,
            sandbox_api_key=sandbox_api_key,
            context=context,
        )

    # --- Mount Operations (→ mount_service sync accessors) ---

    def add_mount(
        self,
        mount_point: str,
        backend_type: str,
        backend_config: dict[str, Any],
        context: Any = None,
    ) -> str:
        return self.mount_service.add_mount_sync(
            mount_point=mount_point,
            backend_type=backend_type,
            backend_config=backend_config,
            context=context,
        )

    def remove_mount(self, mount_point: str, context: Any = None) -> dict[str, Any]:
        return self.mount_service.remove_mount_sync(mount_point=mount_point, context=context)

    def list_mounts(self, context: Any = None) -> builtins.list[dict[str, Any]]:
        return self.mount_service.list_mounts_sync(context=context)

    def get_mount(self, mount_point: str, context: Any = None) -> dict[str, Any] | None:
        return self.mount_service.get_mount_sync(mount_point=mount_point, context=context)

    def aclose(self) -> None:
        """Shutdown: stop BackgroundService + unregister hooks, then close.

        Calls kernel lifecycle methods first, then
        delegates to close() for sync resource cleanup.

        Sync rather than async because ``kernel.service_stop_all`` invokes
        ``asyncio.run()`` for Python BackgroundService coroutines, which
        cannot run on an already-active loop. The FastAPI lifespan dispatches
        this via ``run_in_executor`` for that reason. To dispose async DB
        engines on the lifespan loop (Issue #3775), call
        :meth:`adispose_async_engines` *before* dispatching ``aclose`` to a
        worker thread.
        """
        # Issue #3391: drain deferred OBSERVE background tasks before tearing down.
        self.shutdown()

        if self._kernel is not None:
            self._kernel.service_stop_all(10000)
            # Unregister all hooks
            for name in list(self._hook_specs):
                spec = self._hook_specs.get(name)
                if spec is not None:
                    _unregister_hooks_for_spec(self, spec)
            self._hook_specs.clear()
        self.close()

    async def adispose_async_engines(self) -> None:
        """Dispose record store's async DB engines on the current loop (Issue #3775).

        Async engines hold asyncpg connections whose pending futures are
        bound to the loop that created the pool. Disposing via
        ``sync_engine.dispose()`` from a worker thread (where ``close()`` runs)
        await-only's into a different loop, producing
        ``RuntimeError: Future attached to a different loop`` and leaks
        connections (asyncpg ``InternalClientError``).

        Disposes **only** the async engines. The record store stays attached
        and its sync engine usable so that ``NexusFS.close()`` callbacks
        (e.g. write observer ``flush_sync``) can still write through the sync
        ``session_factory``. Sync engine teardown happens later in
        ``record_store.close()`` *after* those callbacks have run.

        FastAPI lifespan must await this on its own loop before dispatching
        ``aclose`` to a worker thread. Failure preserves the store and is
        logged at warning level; the subsequent sync ``close()`` will still
        run callbacks and attempt best-effort sync dispose of the (still
        live) async engine — better than silently orphaning the store.
        """
        if self._record_store is None:
            return
        aclose_fn = getattr(self._record_store, "aclose", None)
        if aclose_fn is None:
            return  # legacy store without aclose — close() will handle sync dispose
        try:
            await aclose_fn()
        except Exception:
            logger.warning(
                "record_store.aclose failed; sync close() will attempt fallback",
                exc_info=True,
            )

    # ── IPC primitives (inlined from IPCMixin) ─────────────────────────

    def _pipe_destroy(self, path: str) -> dict[str, Any]:
        """Destroy DT_PIPE — close Rust buffer."""

        with contextlib.suppress(Exception):
            self._kernel.destroy_pipe(path)
        return {}

    # ------------------------------------------------------------------
    # Tier 2 public sync pipe methods (kernel passthroughs)
    # ------------------------------------------------------------------
    # These are sync because the underlying Rust kernel calls are sync.
    # They exist so callers don't need to reach into ``self._kernel`` —
    # convenience wrappers, not first-class syscalls (no Tier 1 ``sys_*``
    # name). Used by coalescing consumers (audit drain, dedup work queue)
    # and sync teardown contexts (AcpService) where async wrapping would
    # add event-loop ping-pong without buying anything.

    def pipe_create(self, path: str, capacity: int = 65_536) -> None:
        """Create a DT_PIPE in the kernel registry.

        Sync passthrough to ``Kernel.create_pipe``.
        """
        self._kernel.create_pipe(path, capacity)

    def pipe_close(self, path: str) -> None:
        """Mark a DT_PIPE as closed (signals EOF to readers, keeps registry entry)."""
        self._kernel.close_pipe(path)

    def has_pipe(self, path: str) -> bool:
        """Check if a DT_PIPE exists in the kernel registry."""
        if self._kernel is None:
            return False
        return self._kernel.has_pipe(path)

    # ------------------------------------------------------------------
    # Tier 2 public sync stream methods (kernel passthroughs)
    # ------------------------------------------------------------------
    # Stream counterparts to the pipe convenience methods above. Used by
    # LLM streaming backends (Rust OpenAIBackend / AnthropicBackend via
    # nx.llm_start_streaming) where a tight token-pump loop calls
    # ``stream_write_nowait`` per token and ``stream_read_at`` for
    # offset-based replay — async wrapping would just add ping-pong.

    def stream_create(self, path: str, capacity: int = 65_536) -> None:
        """Create a DT_STREAM in the kernel registry."""
        self._kernel.create_stream(path, capacity)

    def has_stream(self, path: str) -> bool:
        """Check if a DT_STREAM exists in the kernel registry."""
        if self._kernel is None:
            return False
        return self._kernel.has_stream(path)

    def stream_read_at_blocking(self, path: str, offset: int, timeout_ms: int) -> tuple[bytes, int]:
        """Blocking offset-based stream read. Returns (chunk, next_offset).

        Releases the GIL inside Rust during the wait. Callers that need
        async semantics should wrap in ``asyncio.to_thread``.
        """
        _data, _next = self._kernel.stream_read_at_blocking(path, offset, timeout_ms)
        return (bytes(_data), _next)

    def stream_write_nowait(self, path: str, data: bytes) -> int:
        """Non-blocking stream append. Returns byte offset."""
        return self._kernel.stream_write_nowait(path, data)

    def stream_read_at(self, path: str, offset: int) -> tuple[bytes, int] | None:
        """Non-blocking offset-based stream read. Returns (chunk, next_offset) or None."""
        _result = self._kernel.stream_read_at(path, offset)
        if _result is None:
            return None
        return (bytes(_result[0]), _result[1])

    def stream_collect_all(self, path: str) -> bytes:
        """Collect all message payloads from a DT_STREAM, concatenated.

        Single Rust call — no per-frame PyO3 round-trip. Replaces
        manual ``read_at`` loops in LLM backends.
        """
        return bytes(self._kernel.stream_collect_all(path))

    def stream_close(self, path: str) -> None:
        """Mark a DT_STREAM as closed (signals EOF to readers)."""
        self._kernel.close_stream(path)

    def stream_destroy(self, path: str) -> None:
        """Destroy a DT_STREAM — close kernel buffer + custom backend cleanup.

        Sync alternative to ``await sys_unlink(path)``.
        """
        self._stream_destroy(path)

    def _stream_read(self, path: str, *, count: int | None = None, offset: int = 0) -> bytes:
        """Read from DT_STREAM — nowait hot path + Rust blocking slow path (GIL-free)."""
        # Hot path: try nowait first (zero GIL)
        _result = self._kernel.stream_read_at(path, offset)
        if _result is not None:
            return bytes(_result[0])

        # Slow path: block in Rust, release GIL
        _data, _next = self._kernel.stream_read_at_blocking(path, offset, 30000)
        return bytes(_data)

    def _stream_write(self, path: str, data: bytes) -> int:
        """Write to DT_STREAM — non-blocking via Rust kernel (condvar wakes readers), returns byte offset."""
        return self._kernel.stream_write_nowait(path, data)

    def _stream_destroy(self, path: str) -> dict[str, Any]:
        """Destroy DT_STREAM — close Rust buffer."""

        with contextlib.suppress(Exception):
            self._kernel.destroy_stream(path)
        return {}

    def close(self) -> None:
        """Close the filesystem and release resources."""
        flush_error: Exception | None = None

        def _after_flush_cleanup(label: str, close_fn: Callable[[], None]) -> None:
            if flush_error is None:
                close_fn()
                return
            try:
                close_fn()
            except Exception as exc:
                logger.debug("close: %s failed after write buffer flush failure: %s", label, exc)

        # Issue #1793/#1789/#1792: Service close via factory-registered callbacks.
        # Runs BEFORE pipe/IPC close so callbacks can drain pipe buffers
        # (Issue #3399: piped write observer needs to flush before buffers clear).
        for _close_cb in self._close_callbacks:
            try:
                _close_cb()
            except Exception as exc:
                logger.debug("close: callback failed (best-effort): %s", exc)

        # Auto-close all enlisted services that have a close() method
        # (rebac_manager, audit_store, etc.). Reverse registration order.
        if self._kernel is not None:
            self._kernel.service_close_all()

        # Final durability barrier: callbacks/services may write during close.
        if self._kernel is not None:
            try:
                self._kernel.flush_write_buffer(None, None)
            except Exception as exc:
                flush_error = exc
                logger.error("close: write buffer flush failed: %s", exc, exc_info=True)

        # Close IPC primitives — Rust kernel (§4.2)
        # _kernel is None in remote connection mode (no local kernel)
        if self._kernel is not None:
            _after_flush_cleanup("close_all_pipes", self._kernel.close_all_pipes)
            _after_flush_cleanup("close_all_streams", self._kernel.close_all_streams)
        # Close transport pool (persistent gRPC connections)
        if hasattr(self, "_transport_pool") and self._transport_pool is not None:
            _after_flush_cleanup("transport_pool.close_all", self._transport_pool.close_all)

        # Metadata store close is a no-op — kernel manages the redb
        # lifecycle via ``release_metastores`` below.

        # Release Rust-owned redb/SQLite file handles. Without this call the
        # Rust kernel keeps the metastore Box alive until Python GC runs —
        # process-lifetime tests that open the same redb path in a second
        # NexusFS hit ``Database already open. Cannot acquire lock.`` (Issue
        # #3765 Cat-5/6). ``release_metastores`` is idempotent.
        if self._kernel is not None and flush_error is None:
            try:
                _release = getattr(self._kernel, "release_metastores", None)
                if _release is not None:
                    _release()
            except Exception as exc:  # pragma: no cover - best-effort teardown
                logger.debug("kernel.release_metastores failed: %s", exc)
            # Drop this kernel from the shared create_kernel cache so the
            # next ``create_kernel(path)`` in this process gets a fresh
            # kernel with its own metastore wired up (Issue #3765 Cat-5/6).
            try:
                from nexus.fs._kernel_factory import _evict_kernel_cache

                _evict_kernel_cache(self._kernel)
            except Exception as exc:  # pragma: no cover - best-effort
                logger.debug("_evict_kernel_cache failed: %s", exc)

        # Close record store (Services layer SQL connections)
        if self._record_store is not None:
            _after_flush_cleanup("record_store.close", self._record_store.close)

        # Close process-local runtime resources owned by this NexusFS.
        while self._runtime_closeables:
            resource = self._runtime_closeables.pop()
            close_fn = getattr(resource, "close", None)
            if not callable(close_fn):
                continue
            try:
                close_fn()
            except Exception as e:
                logger.debug("Failed to close runtime resource %s: %s", type(resource).__name__, e)

        if flush_error is not None:
            raise flush_error

    def __enter__(self) -> "NexusFS":
        """Context manager entry."""
        return self

    def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
        """Context manager exit."""
        self.close()

    # ── Tier 2: Lock Convenience (moved from ABC) ────────────────

    def lock(
        self,
        path: str,
        mode: str = "exclusive",
        timeout: float = 30.0,
        ttl: float = 60.0,
        max_holders: int = 1,
        *,
        context: "OperationContext | None" = None,
    ) -> str | None:
        """Acquire lock with blocking wait (Tier 2 over sys_lock).

        Retries sys_lock() until acquired or timeout.
        Like fcntl(F_SETLKW) — blocking variant of sys_lock (F_SETLK).
        """
        import time as _time

        deadline = _time.monotonic() + timeout
        while True:
            lock_id = self.sys_lock(
                path,
                mode=mode,
                ttl=ttl,
                max_holders=max_holders,
                context=context,
            )
            if lock_id is not None:
                return lock_id
            remaining = deadline - _time.monotonic()
            if remaining <= 0:
                return None
            _time.sleep(min(0.05, remaining))

    def unlock(self, lock_id: str, path: str, *, context: "OperationContext | None" = None) -> bool:
        """Release lock (Tier 2 alias for sys_unlock)."""
        return self.sys_unlock(path, lock_id, context=context)

    # ── Tier 2: glob/grep (moved from ABC) ────────────────────────

    def glob(self, pattern: str, path: str = "/", context: Any = None) -> builtins.list[str]:
        """Find files matching a glob pattern (like glob(3)).

        Requires SearchService.
        """
        raise NotImplementedError("glob requires SearchService")

    def grep(
        self,
        pattern: str,
        path: str = "/",
        file_pattern: str | None = None,
        ignore_case: bool = False,
        max_results: int = 1000,
        search_mode: str = "auto",
        context: Any = None,
        before_context: int = 0,
        after_context: int = 0,
        invert_match: bool = False,
    ) -> builtins.list[dict[str, Any]]:
        """Search file contents using regex patterns (like grep(1)).

        Requires SearchService.
        """
        raise NotImplementedError("grep requires SearchService")
