// AUTO-GENERATED by scripts/codegen_kernel_abi.py — DO NOT EDIT
//! PyO3 transport wrapper — bridges Python <-> pure Rust kernel.
//!
//! This single file contains ALL PyO3 code for the kernel boundary:
//!   - Direction 1 (WRAPPER): PyKernel wraps pure Rust Kernel
//!   - Direction 2 (DISPATCH): Python hooks/resolvers/observers -> Rust traits
//!   - Direction 3 (PILLAR): Python storage backends -> Rust trait implementations
//!   - Error conversion: KernelError -> PyErr
//!   - FileMetadata conversion helpers
//!
//! Re-generate: python scripts/codegen_kernel_abi.py

#![allow(dead_code)]

use parking_lot::RwLock;
use pyo3::prelude::*;
// Brings ``read_unconditional`` / ``read_yielding_to_writer`` into scope
// — see ``RwLockExt`` in ``kernel.rs`` for the rationale.
use crate::kernel::RwLockExt;
use pyo3::types::{PyBytes, PyDict, PyList};
use std::sync::Arc;

#[cfg(feature = "py-hook-adapters")]
use crate::dispatch::PathResolver;
use crate::hook_registry::HookRegistry;
#[cfg(feature = "py-hook-adapters")]
use crate::hook_registry::InterceptHook;
use crate::kernel::{Kernel, KernelError, OperationContext, WriteBufferFlushHandle};
use crate::meta_store::{FileMetadata, MetaStoreError};
use crate::vfs_router::RouteError;

// ═══════════════════════════════════════════════════════════════════════════
// PyServiceLifecycle — Python → ServiceLifecycle adapter
// ═══════════════════════════════════════════════════════════════════════════

use crate::service_registry::ServiceLifecycle;

/// Wraps a `Py<PyAny>` Python service instance into the language-agnostic
/// `ServiceLifecycle` trait. All Python asyncio / BackgroundService
/// logic stays here in the cdylib layer — the kernel never touches PyO3
/// for service management.
struct PyServiceLifecycle(Py<PyAny>);

/// Run a Python coroutine to completion via stdlib asyncio.
fn run_coro(py: Python<'_>, coro: &Bound<'_, PyAny>, timeout_secs: f64) -> PyResult<()> {
    let asyncio = py.import("asyncio")?;
    let timed = asyncio.call_method1("wait_for", (coro, timeout_secs))?;
    asyncio.call_method1("run", (&timed,))?;
    Ok(())
}

impl ServiceLifecycle for PyServiceLifecycle {
    fn start(&self, timeout_secs: f64) -> Result<(), String> {
        Python::attach(|py| {
            let instance = self.0.bind(py);
            let bg_cls = py
                .import("nexus.contracts.protocols.service_lifecycle")
                .and_then(|m| m.getattr("BackgroundService"))
                .map_err(|e| format!("{e}"))?;
            if !instance.is_instance(&bg_cls).map_err(|e| format!("{e}"))? {
                return Ok(());
            }
            let coro = instance.call_method0("start").map_err(|e| format!("{e}"))?;
            run_coro(py, &coro, timeout_secs).map_err(|e| format!("{e}"))
        })
    }

    fn stop(&self, timeout_secs: f64) -> Result<(), String> {
        Python::attach(|py| {
            let instance = self.0.bind(py);
            let bg_cls = py
                .import("nexus.contracts.protocols.service_lifecycle")
                .and_then(|m| m.getattr("BackgroundService"))
                .map_err(|e| format!("{e}"))?;
            if !instance.is_instance(&bg_cls).map_err(|e| format!("{e}"))? {
                return Ok(());
            }
            let coro = instance.call_method0("stop").map_err(|e| format!("{e}"))?;
            run_coro(py, &coro, timeout_secs).map_err(|e| format!("{e}"))
        })
    }

    fn close(&self) -> Result<(), String> {
        Python::attach(|py| {
            let instance = self.0.bind(py);
            if let Ok(close_fn) = instance.getattr("close") {
                if close_fn.is_callable() {
                    close_fn.call0().map_err(|e| format!("{e}"))?;
                }
            }
            Ok(())
        })
    }

    fn type_name(&self) -> String {
        Python::attach(|py| {
            self.0
                .bind(py)
                .get_type()
                .name()
                .map(|n| n.to_string())
                .unwrap_or_else(|_| "?".to_string())
        })
    }

    fn clone_box(&self) -> Box<dyn ServiceLifecycle> {
        Python::attach(|py| Box::new(PyServiceLifecycle(self.0.clone_ref(py))))
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Error conversion: KernelError -> PyErr (cached exception classes)
// ═══════════════════════════════════════════════════════════════════════════

/// Cached exception class references — initialized once on first use.
struct ExceptionCache {
    invalid_path: Py<PyAny>,
    file_not_found: Py<PyAny>,
    backend_error: Py<PyAny>,
    permission_denied: Py<PyAny>,
}

static EXCEPTION_CACHE: std::sync::OnceLock<Option<ExceptionCache>> = std::sync::OnceLock::new();

fn get_exception_cache(py: Python<'_>) -> Option<&'static ExceptionCache> {
    EXCEPTION_CACHE
        .get_or_init(|| {
            let m = py.import("nexus.contracts.exceptions").ok()?;
            let invalid_path = m.getattr("InvalidPathError").ok()?.unbind();
            let file_not_found = m.getattr("NexusFileNotFoundError").ok()?.unbind();
            let backend_error = m.getattr("BackendError").ok()?.unbind();
            let permission_denied = m.getattr("PermissionDeniedError").ok()?.unbind();
            Some(ExceptionCache {
                invalid_path,
                file_not_found,
                backend_error,
                permission_denied,
            })
        })
        .as_ref()
}

impl From<KernelError> for PyErr {
    fn from(e: KernelError) -> PyErr {
        match e {
            KernelError::InvalidPath(msg) => Python::attach(|py| {
                if let Some(cache) = get_exception_cache(py) {
                    cache
                        .invalid_path
                        .bind(py)
                        .call1((&msg,))
                        .map(PyErr::from_value)
                        .unwrap_or_else(|_| pyo3::exceptions::PyValueError::new_err(msg))
                } else {
                    pyo3::exceptions::PyValueError::new_err(msg)
                }
            }),
            KernelError::FileNotFound(path) => Python::attach(|py| {
                if let Some(cache) = get_exception_cache(py) {
                    cache
                        .file_not_found
                        .bind(py)
                        .call1((&path,))
                        .map(PyErr::from_value)
                        .unwrap_or_else(|_| pyo3::exceptions::PyFileNotFoundError::new_err(path))
                } else {
                    pyo3::exceptions::PyFileNotFoundError::new_err(path)
                }
            }),
            KernelError::FileExists(msg) => pyo3::exceptions::PyFileExistsError::new_err(msg),
            KernelError::Route(RouteError::NotMounted(msg)) => {
                pyo3::exceptions::PyValueError::new_err(msg)
            }
            KernelError::IOError(msg) => pyo3::exceptions::PyIOError::new_err(msg),
            KernelError::TrieError(msg) => pyo3::exceptions::PyValueError::new_err(msg),
            // IPC error variants
            KernelError::PipeFull(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("PipeFull:{msg}"))
            }
            KernelError::PipeEmpty(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("PipeEmpty:{msg}"))
            }
            KernelError::PipeClosed(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("PipeClosed:{msg}"))
            }
            KernelError::PipeExists(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("PipeExists:{msg}"))
            }
            KernelError::PipeNotFound(path) => Python::attach(|py| {
                if let Some(cache) = get_exception_cache(py) {
                    cache
                        .file_not_found
                        .bind(py)
                        .call1((&path,))
                        .map(PyErr::from_value)
                        .unwrap_or_else(|_| pyo3::exceptions::PyFileNotFoundError::new_err(path))
                } else {
                    pyo3::exceptions::PyFileNotFoundError::new_err(path)
                }
            }),
            KernelError::StreamFull(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("StreamFull:{msg}"))
            }
            KernelError::StreamEmpty(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("StreamEmpty:{msg}"))
            }
            KernelError::StreamClosed(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("StreamClosed:{msg}"))
            }
            KernelError::StreamExists(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("StreamExists:{msg}"))
            }
            KernelError::StreamNotFound(path) => Python::attach(|py| {
                if let Some(cache) = get_exception_cache(py) {
                    cache
                        .file_not_found
                        .bind(py)
                        .call1((&path,))
                        .map(PyErr::from_value)
                        .unwrap_or_else(|_| pyo3::exceptions::PyFileNotFoundError::new_err(path))
                } else {
                    pyo3::exceptions::PyFileNotFoundError::new_err(path)
                }
            }),
            KernelError::WouldBlock(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("WouldBlock:{msg}"))
            }
            KernelError::PermissionDenied(msg) => pyo3::exceptions::PyPermissionError::new_err(msg),
            KernelError::BackendError(msg) => Python::attach(|py| {
                if let Some(cache) = get_exception_cache(py) {
                    cache
                        .backend_error
                        .bind(py)
                        .call1((&msg,))
                        .map(PyErr::from_value)
                        .unwrap_or_else(|_| pyo3::exceptions::PyIOError::new_err(msg))
                } else {
                    pyo3::exceptions::PyIOError::new_err(msg)
                }
            }),
            KernelError::Federation(msg) => {
                pyo3::exceptions::PyRuntimeError::new_err(format!("Federation:{msg}"))
            }
        }
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Hook context class cache — OnceLock avoids py.import() per syscall (~1μs)
// ═══════════════════════════════════════════════════════════════════════════

/// Cached hook context class references — initialized once on first hooked syscall.
struct HookContextCache {
    read: Py<PyAny>,
    write: Py<PyAny>,
    delete: Py<PyAny>,
    rename: Py<PyAny>,
    mkdir: Py<PyAny>,
    rmdir: Py<PyAny>,
    copy: Py<PyAny>,
    stat: Py<PyAny>,
    access: Py<PyAny>,
}

static HOOK_CTX_CACHE: std::sync::OnceLock<Option<HookContextCache>> = std::sync::OnceLock::new();

fn get_hook_ctx_cache(py: Python<'_>) -> Option<&'static HookContextCache> {
    HOOK_CTX_CACHE
        .get_or_init(|| {
            let m = py.import("nexus.contracts.vfs_hooks").ok()?;
            Some(HookContextCache {
                read: m.getattr("ReadHookContext").ok()?.unbind(),
                write: m.getattr("WriteHookContext").ok()?.unbind(),
                delete: m.getattr("DeleteHookContext").ok()?.unbind(),
                rename: m.getattr("RenameHookContext").ok()?.unbind(),
                mkdir: m.getattr("MkdirHookContext").ok()?.unbind(),
                rmdir: m.getattr("RmdirHookContext").ok()?.unbind(),
                copy: m.getattr("CopyHookContext").ok()?.unbind(),
                stat: m.getattr("StatHookContext").ok()?.unbind(),
                access: m.getattr("AccessHookContext").ok()?.unbind(),
            })
        })
        .as_ref()
}

// ═══════════════════════════════════════════════════════════════════════════
// FileMetadata conversion helpers (from generated_store.rs)
// ═══════════════════════════════════════════════════════════════════════════

/// Extract Rust FileMetadata from a Python FileMetadata object.
fn extract_metadata(
    py: Python<'_>,
    obj: &Bound<'_, PyAny>,
) -> Result<FileMetadata, MetaStoreError> {
    let get_str = |name: &str| -> Result<String, MetaStoreError> {
        obj.getattr(name)
            .and_then(|v| v.extract::<String>())
            .map_err(|e| MetaStoreError::IOError(format!("field {name}: {e}")))
    };
    let get_opt_str = |name: &str| -> Result<Option<String>, MetaStoreError> {
        match obj.getattr(name) {
            Ok(v) if v.is_none() => Ok(None),
            Ok(v) => v
                .extract::<String>()
                .map(Some)
                .map_err(|e| MetaStoreError::IOError(format!("field {name}: {e}"))),
            Err(e) => Err(MetaStoreError::IOError(format!("field {name}: {e}"))),
        }
    };
    let get_u64_or_zero = |name: &str| -> Result<u64, MetaStoreError> {
        match obj.getattr(name) {
            Ok(v) => v
                .extract::<u64>()
                .map_err(|e| MetaStoreError::IOError(format!("field {name}: {e}"))),
            Err(_) => Ok(0),
        }
    };

    let _ = py;
    Ok(FileMetadata {
        path: get_str("path")?,
        size: obj
            .getattr("size")
            .and_then(|v| v.extract::<u64>())
            .map_err(|e| MetaStoreError::IOError(format!("field size: {e}")))?,
        content_id: get_opt_str("content_id")?,
        gen: get_u64_or_zero("gen")?,
        version: obj
            .getattr("version")
            .and_then(|v| v.extract::<u32>())
            .map_err(|e| MetaStoreError::IOError(format!("field version: {e}")))?,
        entry_type: obj
            .getattr("entry_type")
            .and_then(|v| v.extract::<u8>())
            .map_err(|e| MetaStoreError::IOError(format!("field entry_type: {e}")))?,
        zone_id: get_opt_str("zone_id")?,
        mime_type: get_opt_str("mime_type")?,
        created_at_ms: extract_opt_datetime_ms(obj, "created_at"),
        modified_at_ms: extract_opt_datetime_ms(obj, "modified_at"),
        last_writer_address: get_opt_str("last_writer_address")?,
        target_zone_id: get_opt_str("target_zone_id")?,
        link_target: get_opt_str("link_target").ok().flatten(),
        owner_id: get_opt_str("owner_id").ok().flatten(),
    })
}

/// Convert a Python ``datetime`` attribute to epoch milliseconds.
/// Returns None if the attribute is missing or not a datetime.
fn extract_opt_datetime_ms(obj: &Bound<'_, PyAny>, name: &str) -> Option<i64> {
    let v = obj.getattr(name).ok()?;
    if v.is_none() {
        return None;
    }
    let ts = v.call_method0("timestamp").ok()?;
    let secs = ts.extract::<f64>().ok()?;
    Some((secs * 1000.0) as i64)
}

/// Convert Rust FileMetadata -> Python FileMetadata (for metastore.put()).
fn to_python_metadata<'py>(
    py: Python<'py>,
    meta: &FileMetadata,
) -> Result<Bound<'py, PyAny>, MetaStoreError> {
    fn err(e: PyErr) -> MetaStoreError {
        MetaStoreError::IOError(format!("to_python_metadata: {e}"))
    }
    let cls = py
        .import("nexus.contracts.metadata")
        .map_err(err)?
        .getattr("FileMetadata")
        .map_err(err)?;
    let kwargs = PyDict::new(py);
    kwargs.set_item("path", &meta.path).map_err(err)?;
    kwargs.set_item("size", meta.size).map_err(err)?;
    kwargs
        .set_item("content_id", meta.content_id.as_deref())
        .map_err(err)?;
    kwargs.set_item("gen", meta.gen).map_err(err)?;
    kwargs.set_item("version", meta.version).map_err(err)?;
    kwargs
        .set_item("entry_type", meta.entry_type)
        .map_err(err)?;
    kwargs
        .set_item("zone_id", meta.zone_id.as_deref())
        .map_err(err)?;
    kwargs
        .set_item("mime_type", meta.mime_type.as_deref())
        .map_err(err)?;
    set_optional_datetime(py, &kwargs, "created_at", meta.created_at_ms).map_err(err)?;
    set_optional_datetime(py, &kwargs, "modified_at", meta.modified_at_ms).map_err(err)?;
    if let Some(target) = meta.link_target.as_deref() {
        kwargs.set_item("link_target", target).map_err(err)?;
    }
    if let Some(owner) = meta.owner_id.as_deref() {
        kwargs.set_item("owner_id", owner).map_err(err)?;
    }
    cls.call((), Some(&kwargs)).map_err(err)
}

/// Set a kwargs entry to a UTC datetime built from epoch ms (or None).
fn set_optional_datetime(
    py: Python<'_>,
    kwargs: &Bound<'_, PyDict>,
    key: &str,
    ms: Option<i64>,
) -> PyResult<()> {
    let Some(ms) = ms else {
        return kwargs.set_item(key, py.None());
    };
    let datetime = py.import("datetime")?;
    let utc = datetime.getattr("timezone")?.getattr("utc")?;
    let from_ts = datetime.getattr("datetime")?.getattr("fromtimestamp")?;
    let secs = ms as f64 / 1000.0;
    let dt = from_ts.call((secs, &utc), None)?;
    kwargs.set_item(key, dt)
}

/// Set a stat-dict entry to a UTC ISO-8601 string built from epoch ms.
/// Pure Rust (chrono) — no Python crossing.
fn set_optional_iso_datetime(
    py: Python<'_>,
    dict: &Bound<'_, PyDict>,
    key: &str,
    ms: Option<i64>,
) -> PyResult<()> {
    let Some(ms) = ms else {
        return dict.set_item(key, py.None());
    };
    let secs = ms / 1000;
    let nsecs = ((ms % 1000) * 1_000_000) as u32;
    if let Some(dt) = chrono::DateTime::from_timestamp(secs, nsecs) {
        dict.set_item(key, dt.to_rfc3339_opts(chrono::SecondsFormat::Millis, true))
    } else {
        dict.set_item(key, py.None())
    }
}

/// Convert StatResult → Python dict (same shape as sys_stat output).
fn stat_result_to_pydict<'py>(
    py: Python<'py>,
    s: &crate::kernel::StatResult,
) -> Bound<'py, PyDict> {
    let dict = PyDict::new(py);
    let _ = dict.set_item("path", &s.path);
    let _ = dict.set_item("size", s.size);
    let _ = dict.set_item("last_writer_address", s.last_writer_address.as_deref());
    let _ = dict.set_item("content_id", s.content_id.as_deref());
    let _ = dict.set_item("mime_type", &s.mime_type);
    let _ = set_optional_iso_datetime(py, &dict, "created_at", s.created_at_ms);
    let _ = set_optional_iso_datetime(py, &dict, "modified_at", s.modified_at_ms);
    let _ = dict.set_item("is_directory", s.is_directory);
    let _ = dict.set_item("entry_type", s.entry_type);
    let _ = dict.set_item("mode", s.mode);
    let _ = dict.set_item("version", s.version);
    let _ = dict.set_item("zone_id", s.zone_id.as_deref());
    let _ = dict.set_item("link_target", s.link_target.as_deref());
    let _ = dict.set_item("owner_id", s.owner_id.as_deref());
    let _ = dict.set_item("lock", py.None());
    dict
}

/// Convert Rust OperationContext -> Python OperationContext.
///
/// Forwards ALL fields so Python hooks see the full credential.
fn rust_ctx_to_python<'py>(
    py: Python<'py>,
    ctx: &OperationContext,
    backend_path: &str,
) -> Result<Bound<'py, PyAny>, String> {
    let cls = py
        .import("nexus.contracts.types")
        .and_then(|m| m.getattr("OperationContext"))
        .map_err(|e| format!("import OperationContext: {e}"))?;
    let kwargs = PyDict::new(py);
    let _ = kwargs.set_item("user_id", &ctx.user_id);
    // Use context_zone_id (caller's zone, may be None) for Python context,
    // NOT routing zone_id (always set to NexusFS instance zone).
    match &ctx.context_zone_id {
        Some(z) => {
            let _ = kwargs.set_item("zone_id", z);
        }
        None => {
            let _ = kwargs.set_item("zone_id", py.None());
        }
    }
    let _ = kwargs.set_item("is_admin", ctx.is_admin);
    let _ = kwargs.set_item("is_system", ctx.is_system);
    let _ = kwargs.set_item("backend_path", backend_path);
    let groups = PyList::new(py, &ctx.groups).map_err(|e| format!("groups: {e}"))?;
    let _ = kwargs.set_item("groups", groups);
    if let Some(ref agent_id) = ctx.agent_id {
        let _ = kwargs.set_item("agent_id", agent_id);
    }
    // admin_capabilities: Vec<String> → Python set[str]
    let cap_set = pyo3::types::PySet::empty(py).map_err(|e| format!("set(): {e}"))?;
    for cap in &ctx.admin_capabilities {
        let _ = cap_set.add(cap);
    }
    let _ = kwargs.set_item("admin_capabilities", &cap_set);
    let _ = kwargs.set_item("subject_type", &ctx.subject_type);
    if let Some(ref sid) = ctx.subject_id {
        let _ = kwargs.set_item("subject_id", sid);
    }
    if !ctx.request_id.is_empty() {
        let _ = kwargs.set_item("request_id", &ctx.request_id);
    }
    // zone_perms: Vec<(String, String)> → Python tuple of (zone_id, perm_chars) pairs
    if !ctx.zone_perms.is_empty() {
        let zp = PyList::new(
            py,
            ctx.zone_perms.iter().map(|(z, p)| (z.as_str(), p.as_str())),
        )
        .map_err(|e| format!("zone_perms: {e}"))?;
        let _ = kwargs.set_item("zone_perms", zp);
    }
    cls.call((), Some(&kwargs))
        .map_err(|e| format!("OperationContext(): {e}"))
}

// ═══════════════════════════════════════════════════════════════════════════
// Direction 3 (PILLAR): Store adapters — Python ABC -> Rust trait
// ═══════════════════════════════════════════════════════════════════════════

// ═══════════════════════════════════════════════════════════════════════════
// Direction 2 (DISPATCH): Python hooks/resolvers/observers -> Rust traits
// ═══════════════════════════════════════════════════════════════════════════

// ── PyInterceptHookAdapter ──────────────────────────────────────

/// Wraps Python -> Rust `InterceptHook` trait.
#[cfg(feature = "py-hook-adapters")]
pub(crate) struct PyInterceptHookAdapter {
    inner: Py<PyAny>,
    hook_name: String,
}

#[cfg(feature = "py-hook-adapters")]
unsafe impl Send for PyInterceptHookAdapter {}
#[cfg(feature = "py-hook-adapters")]
unsafe impl Sync for PyInterceptHookAdapter {}

#[cfg(feature = "py-hook-adapters")]
impl PyInterceptHookAdapter {
    pub(crate) fn new(py: Python<'_>, hook: Py<PyAny>) -> Self {
        let name = hook
            .bind(py)
            .getattr("name")
            .and_then(|n| n.extract::<String>())
            .unwrap_or_else(|_| "<hook>".to_string());
        Self {
            inner: hook,
            hook_name: name,
        }
    }
}

#[cfg(feature = "py-hook-adapters")]
impl InterceptHook for PyInterceptHookAdapter {
    fn name(&self) -> &str {
        &self.hook_name
    }

    fn on_pre_read(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_read") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_read(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_read") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_write(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_write") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_write(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_write") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_delete(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_delete") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_delete(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_delete") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_rename(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_rename") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_rename(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_rename") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_mkdir(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_mkdir") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_mkdir(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_mkdir") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_rmdir(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_rmdir") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_rmdir(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_rmdir") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_copy(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_copy") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_copy(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_copy") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_stat(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_stat") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_stat(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_stat") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_access(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_access") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_access(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_access") {
                let _ = method.call1((ctx,));
            }
        });
    }

    fn on_pre_write_batch(&self, ctx: &Py<PyAny>) -> Result<(), PyErr> {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_pre_write_batch") {
                method.call1((ctx,))?;
            }
            Ok(())
        })
    }

    fn on_post_write_batch(&self, ctx: &Py<PyAny>) {
        Python::attach(|py| {
            let hook = self.inner.bind(py);
            if let Ok(method) = hook.getattr("on_post_write_batch") {
                let _ = method.call1((ctx,));
            }
        });
    }
}

// ── PyPathResolverAdapter ───────────────────────────────────────

/// Wraps Python -> Rust `PathResolver` trait.
#[cfg(feature = "py-hook-adapters")]
pub(crate) struct PyPathResolverAdapter {
    inner: Py<PyAny>,
}

#[cfg(feature = "py-hook-adapters")]
unsafe impl Send for PyPathResolverAdapter {}
#[cfg(feature = "py-hook-adapters")]
unsafe impl Sync for PyPathResolverAdapter {}

#[cfg(feature = "py-hook-adapters")]
impl PathResolver for PyPathResolverAdapter {
    fn try_read(&self, path: &str) -> Option<Vec<u8>> {
        Python::attach(|py| {
            let result = self.inner.call_method1(py, "try_read", (path,)).ok()?;
            if result.is_none(py) {
                return None;
            }
            result.extract::<Vec<u8>>(py).ok()
        })
    }

    fn try_write(&self, path: &str, content: &[u8]) -> Option<()> {
        Python::attach(|py| {
            self.inner
                .call_method1(py, "try_write", (path, content))
                .ok()?;
            Some(())
        })
    }

    fn try_delete(&self, path: &str) -> Option<()> {
        Python::attach(|py| {
            self.inner.call_method1(py, "try_delete", (path,)).ok()?;
            Some(())
        })
    }
}

// ── PyPermissionProviderAdapter ──────────────────────────────────────

use crate::core::dispatch::{Permission, PermissionDecision, PermissionProvider};

pub(crate) struct PyPermissionProviderAdapter {
    checker: Py<PyAny>,
}

unsafe impl Send for PyPermissionProviderAdapter {}

unsafe impl Sync for PyPermissionProviderAdapter {}

impl PermissionProvider for PyPermissionProviderAdapter {
    fn check(
        &self,
        path: &str,
        permission: Permission,
        _ctx: &contracts::OperationContext,
    ) -> PermissionDecision {
        Python::attach(|py| {
            let checker = self.checker.bind(py);
            let perm_mod = match py.import("nexus.contracts.types") {
                Ok(m) => m,
                Err(_) => return PermissionDecision::Unknown,
            };
            let perm_cls = match perm_mod.getattr("Permission") {
                Ok(c) => c,
                Err(_) => return PermissionDecision::Unknown,
            };
            let py_perm = match perm_cls.getattr(permission.as_str()) {
                Ok(p) => p,
                Err(_) => return PermissionDecision::Unknown,
            };
            let py_ctx = match rust_ctx_to_python(py, _ctx, "") {
                Ok(c) => c,
                Err(_) => return PermissionDecision::Unknown,
            };
            match checker.call_method1("check", (path, py_perm, py_ctx)) {
                Ok(_) => PermissionDecision::Allow,
                Err(_) => PermissionDecision::Deny(format!(
                    "permission denied: {} on '{}'",
                    permission.as_str(),
                    path
                )),
            }
        })
    }
}

// ═══════════════════════════════════════════════════════════════════════════
// Direction 1 (WRAPPER): PyO3 wrapper types
// ═══════════════════════════════════════════════════════════════════════════

// ── PyOperationContext ──────────────────────────────────────────

/// Python-facing OperationContext (wraps pure Rust OperationContext).
#[pyclass(get_all, from_py_object)]
#[derive(Clone, Debug)]
pub struct PyOperationContext {
    pub user_id: String,
    pub zone_id: String,
    pub is_admin: bool,
    pub agent_id: Option<String>,
    pub is_system: bool,
    pub groups: Vec<String>,
    pub admin_capabilities: Vec<String>,
    pub subject_type: String,
    pub subject_id: Option<String>,
    pub request_id: String,
    pub context_zone_id: Option<String>,
    pub zone_perms: Vec<(String, String)>,
}

#[pymethods]
impl PyOperationContext {
    #[new]
    #[pyo3(signature = (user_id="anonymous", zone_id="root", is_admin=false, agent_id=None, is_system=false, groups=vec![], admin_capabilities=vec![], subject_type="user", subject_id=None, request_id="", context_zone_id=None, zone_perms=vec![]))]
    #[allow(clippy::too_many_arguments)]
    fn new(
        user_id: &str,
        zone_id: &str,
        is_admin: bool,
        agent_id: Option<&str>,
        is_system: bool,
        groups: Vec<String>,
        admin_capabilities: Vec<String>,
        subject_type: &str,
        subject_id: Option<&str>,
        request_id: &str,
        context_zone_id: Option<&str>,
        zone_perms: Vec<(String, String)>,
    ) -> Self {
        Self {
            user_id: user_id.to_string(),
            zone_id: zone_id.to_string(),
            is_admin,
            agent_id: agent_id.map(|s| s.to_string()),
            is_system,
            groups,
            admin_capabilities,
            subject_type: subject_type.to_string(),
            subject_id: subject_id.map(|s| s.to_string()),
            request_id: request_id.to_string(),
            context_zone_id: context_zone_id.map(|s| s.to_string()),
            zone_perms,
        }
    }
}

impl PyOperationContext {
    /// Convert to pure Rust OperationContext for kernel calls.
    fn to_rust(&self) -> OperationContext {
        OperationContext {
            user_id: self.user_id.clone(),
            zone_id: self.zone_id.clone(),
            is_admin: self.is_admin,
            agent_id: self.agent_id.clone(),
            is_system: self.is_system,
            groups: self.groups.clone(),
            admin_capabilities: self.admin_capabilities.clone(),
            subject_type: self.subject_type.clone(),
            subject_id: self.subject_id.clone(),
            request_id: self.request_id.clone(),
            context_zone_id: self.context_zone_id.clone(),
            zone_perms: self.zone_perms.clone(),
        }
    }
}

// ── PySysReadResult ─────────────────────────────────────────────

/// Python-facing SysReadResult (data is PyBytes, not Vec<u8>).
#[pyclass(get_all)]
pub struct PySysReadResult {
    pub data: Option<Py<PyBytes>>,
    pub post_hook_needed: bool,
    pub content_id: Option<String>,
    pub gen: u64,
    pub entry_type: u8,
    pub stream_next_offset: Option<usize>,
}

// ── PySysWriteResult ────────────────────────────────────────────

/// Python-facing SysWriteResult.
#[pyclass(get_all)]
pub struct PySysWriteResult {
    pub hit: bool,
    pub content_id: Option<String>,
    pub post_hook_needed: bool,
    pub version: u32,
    pub gen: u64,
    pub size: u64,
    pub is_new: bool,
    pub old_content_id: Option<String>,
    pub old_size: Option<u64>,
    pub old_version: Option<u32>,
    pub old_modified_at_ms: Option<i64>,
}

// ── PyFlushWriteBufferResult ───────────────────────────────────

/// Python-facing FlushWriteBufferResult.
#[pyclass(get_all)]
pub struct PyFlushWriteBufferResult {
    pub flushed: usize,
    pub failed: usize,
    pub errors: Vec<String>,
}

// ── PySysUnlinkResult ───────────────────────────────────────────

/// Python-facing SysUnlinkResult.
#[pyclass(get_all)]
pub struct PySysUnlinkResult {
    pub hit: bool,
    pub entry_type: u8,
    pub post_hook_needed: bool,
    pub path: String,
    pub content_id: Option<String>,
    pub size: u64,
}

// ── PySysRenameResult ───────────────────────────────────────────

/// Python-facing SysRenameResult.
#[pyclass(get_all)]
pub struct PySysRenameResult {
    pub hit: bool,
    pub success: bool,
    pub post_hook_needed: bool,
    pub is_directory: bool,
    pub old_content_id: Option<String>,
    pub old_size: Option<u64>,
    pub old_version: Option<u32>,
    pub old_modified_at_ms: Option<i64>,
}

// ── PySysMkdirResult ────────────────────────────────────────────

/// Python-facing SysMkdirResult.
#[pyclass(get_all)]
pub struct PySysMkdirResult {
    pub hit: bool,
    pub post_hook_needed: bool,
}

// ── PySysCopyResult ─────────────────────────────────────────────

/// Python-facing SysCopyResult.
#[pyclass(get_all)]
pub struct PySysCopyResult {
    pub hit: bool,
    pub post_hook_needed: bool,
    pub dst_path: String,
    pub content_id: Option<String>,
    pub size: u64,
    pub version: u32,
    pub gen: u64,
}

// ── PyBatchReadItem ──────────────────────────────────────────────

/// Per-item result for `sys_read_batch`. `error_kind == ""` means success.
/// On error `data` / `content_id` are `None` and counters are zero.
#[pyclass(get_all)]
pub struct PyBatchReadItem {
    pub data: Option<Py<PyBytes>>,
    pub content_id: Option<String>,
    pub gen: u64,
    pub entry_type: u8,
    pub post_hook_needed: bool,
    /// Empty string on success; one of "not_found" / "permission_denied" /
    /// "invalid_path" / "io_error" on failure.
    pub error_kind: String,
    /// Human-readable error detail; empty string on success.
    pub error_message: String,
}

// ═══════════════════════════════════════════════════════════════════════════
// PyKernel — wraps pure Rust Kernel + owns Hook registry
// ═══════════════════════════════════════════════════════════════════════════

/// Python-facing Kernel. Wraps the pure Rust `Kernel` and adds:
///   - Hook registry (PyO3-specific, stored here not in Rust Kernel)
///   - PRE-INTERCEPT dispatch (requires GIL for Python hook contexts)
///   - Type conversion (Vec<u8> -> PyBytes, StatResult -> PyDict, etc.)
#[pyclass]
pub struct PyKernel {
    /// Crate-visible so `grpc_server.rs` (and any other
    /// kernel-internal task spawner) can clone the Arc
    /// without an extra accessor method that codegen would
    /// have to preserve. Not exposed to Python.
    pub(crate) inner: Arc<Kernel>,
    _write_buffer_flusher: Option<WriteBufferFlushHandle>,
    // RwLock (not Mutex) so a hook callback can re-enter
    // ``sys_*`` without deadlocking. The recursion is real:
    // ReBAC's permission_hook reads its own ``/__sys__/rebac/...``
    // config via ``sys_read`` during a permission check, which
    // re-enters dispatch_pre_hooks on the same thread. Read sites
    // call ``read_unconditional`` (the ``RwLockExt`` rename for
    // parking_lot's ``read_recursive``) so the recursive shared
    // acquisition always succeeds; writes happen only at hook
    // register / unregister, never during dispatch, so the writer-
    // starvation cost does not apply.
    hooks: RwLock<HookRegistry>,
}

// Rust-side helpers on PyKernel — NOT exposed to Python (no #[pymethods]).
impl PyKernel {
    /// Compute the kernel's canonical key for a `(mount_point, zone_id)`
    /// pair. Forwards to `Kernel::canonical_mount_key`.
    pub fn canonical_mount_key(mount_point: &str, zone_id: &str) -> String {
        Kernel::canonical_mount_key(mount_point, zone_id)
    }

    /// Borrow the inner `&Kernel`.  Phase 3: peer crates
    /// (services, future transport / raft glue) reach the
    /// kernel's in-tree Rust API surface
    /// (`register_native_hook`, `prepare_audit_stream`,
    /// `sys_*` direct) through this accessor instead of the
    /// `pub(crate) inner` field, which other crates can't see.
    pub fn kernel_ref(&self) -> &Kernel {
        &self.inner
    }

    /// Clone the inner `Arc<Kernel>`.  Same Phase 3 motivation
    /// as [`Self::kernel_ref`] but returns an owned `Arc<Kernel>`
    /// for callers that need to spawn tasks holding the kernel
    /// (e.g. `transport::grpc::spawn` clones the Arc into a tonic
    /// worker task).
    pub fn kernel_arc(&self) -> Arc<Kernel> {
        Arc::clone(&self.inner)
    }
}

#[pymethods]
impl PyKernel {
    // ── Constructor ────────────────────────────────────────────────────

    #[new]
    fn new() -> Self {
        let inner = Arc::new(Kernel::new());
        // Write-buffer background flusher is NOT started here.
        // Call `start_write_buffer_flusher(interval_ms)` explicitly
        // from Python production code paths that need coalescing.
        // Starting it unconditionally caused 13-min hangs on macOS
        // CI: each PyKernel spawned a non-daemon thread that
        // prevented xdist worker processes from exiting.
        Self {
            inner,
            _write_buffer_flusher: None,
            hooks: RwLock::new(HookRegistry::new()),
        }
    }

    /// Start the background write-buffer flusher thread.
    /// Call from production entry points (NexusFS factory, connect()).
    /// NOT called by default in `new()` — test kernels don't need it.
    fn start_write_buffer_flusher(&mut self, interval_ms: u64) {
        if self._write_buffer_flusher.is_none() {
            self._write_buffer_flusher = Some(Kernel::spawn_write_buffer_flusher(
                &self.inner,
                std::time::Duration::from_millis(interval_ms),
            ));
        }
    }

    // ── Lock Manager wiring ──────────────────────────────────────────

    fn set_vfs_lock_timeout(&self, timeout_ms: u64) {
        self.inner.set_vfs_lock_timeout(timeout_ms);
    }

    /// Set node advertise address for origin-aware metadata.
    fn set_self_address(&self, addr: &str) {
        self.inner.set_self_address(addr);
    }

    // ── §13 Permission gate wiring ──────────────────────────────────

    /// Register a Python permission provider (wraps in PyPermissionProviderAdapter).
    fn set_permission_provider(&self, provider: Py<PyAny>) {
        let adapter = PyPermissionProviderAdapter { checker: provider };
        self.inner
            .set_permission_provider(std::sync::Arc::new(adapter));
    }

    /// Configure admin bypass (default: true).
    fn set_permission_admin_bypass(&self, enabled: bool) {
        self.inner.set_permission_admin_bypass(enabled);
    }

    /// Invalidate permission lease for a specific path.
    fn permission_lease_invalidate_path(&self, path: &str) {
        self.inner.permission_lease_invalidate_path(path);
    }

    /// Invalidate permission leases for a specific agent.
    fn permission_lease_invalidate_agent(&self, agent_id: &str) {
        self.inner.permission_lease_invalidate_agent(agent_id);
    }

    /// Invalidate all permission leases.
    fn permission_lease_invalidate_all(&self) {
        self.inner.permission_lease_invalidate_all();
    }

    // ── MetaStore wiring ──────────────────────────────────────────────

    /// Wire LocalMetaStore by path — Rust kernel opens redb directly.
    /// Eliminates GIL crossing on every metastore.get/put.
    fn set_metastore_path(&self, path: &str) -> PyResult<()> {
        self.inner.set_metastore_path(path).map_err(Into::into)
    }

    /// Drop global + per-mount redb metastores so a subsequent kernel
    /// can reopen the same redb path without ``Database already open``.
    /// Called by Python ``NexusFS.close`` / nested ``ephemeral_mount``.
    fn release_metastores(&self) {
        self.inner.release_metastores()
    }

    #[pyo3(signature = (prefix, recursive=true, limit=1000, cursor=None))]
    fn metastore_list_paginated(
        &self,
        py: Python<'_>,
        prefix: &str,
        recursive: bool,
        limit: usize,
        cursor: Option<&str>,
    ) -> PyResult<Py<PyAny>> {
        let page = self
            .inner
            .metastore_list_paginated(prefix, recursive, limit, cursor)
            .map_err::<PyErr, _>(Into::into)?;
        let dict = PyDict::new(py);
        let items: Vec<Py<PyAny>> = page
            .items
            .iter()
            .map(|m| {
                to_python_metadata(py, m)
                    .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("{e:?}")))
                    .map(Into::into)
            })
            .collect::<PyResult<Vec<_>>>()?;
        dict.set_item("items", items)?;
        dict.set_item("next_cursor", page.next_cursor)?;
        dict.set_item("has_more", page.has_more)?;
        dict.set_item("total_count", page.total_count)?;
        Ok(dict.into())
    }

    // ── Tier 2 convenience methods ────────────────────────────────────

    #[pyo3(signature = (paths, zone_id="root"))]
    fn stat_batch(
        &self,
        py: Python<'_>,
        paths: Vec<String>,
        zone_id: &str,
    ) -> Vec<Option<Py<PyAny>>> {
        use crate::kernel::convenience::KernelConvenience;
        let results = self.inner.stat_batch(&paths, zone_id);
        results
            .into_iter()
            .map(|opt| opt.map(|stat| stat_result_to_pydict(py, &stat).into()))
            .collect()
    }

    fn set_xattr(&self, path: &str, key: &str, value: String) -> PyResult<()> {
        use crate::kernel::convenience::KernelConvenience;
        self.inner
            .set_xattr(path, key, value, contracts::ROOT_ZONE_ID)
            .map_err(Into::into)
    }

    fn get_xattr(&self, path: &str, key: &str) -> PyResult<Option<String>> {
        use crate::kernel::convenience::KernelConvenience;
        self.inner
            .get_xattr(path, key, contracts::ROOT_ZONE_ID)
            .map_err(Into::into)
    }

    fn get_xattr_bulk(&self, py: Python<'_>, paths: Vec<String>, key: &str) -> PyResult<Py<PyAny>> {
        use crate::kernel::convenience::KernelConvenience;
        let pairs = self
            .inner
            .get_xattr_bulk(&paths, key, contracts::ROOT_ZONE_ID)
            .map_err::<PyErr, _>(Into::into)?;
        let dict = PyDict::new(py);
        for (path, value) in pairs {
            match value {
                Some(text) => dict.set_item(path, text)?,
                None => dict.set_item(path, py.None())?,
            }
        }
        Ok(dict.into())
    }

    /// Tier 2 write: create-or-overwrite (composes sys_write + setattr_update).
    fn write<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        ctx: &PyOperationContext,
        content: &[u8],
        offset: u64,
    ) -> PyResult<PySysWriteResult> {
        use crate::kernel::convenience::KernelConvenience;

        // PRE-INTERCEPT hooks (same as sys_write)
        if self.inner.has_hooks("write") {
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let whc = hc.write.bind(py).call1((path, content, py_ctx))?.unbind();
            self.dispatch_pre_hooks_inner("write", &whc)?;
        }

        let rust_ctx = ctx.to_rust();
        let content_owned = content.to_vec();
        let result = py.detach(|| self.inner.write(path, &rust_ctx, &content_owned, offset));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysWriteResult {
            hit: result.hit,
            content_id: result.content_id,
            post_hook_needed: result.post_hook_needed,
            version: result.version,
            gen: result.gen,
            size: result.size,
            is_new: result.is_new,
            old_content_id: result.old_content_id,
            old_size: result.old_size,
            old_version: result.old_version,
            old_modified_at_ms: result.old_modified_at_ms,
        })
    }

    #[pyo3(signature = (path, zone_id="root"))]
    fn get_content_id(&self, path: &str, zone_id: &str) -> Option<String> {
        use crate::kernel::convenience::KernelConvenience;
        self.inner.get_content_id(path, zone_id)
    }

    #[pyo3(signature = (path, zone_id="root"))]
    fn is_directory(&self, path: &str, zone_id: &str) -> bool {
        use crate::kernel::convenience::KernelConvenience;
        self.inner.is_directory(path, zone_id)
    }

    #[pyo3(signature = (zone_id="root"))]
    fn get_top_level_mounts(&self, zone_id: &str) -> Vec<String> {
        use crate::kernel::convenience::KernelConvenience;
        self.inner.get_top_level_mounts(zone_id)
    }

    #[pyo3(signature = (paths, zone_id="root"))]
    fn exists_batch(&self, paths: Vec<String>, zone_id: &str) -> Vec<bool> {
        use crate::kernel::convenience::KernelConvenience;
        self.inner.exists_batch(&paths, zone_id)
    }

    #[pyo3(signature = (parent_path, zone_id="root", is_admin=false, limit=0, cursor=None))]
    fn readdir_paged(
        &self,
        py: Python<'_>,
        parent_path: &str,
        zone_id: &str,
        is_admin: bool,
        limit: usize,
        cursor: Option<&str>,
    ) -> Py<PyAny> {
        let result = self
            .inner
            .readdir_paged(parent_path, zone_id, is_admin, limit, cursor);
        let dict = PyDict::new(py);
        let _ = dict.set_item("items", &result.items);
        let _ = dict.set_item("next_cursor", &result.next_cursor);
        let _ = dict.set_item("has_more", result.has_more);
        dict.into()
    }

    // ── Advisory lock syscalls (F4 C5) ──────────────────────────────

    #[pyo3(signature = (path, lock_id="", mode="exclusive", max_holders=1, ttl_secs=60, holder_info=""))]
    #[allow(clippy::too_many_arguments)]
    fn sys_lock(
        &self,
        path: &str,
        lock_id: &str,
        mode: &str,
        max_holders: u32,
        ttl_secs: u64,
        holder_info: &str,
    ) -> PyResult<Option<String>> {
        let parsed_mode = match mode.to_ascii_lowercase().as_str() {
            "exclusive" => crate::lock_manager::KernelLockMode::Exclusive,
            "shared" => crate::lock_manager::KernelLockMode::Shared,
            other => {
                return Err(pyo3::exceptions::PyValueError::new_err(format!(
                    "invalid lock mode '{}': expected 'exclusive' or 'shared'",
                    other
                )));
            }
        };
        self.inner
            .sys_lock(
                path,
                lock_id,
                parsed_mode,
                max_holders,
                ttl_secs,
                holder_info,
            )
            .map_err(Into::into)
    }

    #[pyo3(signature = (path, lock_id="", force=false))]
    fn sys_unlock(&self, path: &str, lock_id: &str, force: bool) -> PyResult<bool> {
        self.inner
            .sys_unlock(path, lock_id, force)
            .map_err(Into::into)
    }

    /// Linearizable enumeration of locks under `prefix`, capped
    /// at `limit`. Returns a list of dicts shaped exactly like
    /// the old Python `_format_lock_info` output:
    ///   { "path": str, "mode": "mutex" | "semaphore",
    ///     "max_holders": int,
    ///     "holders": [ { "lock_id": str, "holder_info": str,
    ///                     "acquired_at": int,
    ///                     "expires_at": int } ] }
    ///
    /// Internal metastore-proxy helper, NOT a separate syscall.
    /// Called by the Python `NexusFS.sys_readdir` intercept for
    /// `/__sys__/locks/` (Linux /proc/locks convention), which
    /// preserves the existing user-facing `sys_readdir` API.
    /// The per-holder `LockMode` tracked inside the state
    /// machine for RW conflict correctness is intentionally
    /// NOT surfaced here — the dict shape matches Python's
    /// pre-F4 `_format_lock_info` contract so the surviving
    /// `/api/v2/locks` + `nx locks list` consumers keep
    /// parsing unchanged.
    #[pyo3(signature = (prefix="", limit=1024))]
    fn metastore_list_locks<'py>(
        &self,
        py: Python<'py>,
        prefix: &str,
        limit: usize,
    ) -> PyResult<Vec<Bound<'py, PyDict>>> {
        let locks = self
            .inner
            .metastore_list_locks(prefix, limit)
            .map_err::<PyErr, _>(Into::into)?;
        let mut out = Vec::with_capacity(locks.len());
        for lock in &locks {
            let dict = PyDict::new(py);
            dict.set_item("path", &lock.path)?;
            // Lock-level mode is the computed display label,
            // not the stored per-holder conflict mode.
            let label = if lock.max_holders == 1 {
                "mutex"
            } else {
                "semaphore"
            };
            dict.set_item("mode", label)?;
            dict.set_item("max_holders", lock.max_holders)?;
            let holders = pyo3::types::PyList::empty(py);
            for h in &lock.holders {
                let h_dict = PyDict::new(py);
                h_dict.set_item("lock_id", &h.lock_id)?;
                h_dict.set_item("holder_info", &h.holder_info)?;
                h_dict.set_item("acquired_at", h.acquired_at_secs)?;
                h_dict.set_item("expires_at", h.expires_at_secs)?;
                holders.append(h_dict)?;
            }
            dict.set_item("holders", holders)?;
            out.push(dict);
        }
        Ok(out)
    }

    // ── R10c: direct CAS surface — PyKernel delegators ───────────────
    //
    // Thin wrappers around Kernel::cas_* that release the GIL for the
    // storage work. Each method mirrors a Python CASAddressingEngine
    // hot-path call so the Python delegator can collapse to `return
    // self._kernel.cas_*(...)`. Error conversion reuses the KernelError
    // → PyErr pipeline so NotFound surfaces as NexusFileNotFoundError
    // and I/O surfaces as BackendError, both with mount + op breadcrumbs.

    #[pyo3(signature = (mount_point, zone_id, content, *, ttl_seconds=None))]
    fn cas_write<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content: Vec<u8>,
        ttl_seconds: Option<u64>,
    ) -> PyResult<(String, bool)> {
        py.detach(|| {
            self.inner
                .cas_write(mount_point, zone_id, &content, ttl_seconds)
        })
        .map_err(Into::into)
    }

    #[pyo3(signature = (mount_point, zone_id, content_id, *, origins=None))]
    fn cas_read<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content_id: &str,
        origins: Option<Vec<String>>,
    ) -> PyResult<Py<PyBytes>> {
        let origins_vec = origins.unwrap_or_default();
        let bytes = py
            .detach(|| {
                self.inner
                    .cas_read(mount_point, zone_id, content_id, &origins_vec)
            })
            .map_err::<PyErr, _>(Into::into)?;
        Ok(PyBytes::new(py, &bytes).unbind())
    }

    fn cas_delete<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content_id: &str,
    ) -> PyResult<()> {
        py.detach(|| self.inner.cas_delete(mount_point, zone_id, content_id))
            .map_err(Into::into)
    }

    fn cas_exists<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content_id: &str,
    ) -> PyResult<bool> {
        py.detach(|| self.inner.cas_exists(mount_point, zone_id, content_id))
            .map_err(Into::into)
    }

    fn cas_size<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content_id: &str,
    ) -> PyResult<u64> {
        py.detach(|| self.inner.cas_size(mount_point, zone_id, content_id))
            .map_err(Into::into)
    }

    fn cas_is_chunked<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        content_id: &str,
    ) -> PyResult<bool> {
        py.detach(|| self.inner.cas_is_chunked(mount_point, zone_id, content_id))
            .map_err(Into::into)
    }

    #[pyo3(signature = (mount_point, zone_id, old_hash, buf, offset, *, origins=None))]
    #[allow(clippy::too_many_arguments)]
    fn cas_write_partial<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        old_hash: &str,
        buf: Vec<u8>,
        offset: u64,
        origins: Option<Vec<String>>,
    ) -> PyResult<String> {
        let origins_vec = origins.unwrap_or_default();
        py.detach(|| {
            self.inner
                .cas_write_partial(mount_point, zone_id, old_hash, &buf, offset, &origins_vec)
        })
        .map_err(Into::into)
    }

    // ── R10d: LLM streaming entry point (OpenAI / SSE) ───────────────
    //
    // Resolves (mount_point, zone_id) → OpenAIBackend via
    // `ObjectStore::as_llm_streaming()`, then runs the full SSE → DT_STREAM
    // → CAS-persist pipeline in a GIL-free worker. Caller is expected to
    // invoke via `asyncio.to_thread(...)` so the event loop is not blocked
    // for the duration of the completion.

    #[cfg(feature = "connectors")]
    fn llm_start_streaming<'py>(
        &self,
        py: Python<'py>,
        mount_point: &str,
        zone_id: &str,
        request_bytes: Vec<u8>,
        stream_path: &str,
    ) -> PyResult<()> {
        let canonical = crate::vfs_router::canonicalize_mount_path(mount_point, zone_id);
        let entry = self
            .inner
            .vfs_router
            .get_canonical(&canonical)
            .ok_or_else(|| {
                pyo3::exceptions::PyFileNotFoundError::new_err(format!(
                    "llm_start_streaming: mount not found: {}@{}",
                    mount_point, zone_id
                ))
            })?;
        let backend = entry.backend.as_ref().ok_or_else(|| {
            pyo3::exceptions::PyRuntimeError::new_err(format!(
                "llm_start_streaming: mount has no backend at {}",
                mount_point
            ))
        })?;
        let llm = backend.as_llm_streaming().ok_or_else(|| {
            pyo3::exceptions::PyRuntimeError::new_err(format!(
                "llm_start_streaming: backend at mount {} does not support streaming",
                mount_point
            ))
        })?;
        let stream_manager = Arc::clone(&self.inner.stream_manager);
        let stream_path_owned = stream_path.to_string();
        py.detach(move || {
            // Create the DT_STREAM buffer before run_streaming writes to it.
            // 64 KiB matches the default capacity used in unit tests.
            let _ = stream_manager.create(&stream_path_owned, 64 * 1024);
            llm.run_streaming(&request_bytes, &stream_path_owned, &stream_manager)
                .map_err(|e| {
                    pyo3::exceptions::PyRuntimeError::new_err(format!("llm_start_streaming: {}", e))
                })
        })
    }

    // ── sys_setattr — unified mount/attr syscall ─────────────────────

    #[pyo3(signature = (path, entry_type, backend_name="", local_root=None, fsync=false, backend_type="", follow_symlinks=true, openai_base_url=None, openai_api_key=None, openai_model=None, openai_blob_root=None, anthropic_base_url=None, anthropic_api_key=None, anthropic_model=None, anthropic_blob_root=None, s3_bucket=None, s3_prefix=None, aws_region=None, aws_access_key=None, aws_secret_key=None, s3_endpoint=None, gcs_bucket=None, gcs_prefix=None, access_token=None, root_folder_id=None, bot_token=None, default_channel=None, hn_stories_per_feed=None, hn_include_comments=None, cli_command=None, cli_service=None, cli_auth_env_json=None, x_bearer_token=None, metastore_path=None, io_profile="balanced", zone_id="root", is_external=false, capacity=65536, mime_type=None, modified_at_ms=None, content_id=None, size=None, version=None, created_at_ms=None, read_fd=None, write_fd=None, server_address=None, remote_auth_token=None, remote_ca_pem=None, remote_cert_pem=None, remote_key_pem=None, remote_timeout=30.0, link_target=None, source=None))]
    #[allow(clippy::too_many_arguments)]
    fn sys_setattr<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        entry_type: i32,
        backend_name: &str,
        local_root: Option<&str>,
        fsync: bool,
        backend_type: &str,
        follow_symlinks: bool,
        openai_base_url: Option<&str>,
        openai_api_key: Option<&str>,
        openai_model: Option<&str>,
        openai_blob_root: Option<&str>,
        anthropic_base_url: Option<&str>,
        anthropic_api_key: Option<&str>,
        anthropic_model: Option<&str>,
        anthropic_blob_root: Option<&str>,
        s3_bucket: Option<&str>,
        s3_prefix: Option<&str>,
        aws_region: Option<&str>,
        aws_access_key: Option<&str>,
        aws_secret_key: Option<&str>,
        s3_endpoint: Option<&str>,
        gcs_bucket: Option<&str>,
        gcs_prefix: Option<&str>,
        access_token: Option<&str>,
        root_folder_id: Option<&str>,
        bot_token: Option<&str>,
        default_channel: Option<&str>,
        hn_stories_per_feed: Option<usize>,
        hn_include_comments: Option<bool>,
        cli_command: Option<&str>,
        cli_service: Option<&str>,
        cli_auth_env_json: Option<&str>,
        x_bearer_token: Option<&str>,
        metastore_path: Option<&str>,
        io_profile: &str,
        zone_id: &str,
        is_external: bool,
        capacity: usize,
        mime_type: Option<&str>,
        modified_at_ms: Option<i64>,
        content_id: Option<&str>,
        size: Option<u64>,
        version: Option<u32>,
        created_at_ms: Option<i64>,
        read_fd: Option<i32>,
        write_fd: Option<i32>,
        server_address: Option<&str>,
        remote_auth_token: Option<&str>,
        remote_ca_pem: Option<&[u8]>,
        remote_cert_pem: Option<&[u8]>,
        remote_key_pem: Option<&[u8]>,
        remote_timeout: f64,
        link_target: Option<&str>,
        source: Option<&str>,
    ) -> PyResult<Py<PyAny>> {
        // 17-way backend-type construction lives in
        // `backends::python::factory::DefaultObjectStoreProvider`.
        // Kernel reaches concrete backend types through the §3.B.2
        // `ObjectStoreProvider` trait, installed by `nexus-cdylib`
        // at module init.
        let provider = crate::hal::object_store_provider::get_provider()
            .ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err(
                    "sys_setattr: ObjectStoreProvider not registered — non-cdylib build needs to call kernel::hal::object_store_provider::set_provider at boot",
                )
            })?;
        // peer_client is a `RwLock<Arc<dyn PeerBlobClient>>` so the
        // cdylib's `install_transport_wiring` swaps the Noop default
        // for the real concrete impl post-boot. Lock and clone the
        // inner `Arc` for the provider call so the read guard does
        // not outlive this scope.
        let peer_client_arc: Arc<dyn crate::hal::peer::PeerBlobClient> =
            Arc::clone(&self.inner.peer_client.read());
        // Snapshot of self_address — `cas_local` plumbs into the per-mount
        // `GrpcChunkFetcher` so chunk-miss scatter skips this node.  Snapshotted
        // once here so the provider call sees a consistent `&str` borrow.
        let self_addr_snapshot: Option<String> = self.inner.self_address_string();
        let provider_args = crate::hal::object_store_provider::ObjectStoreProviderArgs {
            backend_type,
            backend_name,
            local_root,
            fsync,
            follow_symlinks,
            openai_base_url,
            openai_api_key,
            openai_model,
            openai_blob_root,
            anthropic_base_url,
            anthropic_api_key,
            anthropic_model,
            anthropic_blob_root,
            s3_bucket,
            s3_prefix,
            aws_region,
            aws_access_key,
            aws_secret_key,
            s3_endpoint,
            gcs_bucket,
            gcs_prefix,
            access_token,
            root_folder_id,
            bot_token,
            default_channel,
            hn_stories_per_feed,
            hn_include_comments,
            cli_command,
            cli_service,
            cli_auth_env_json,
            x_bearer_token,
            server_address,
            remote_auth_token,
            remote_ca_pem,
            remote_cert_pem,
            remote_key_pem,
            remote_timeout,
            peer_client: &peer_client_arc,
            self_address: self_addr_snapshot.as_deref(),
            runtime: self.inner.runtime(),
        };
        let backend_result = provider
            .build(&provider_args)
            .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
        let backend = backend_result.backend;
        let remote_metastore = backend_result.pending_remote_meta_store;

        // MetaStore resolution: metastore_path -> LocalMetaStore.
        // Federation DT_MOUNT auto-resolves its raft backing via
        // `resolve_federation_mount_backing` inside Kernel::sys_setattr
        // (R20.18.3), so Python always passes `None` for raft_backend.
        let metastore: Option<Arc<dyn crate::meta_store::MetaStore>> =
            if let Some(ms_path) = metastore_path {
                let ms = crate::meta_store::LocalMetaStore::open(std::path::Path::new(ms_path))
                    .map_err(|e| {
                        pyo3::exceptions::PyIOError::new_err(format!("LocalMetaStore: {e:?}"))
                    })?;
                Some(Arc::new(ms) as Arc<dyn crate::meta_store::MetaStore>)
            } else {
                None
            };

        let result = self
            .inner
            .sys_setattr(
                path,
                entry_type,
                backend_name,
                backend,
                metastore,
                None, // raft_backend — auto-resolved inside Kernel::sys_setattr
                io_profile,
                zone_id,
                is_external,
                capacity,
                read_fd,
                write_fd,
                mime_type,
                modified_at_ms,
                content_id,
                size,
                version,
                created_at_ms,
                link_target,
                source,
                remote_metastore,
            )
            .map_err::<PyErr, _>(Into::into)?;

        let dict = PyDict::new(py);
        dict.set_item("path", result.path)?;
        dict.set_item("created", result.created)?;
        dict.set_item("entry_type", result.entry_type)?;
        if let Some(bn) = result.backend_name {
            dict.set_item("backend_name", bn)?;
        }
        if let Some(cap) = result.capacity {
            dict.set_item("capacity", cap)?;
        }
        if !result.updated.is_empty() {
            dict.set_item("updated", result.updated)?;
        }
        if let Some(shm) = result.shm_path {
            dict.set_item("shm_path", shm)?;
        }
        if let Some(fd) = result.data_rd_fd {
            dict.set_item("data_rd_fd", fd)?;
        }
        if let Some(fd) = result.space_rd_fd {
            dict.set_item("space_rd_fd", fd)?;
        }
        Ok(dict.into())
    }

    // ── Router proxy methods ───────────────────────────────────────────

    fn has_mount(&self, mount_point: &str, zone_id: &str) -> bool {
        self.inner.has_mount(mount_point, zone_id)
    }

    fn get_mount_points(&self) -> Vec<String> {
        self.inner.get_mount_points()
    }

    #[pyo3(signature = (mount_point, zone_id="root"))]
    fn kernel_unmount(&self, mount_point: &str, zone_id: &str) -> bool {
        self.inner.dlc.unmount(&self.inner, mount_point, zone_id)
    }

    // ── IPC Registry — Pipe methods ──────────────────────────────────

    fn create_pipe(&self, path: &str, capacity: usize) -> PyResult<()> {
        self.inner.create_pipe(path, capacity).map_err(Into::into)
    }

    fn destroy_pipe(&self, path: &str) -> PyResult<()> {
        self.inner.destroy_pipe(path).map_err(Into::into)
    }

    fn close_pipe(&self, path: &str) -> PyResult<()> {
        self.inner.close_pipe(path).map_err(Into::into)
    }

    fn has_pipe(&self, path: &str) -> bool {
        self.inner.has_pipe(path)
    }

    fn pipe_write_nowait(&self, path: &str, data: &[u8]) -> PyResult<usize> {
        self.inner.pipe_write_nowait(path, data).map_err(Into::into)
    }

    fn pipe_read_nowait<'py>(
        &self,
        py: Python<'py>,
        path: &str,
    ) -> PyResult<Option<Bound<'py, PyBytes>>> {
        match self.inner.pipe_read_nowait(path) {
            Ok(Some(data)) => Ok(Some(PyBytes::new(py, &data))),
            Ok(None) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    fn list_pipes(&self) -> Vec<String> {
        self.inner.list_pipes()
    }

    fn close_all_pipes(&self) {
        self.inner.close_all_pipes()
    }

    // ── IPC Registry — Stream methods ────────────────────────────────

    fn create_stream(&self, path: &str, capacity: usize) -> PyResult<()> {
        self.inner.create_stream(path, capacity).map_err(Into::into)
    }

    fn destroy_stream(&self, path: &str) -> PyResult<()> {
        self.inner.destroy_stream(path).map_err(Into::into)
    }

    fn close_stream(&self, path: &str) -> PyResult<()> {
        self.inner.close_stream(path).map_err(Into::into)
    }

    fn has_stream(&self, path: &str) -> bool {
        self.inner.has_stream(path)
    }

    fn stream_write_nowait(&self, path: &str, data: &[u8]) -> PyResult<usize> {
        self.inner
            .stream_write_nowait(path, data)
            .map_err(Into::into)
    }

    fn stream_read_at<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        offset: usize,
    ) -> PyResult<Option<(Bound<'py, PyBytes>, usize)>> {
        match self.inner.stream_read_at(path, offset) {
            Ok(Some((data, next))) => Ok(Some((PyBytes::new(py, &data), next))),
            Ok(None) => Ok(None),
            Err(e) => Err(e.into()),
        }
    }

    fn stream_read_batch<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        offset: usize,
        count: usize,
    ) -> PyResult<(Vec<Bound<'py, PyBytes>>, usize)> {
        let (msgs, next) = self
            .inner
            .stream_read_batch(path, offset, count)
            .map_err(|e| -> PyErr { e.into() })?;
        let py_msgs: Vec<Bound<'py, PyBytes>> = msgs.iter().map(|m| PyBytes::new(py, m)).collect();
        Ok((py_msgs, next))
    }

    fn stream_collect_all<'py>(
        &self,
        py: Python<'py>,
        path: &str,
    ) -> PyResult<Bound<'py, PyBytes>> {
        let data = self
            .inner
            .stream_collect_all(path)
            .map_err(|e| -> PyErr { e.into() })?;
        Ok(PyBytes::new(py, &data))
    }

    fn list_streams(&self) -> Vec<String> {
        self.inner.list_streams()
    }

    fn close_all_streams(&self) {
        self.inner.close_all_streams()
    }

    // ── Trie proxy methods ─────────────────────────────────────────────

    fn trie_register(&self, pattern: &str, resolver_idx: usize) -> PyResult<()> {
        self.inner
            .trie_register(pattern, resolver_idx)
            .map_err(Into::into)
    }

    fn trie_unregister(&self, resolver_idx: usize) -> bool {
        self.inner.trie_unregister(resolver_idx)
    }

    fn trie_lookup(&self, path: &str) -> Option<usize> {
        self.inner.trie_lookup(path)
    }

    fn trie_len(&self) -> usize {
        self.inner.trie_len()
    }

    // ── Hook proxy methods ─────────────────────────────────────────────

    fn register_hook(&self, py: Python<'_>, op: &str, hook: Py<PyAny>) -> PyResult<()> {
        let hook_ref = hook.bind(py);
        let name: String = hook_ref
            .getattr("name")
            .and_then(|n| n.extract())
            .unwrap_or_else(|_| "<hook>".to_string());

        let pre_attr = format!("on_pre_{op}");
        let has_pre = hook_ref
            .getattr(pre_attr.as_str())
            .map(|attr| !attr.is_none())
            .unwrap_or(false);

        let post_attr = format!("on_post_{op}");
        let is_async_post = match hook_ref.getattr(post_attr.as_str()) {
            Ok(post_fn) => py
                .import("inspect")?
                .call_method1("iscoroutinefunction", (post_fn,))?
                .extract::<bool>()
                .unwrap_or(false),
            Err(_) => false,
        };

        #[cfg(feature = "py-hook-adapters")]
        {
            let adapter = PyInterceptHookAdapter::new(py, hook.clone_ref(py));
            self.hooks
                .write()
                .register(op, Box::new(adapter), hook, has_pre, is_async_post, name);
            Ok(())
        }

        #[cfg(not(feature = "py-hook-adapters"))]
        {
            let _ = (hook, has_pre, is_async_post, name);
            Err(pyo3::exceptions::PyRuntimeError::new_err(
                "register_hook requires feature 'py-hook-adapters'",
            ))
        }
    }

    fn unregister_hook(&self, py: Python<'_>, op: &str, hook: &Bound<'_, PyAny>) -> bool {
        self.hooks.write().unregister(py, op, hook)
    }

    fn get_pre_hooks(&self, py: Python<'_>, op: &str) -> Vec<Py<PyAny>> {
        self.hooks.read_unconditional().get_pre_hooks(py, op)
    }

    fn get_post_hooks(&self, py: Python<'_>, op: &str) -> (Vec<Py<PyAny>>, Vec<Py<PyAny>>) {
        self.hooks.read_unconditional().get_post_hooks(py, op)
    }

    fn get_all_hooks(&self, py: Python<'_>, op: &str) -> Vec<Py<PyAny>> {
        self.hooks.read_unconditional().get_all_hooks(py, op)
    }

    fn hook_count(&self, op: &str) -> usize {
        self.hooks.read_unconditional().count(op)
    }

    // ── sys_watch (inotify equivalent) ───────────────────────────────

    /// sys_watch — block until a matching file event or timeout.
    /// Returns (event_type, path) tuple or None.
    #[pyo3(signature = (pattern, timeout_ms))]
    fn sys_watch(
        &self,
        py: Python<'_>,
        pattern: &str,
        timeout_ms: u64,
    ) -> Option<(String, String)> {
        let event = py.detach(|| self.inner.sys_watch(pattern, timeout_ms));
        event.map(|e| (e.event_type.as_str().to_string(), e.path.clone()))
    }

    // ── Observer dispatch (pure Rust) ────────────────────────────────

    /// Dispatch event to all registered Rust-native observers (for DLC mount/unmount).
    #[pyo3(signature = (event_type, path))]
    fn dispatch_event(&self, event_type: &str, path: &str) -> PyResult<()> {
        use crate::dispatch::FileEventType;
        let etype = match event_type {
            "file_write" => FileEventType::FileWrite,
            "file_delete" => FileEventType::FileDelete,
            "file_rename" => FileEventType::FileRename,
            "metadata_change" => FileEventType::MetadataChange,
            "dir_create" => FileEventType::DirCreate,
            "dir_delete" => FileEventType::DirDelete,
            "file_copy" => FileEventType::FileCopy,
            "mount" => FileEventType::Mount,
            "unmount" => FileEventType::Unmount,
            other => {
                return Err(pyo3::exceptions::PyValueError::new_err(format!(
                    "unknown event type: {other}"
                )))
            }
        };
        self.inner.dispatch_event(etype, path);
        Ok(())
    }

    /// Flush pending Rust-native observer tasks (blocks until pool drains).
    fn flush_observers(&self) {
        self.inner.flush_observers();
    }

    /// Total observers (Rust-native + event buffers).
    fn kernel_observer_count(&self) -> usize {
        self.inner.observer_count()
    }

    // ── Hook counts ────────────────────────────────────────────────────

    fn set_hook_count(&self, op: &str, count: u64) {
        self.inner.set_hook_count(op, count);
    }

    // ── Public hook dispatch (for Tier 2 Python callers) ──────────────

    /// Dispatch pre-hooks for an operation via Rust InterceptHook trait.
    ///
    /// Called by Tier 2 Python methods (read_range, stream, write, copy, etc.)
    /// that build their own hook context objects. Tier 1 syscalls (sys_read,
    /// sys_write, etc.) dispatch pre-hooks internally.
    #[pyo3(signature = (op, hook_ctx))]
    fn dispatch_pre_hooks(&self, op: &str, hook_ctx: Py<PyAny>) -> PyResult<()> {
        if !self.inner.has_hooks(op) {
            return Ok(());
        }
        let hooks = self.hooks.read_unconditional();
        let impls = hooks.get_pre_hook_impls(op);
        for hook in impls {
            match op {
                "read" => hook.on_pre_read(&hook_ctx)?,
                "write" => hook.on_pre_write(&hook_ctx)?,
                "delete" => hook.on_pre_delete(&hook_ctx)?,
                "rename" => hook.on_pre_rename(&hook_ctx)?,
                "mkdir" => hook.on_pre_mkdir(&hook_ctx)?,
                "rmdir" => hook.on_pre_rmdir(&hook_ctx)?,
                "copy" => hook.on_pre_copy(&hook_ctx)?,
                "stat" => hook.on_pre_stat(&hook_ctx)?,
                "access" => hook.on_pre_access(&hook_ctx)?,
                "write_batch" => hook.on_pre_write_batch(&hook_ctx)?,
                _ => {}
            }
        }
        Ok(())
    }

    // ── Post-hook dispatch (sync Rust + return async for Python) ───────���

    /// Dispatch post-hooks: sync in Rust, return async hooks for Python.
    ///
    /// Sync post-hooks: serial, fault-isolated (fire-and-forget).
    /// Returns async hooks as Vec<Py<PyAny>> for Python asyncio.gather.
    #[pyo3(signature = (op, hook_ctx))]
    fn dispatch_post_hooks(
        &self,
        py: Python<'_>,
        op: &str,
        hook_ctx: Py<PyAny>,
    ) -> PyResult<Vec<Py<PyAny>>> {
        if !self.inner.has_hooks(op) {
            return Ok(Vec::new());
        }
        // 1. Dispatch sync post-hooks in Rust (fire-and-forget)
        self.dispatch_post_hooks_sync(op, &hook_ctx);

        // 2. Return async hooks for Python to schedule
        let hooks = self.hooks.read_unconditional();
        let (_, async_hooks) = hooks.get_post_hooks(py, op);
        Ok(async_hooks)
    }

    // ── Batch stat permission check (single FFI call for N paths) ──────

    /// Batch stat permission check: dispatch stat pre-hooks for each path,
    /// returning a bool per path (true = allowed, false = denied).
    ///
    /// Reduces N PyO3 boundary crossings to 1 for batch operations like
    /// read_bulk and stat_bulk that check permissions on many paths.
    #[pyo3(signature = (paths, ctx, permission))]
    fn dispatch_pre_hooks_batch_stat(
        &self,
        py: Python<'_>,
        paths: Vec<String>,
        ctx: &PyOperationContext,
        permission: &str,
    ) -> PyResult<Vec<bool>> {
        // Fast path: no stat hooks → all paths allowed
        if !self.inner.has_hooks("stat") {
            return Ok(vec![true; paths.len()]);
        }

        // Build Python context once for all paths
        let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
            .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
        let hc = get_hook_ctx_cache(py).ok_or_else(|| {
            pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
        })?;

        // Cache PermissionDeniedError class for isinstance check
        let perm_denied_cls = get_exception_cache(py).map(|c| c.permission_denied.bind(py));

        let mut results = Vec::with_capacity(paths.len());
        for path in &paths {
            let shc = hc
                .stat
                .bind(py)
                .call1((path.as_str(), &py_ctx, permission))?
                .unbind();
            match self.dispatch_pre_hooks_inner("stat", &shc) {
                Ok(()) => results.push(true),
                Err(e) => {
                    // PermissionDeniedError or builtin PermissionError → denied;
                    // other errors propagate.  Issue #3786 / Codex Round 10 finding #3:
                    // PermissionChecker.check raises the builtin on normal denials,
                    // so we must catch both classes here or batch_stat would surface
                    // a 500 instead of a per-path False.
                    if let Some(cls) = perm_denied_cls {
                        if e.is_instance(py, cls) {
                            results.push(false);
                            continue;
                        }
                    }
                    if e.is_instance_of::<pyo3::exceptions::PyPermissionError>(py) {
                        results.push(false);
                        continue;
                    }
                    return Err(e);
                }
            }
        }
        Ok(results)
    }

    // ── sys_read ───────────────────────────────────────────────────────

    #[pyo3(signature = (path, ctx, timeout_ms=5000, offset=0))]
    fn sys_read<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        ctx: &PyOperationContext,
        timeout_ms: u64,
        offset: u64,
    ) -> PyResult<PySysReadResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("read") {
            // Convert Rust ctx to Python OperationContext dataclass (full round-trip)
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let rhc = hc.read.bind(py).call1((path, py_ctx))?.unbind();
            self.dispatch_pre_hooks_inner("read", &rhc)?;
        }

        // 2. Call pure Rust kernel (releasing GIL for VFS lock blocking)
        let rust_ctx = ctx.to_rust();
        let result = py.detach(|| self.inner.sys_read_one(path, &rust_ctx, timeout_ms, offset));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        // 3. Convert Vec<u8> -> PyBytes
        Ok(PySysReadResult {
            data: result.data.map(|d| PyBytes::new(py, &d).into()),
            post_hook_needed: result.post_hook_needed,
            content_id: result.content_id,
            gen: result.gen,
            entry_type: result.entry_type,
            stream_next_offset: result.stream_next_offset,
        })
    }

    // ── sys_write ──────────────────────────────────────────────────────

    #[pyo3(signature = (path, ctx, content, offset=0))]
    fn sys_write<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        ctx: &PyOperationContext,
        content: &[u8],
        offset: u64,
    ) -> PyResult<PySysWriteResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("write") {
            // Convert Rust ctx to Python OperationContext dataclass (full round-trip)
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let whc = hc.write.bind(py).call1((path, content, py_ctx))?.unbind();
            self.dispatch_pre_hooks_inner("write", &whc)?;
        }

        // 2. Call pure Rust kernel (releasing GIL for VFS lock blocking)
        let rust_ctx = ctx.to_rust();
        let content_owned = content.to_vec();
        let result = py.detach(|| {
            self.inner
                .sys_write_one(path, &rust_ctx, &content_owned, offset)
        });
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysWriteResult {
            hit: result.hit,
            content_id: result.content_id,
            post_hook_needed: result.post_hook_needed,
            version: result.version,
            gen: result.gen,
            size: result.size,
            is_new: result.is_new,
            old_content_id: result.old_content_id,
            old_size: result.old_size,
            old_version: result.old_version,
            old_modified_at_ms: result.old_modified_at_ms,
        })
    }

    // ── write-buffer flushing ─────────────────────────────────────────

    #[pyo3(signature = (path=None, zone_id=None))]
    fn flush_write_buffer(
        &self,
        py: Python<'_>,
        path: Option<&str>,
        zone_id: Option<&str>,
    ) -> PyResult<PyFlushWriteBufferResult> {
        let result = py.detach(|| self.inner.flush_write_buffer(path, zone_id));
        let result = result.map_err(|e| -> PyErr { e.into() })?;
        Ok(PyFlushWriteBufferResult {
            flushed: result.flushed,
            failed: result.failed,
            errors: result.errors,
        })
    }

    fn flush_due_write_buffer(&self, py: Python<'_>) -> PyResult<PyFlushWriteBufferResult> {
        let result = py.detach(|| self.inner.flush_due_write_buffer());
        let result = result.map_err(|e| -> PyErr { e.into() })?;
        Ok(PyFlushWriteBufferResult {
            flushed: result.flushed,
            failed: result.failed,
            errors: result.errors,
        })
    }

    // ── sys_stat ───────────────────────────────────────────────────────

    fn sys_stat<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        zone_id: &str,
    ) -> PyResult<Option<Bound<'py, PyDict>>> {
        match self.inner.sys_stat(path, zone_id) {
            Some(s) => {
                let dict = PyDict::new(py);
                dict.set_item("path", &s.path)?;
                dict.set_item("size", s.size)?;
                dict.set_item("last_writer_address", s.last_writer_address.as_deref())?;
                dict.set_item("content_id", s.content_id.as_deref())?;
                dict.set_item("mime_type", &s.mime_type)?;
                set_optional_iso_datetime(py, &dict, "created_at", s.created_at_ms)?;
                set_optional_iso_datetime(py, &dict, "modified_at", s.modified_at_ms)?;
                dict.set_item("is_directory", s.is_directory)?;
                dict.set_item("entry_type", s.entry_type)?;
                dict.set_item("mode", s.mode)?;
                dict.set_item("version", s.version)?;
                dict.set_item("gen", s.gen)?;
                dict.set_item("zone_id", s.zone_id.as_deref())?;
                dict.set_item("link_target", s.link_target.as_deref())?;
                dict.set_item("owner_id", s.owner_id.as_deref())?;
                match &s.lock {
                    Some(lock) => {
                        let lock_dict = PyDict::new(py);
                        let label = if lock.max_holders == 1 {
                            "mutex"
                        } else {
                            "semaphore"
                        };
                        lock_dict.set_item("mode", label)?;
                        lock_dict.set_item("max_holders", lock.max_holders)?;
                        let holders = pyo3::types::PyList::empty(py);
                        for h in &lock.holders {
                            let h_dict = PyDict::new(py);
                            h_dict.set_item("lock_id", &h.lock_id)?;
                            h_dict.set_item("holder_info", &h.holder_info)?;
                            h_dict.set_item("acquired_at", h.acquired_at_secs)?;
                            h_dict.set_item("expires_at", h.expires_at_secs)?;
                            holders.append(h_dict)?;
                        }
                        lock_dict.set_item("holders", holders)?;
                        dict.set_item("lock", lock_dict)?;
                    }
                    None => {
                        dict.set_item("lock", py.None())?;
                    }
                }
                Ok(Some(dict))
            }
            None => Ok(None),
        }
    }

    // ── sys_unlink ────────────────────────────────────────────────────

    #[pyo3(signature = (path, ctx, recursive=false))]
    fn sys_unlink(
        &self,
        py: Python<'_>,
        path: &str,
        ctx: &PyOperationContext,
        recursive: bool,
    ) -> PyResult<PySysUnlinkResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("delete") {
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let dhc = hc.delete.bind(py).call1((path, py_ctx))?.unbind();
            self.dispatch_pre_hooks_inner("delete", &dhc)?;
        }

        // 2. Call pure Rust kernel
        let rust_ctx = ctx.to_rust();
        let result = py.detach(|| self.inner.sys_unlink_one(path, &rust_ctx, recursive));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysUnlinkResult {
            hit: result.hit,
            entry_type: result.entry_type,
            post_hook_needed: result.post_hook_needed,
            path: result.path,
            content_id: result.content_id,
            size: result.size,
        })
    }

    // ── sys_rename ────────────────────────────────────────────────────

    #[pyo3(signature = (old_path, new_path, ctx))]
    fn sys_rename(
        &self,
        py: Python<'_>,
        old_path: &str,
        new_path: &str,
        ctx: &PyOperationContext,
    ) -> PyResult<PySysRenameResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("rename") {
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let rhc = hc
                .rename
                .bind(py)
                .call1((old_path, new_path, py_ctx))?
                .unbind();
            self.dispatch_pre_hooks_inner("rename", &rhc)?;
        }

        // 2. Call pure Rust kernel
        let rust_ctx = ctx.to_rust();
        let result = py.detach(|| self.inner.sys_rename(old_path, new_path, &rust_ctx));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysRenameResult {
            hit: result.hit,
            success: result.success,
            post_hook_needed: result.post_hook_needed,
            is_directory: result.is_directory,
            old_content_id: result.old_content_id,
            old_size: result.old_size,
            old_version: result.old_version,
            old_modified_at_ms: result.old_modified_at_ms,
        })
    }

    // ── sys_copy ──────────────────────────────────────────────────────

    #[pyo3(signature = (src_path, dst_path, ctx))]
    fn sys_copy(
        &self,
        py: Python<'_>,
        src_path: &str,
        dst_path: &str,
        ctx: &PyOperationContext,
    ) -> PyResult<PySysCopyResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("copy") {
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let chc = hc
                .copy
                .bind(py)
                .call1((src_path, dst_path, py_ctx))?
                .unbind();
            self.dispatch_pre_hooks_inner("copy", &chc)?;
        }

        // 2. Call pure Rust kernel
        let rust_ctx = ctx.to_rust();
        let result = py.detach(|| self.inner.sys_copy(src_path, dst_path, &rust_ctx));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysCopyResult {
            hit: result.hit,
            post_hook_needed: result.post_hook_needed,
            dst_path: result.dst_path,
            content_id: result.content_id,
            size: result.size,
            version: result.version,
            gen: result.gen,
        })
    }

    // ── sys_mkdir ─────────────────────────────────────────────────────

    #[pyo3(signature = (path, ctx, parents=true, exist_ok=true))]
    fn sys_mkdir(
        &self,
        py: Python<'_>,
        path: &str,
        ctx: &PyOperationContext,
        parents: bool,
        exist_ok: bool,
    ) -> PyResult<PySysMkdirResult> {
        // 1. PRE-INTERCEPT hooks (GIL, abort on exception)
        if self.inner.has_hooks("mkdir") {
            let py_ctx = rust_ctx_to_python(py, &ctx.to_rust(), "")
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)?;
            let hc = get_hook_ctx_cache(py).ok_or_else(|| {
                pyo3::exceptions::PyRuntimeError::new_err("hook context cache init failed")
            })?;
            let mhc = hc.mkdir.bind(py).call1((path, py_ctx))?.unbind();
            self.dispatch_pre_hooks_inner("mkdir", &mhc)?;
        }

        // 2. Call pure Rust kernel (full mkdir)
        let rust_ctx = ctx.to_rust();
        let result = py.detach(|| self.inner.sys_mkdir(path, &rust_ctx, parents, exist_ok));
        let result = result.map_err(|e| -> PyErr { e.into() })?;

        Ok(PySysMkdirResult {
            hit: result.hit,
            post_hook_needed: result.post_hook_needed,
        })
    }

    // ── Tier 2 convenience methods ────────────────────────────────────

    #[pyo3(signature = (path, zone_id))]
    fn access(&self, path: &str, zone_id: &str) -> bool {
        self.inner.access(path, zone_id)
    }

    #[pyo3(signature = (parent_path, zone_id, is_admin=false))]
    fn readdir(&self, parent_path: &str, zone_id: &str, is_admin: bool) -> Vec<(String, u8)> {
        self.inner.readdir(parent_path, zone_id, is_admin)
    }

    /// Phase 6: glob match against the metastore-recursive listing of
    /// `prefix`.  Replaces `nexus.fs._helpers.glob` — pure Rust, no
    /// Python fallback (`lib::glob::glob_match` covers the same
    /// `globset` syntax the Python `fnmatch` fallback used).
    #[pyo3(signature = (pattern, prefix="/", zone_id="root"))]
    fn sys_glob(&self, pattern: &str, prefix: &str, zone_id: &str) -> PyResult<Vec<String>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        self.inner
            .sys_glob(pattern, prefix, &ctx)
            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("sys_glob: {:?}", e)))
    }

    /// Phase 6: grep recursive — walk every regular file under
    /// `prefix`, scan content via `lib::search::search_lines`, return
    /// up to `max_results` matches.  Replaces `nexus.fs._helpers.grep`.
    /// Each match comes back as a `dict` matching the historical
    /// shape: `{file, line, content, match}`.
    ///
    /// When `disk_paths` is non-empty the metastore walk is skipped:
    /// the kernel reads each absolute path from disk directly.  Used
    /// by the search-tier cache fast path where the cached blob's
    /// on-disk location is already known.
    #[pyo3(signature = (pattern, prefix="/", ignore_case=false, max_results=1000, zone_id="root", disk_paths=None))]
    #[allow(clippy::too_many_arguments)]
    fn sys_grep<'py>(
        &self,
        py: Python<'py>,
        pattern: &str,
        prefix: &str,
        ignore_case: bool,
        max_results: usize,
        zone_id: &str,
        disk_paths: Option<Vec<String>>,
    ) -> PyResult<Bound<'py, PyList>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        let paths = disk_paths.unwrap_or_default();
        let matches = self
            .inner
            .sys_grep(pattern, prefix, ignore_case, max_results, &paths, &ctx)
            .map_err(|e| pyo3::exceptions::PyRuntimeError::new_err(format!("sys_grep: {:?}", e)))?;
        let out = PyList::empty(py);
        for m in matches {
            let d = PyDict::new(py);
            d.set_item("file", m.file)?;
            d.set_item("line", m.line)?;
            d.set_item("content", m.content)?;
            d.set_item("match", m.match_text)?;
            out.append(d)?;
        }
        Ok(out)
    }

    /// Simplified sys_read that takes (path, zone_id) — creates a minimal
    /// OperationContext internally.  Used by service-tier callers that don't
    /// have a full OperationContext handy.
    fn sys_read_raw<'py>(&self, py: Python<'py>, path: &str, zone_id: &str) -> PyResult<Py<PyAny>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        let result = self.inner.sys_read_one(path, &ctx, 5000, 0).map_err(|e| {
            pyo3::exceptions::PyRuntimeError::new_err(format!("sys_read_raw: {:?}", e))
        })?;
        match result.data {
            Some(bytes) => Ok(pyo3::types::PyBytes::new(py, &bytes).into()),
            None => Err(pyo3::exceptions::PyFileNotFoundError::new_err(format!(
                "File not found: {}",
                path
            ))),
        }
    }

    #[pyo3(signature = (path, zone_id="root", strict_json=true))]
    fn sys_cat<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        zone_id: &str,
        strict_json: bool,
    ) -> PyResult<Py<PyBytes>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        let result = self
            .inner
            .sys_cat(path, &ctx, strict_json)
            .map_err(|e| -> PyErr { e.into() })?;
        Ok(PyBytes::new(py, &result.data).unbind())
    }

    #[pyo3(signature = (path, zone_id="root"))]
    fn op_metadata_for_path<'py>(
        &self,
        py: Python<'py>,
        path: &str,
        zone_id: &str,
    ) -> PyResult<Bound<'py, PyDict>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        let result = self
            .inner
            .op_metadata_for_path(path, &ctx)
            .map_err(|e| -> PyErr { e.into() })?;
        let dict = PyDict::new(py);
        dict.set_item("filetype", result.filetype.as_str())?;
        dict.set_item("backend", result.backend.as_str())?;
        dict.set_item("mime_type", result.mime_type.as_deref())?;
        dict.set_item("backend_name", result.backend_name)?;
        Ok(dict)
    }

    #[pyo3(signature = (path, zone_id="root"))]
    fn backend_fingerprint(&self, path: &str, zone_id: &str) -> PyResult<Option<String>> {
        let ctx = OperationContext::new("system", zone_id, true, None, true);
        self.inner
            .backend_fingerprint(path, &ctx)
            .map_err(|e| -> PyErr { e.into() })
    }

    // ── Batch syscalls ─────────────────────────────────────────────────

    /// Batch write: validate + route + lock + write + metastore + dcache.
    /// Returns list of SysWriteResult (one per item).
    #[pyo3(signature = (items, ctx))]
    fn sys_write_batch<'py>(
        &self,
        py: Python<'py>,
        items: Vec<(String, Vec<u8>)>,
        ctx: &PyOperationContext,
    ) -> PyResult<Vec<PySysWriteResult>> {
        let rust_ctx = ctx.to_rust();
        let reqs: Vec<crate::kernel::WriteRequest> = items
            .into_iter()
            .map(|(path, content)| crate::kernel::WriteRequest {
                path,
                content,
                offset: 0,
            })
            .collect();
        let results = py.detach(|| self.inner.sys_write(&reqs, &rust_ctx));
        Ok(results
            .into_iter()
            .map(|r| {
                let r = match r {
                    Ok(r) => r,
                    Err(_) => crate::kernel::SysWriteResult {
                        hit: false,
                        content_id: None,
                        post_hook_needed: false,
                        version: 0,
                        gen: 0,
                        size: 0,
                        is_new: false,
                        old_content_id: None,
                        old_size: None,
                        old_version: None,
                        old_modified_at_ms: None,
                    },
                };
                PySysWriteResult {
                    hit: r.hit,
                    content_id: r.content_id,
                    post_hook_needed: r.post_hook_needed,
                    version: r.version,
                    gen: r.gen,
                    size: r.size,
                    is_new: r.is_new,
                    old_content_id: r.old_content_id,
                    old_size: r.old_size,
                    old_version: r.old_version,
                    old_modified_at_ms: r.old_modified_at_ms,
                }
            })
            .collect())
    }

    /// Batch read. Accepts either `list[str]` (legacy) or
    /// `list[tuple[str, int, int | None]]`. Returns `list[PyBatchReadItem]`.
    #[pyo3(signature = (reqs, ctx))]
    fn sys_read_batch<'py>(
        &self,
        py: Python<'py>,
        reqs: Bound<'py, PyAny>,
        ctx: &PyOperationContext,
    ) -> PyResult<Vec<PyBatchReadItem>> {
        // Parse either shape: list[str] (legacy) or list[(str, int, int|None)].
        let rust_reqs: Vec<crate::kernel::ReadRequest> =
            if let Ok(paths) = reqs.extract::<Vec<String>>() {
                paths
                    .into_iter()
                    .map(|p| crate::kernel::ReadRequest {
                        path: p,
                        offset: 0,
                        len: None,
                        timeout_ms: 5000,
                    })
                    .collect()
            } else {
                let tuples: Vec<(String, u64, Option<u64>)> = reqs.extract()?;
                tuples
                    .into_iter()
                    .map(|(p, off, len)| crate::kernel::ReadRequest {
                        path: p,
                        offset: off,
                        len,
                        timeout_ms: 5000,
                    })
                    .collect()
            };
        let rust_ctx = ctx.to_rust();
        let results = py.detach(|| self.inner.sys_read(&rust_reqs, &rust_ctx));
        Ok(results
            .into_iter()
            .map(|r| match r {
                Ok(r) => PyBatchReadItem {
                    data: r.data.map(|d| PyBytes::new(py, &d).into()),
                    content_id: r.content_id,
                    gen: r.gen,
                    entry_type: r.entry_type,
                    post_hook_needed: r.post_hook_needed,
                    error_kind: String::new(),
                    error_message: String::new(),
                },
                Err(e) => {
                    let (kind, msg) = crate::batch_read_py::batch_err_kind_msg(&e);
                    PyBatchReadItem {
                        data: None,
                        content_id: None,
                        gen: 0,
                        entry_type: 0,
                        post_hook_needed: false,
                        error_kind: kind,
                        error_message: msg,
                    }
                }
            })
            .collect())
    }

    /// Batch delete: loops sys_unlink for each path.
    /// Returns list of SysUnlinkResult (one per path).
    #[pyo3(signature = (paths, ctx))]
    fn sys_unlink_batch(
        &self,
        py: Python<'_>,
        paths: Vec<String>,
        ctx: &PyOperationContext,
    ) -> PyResult<Vec<PySysUnlinkResult>> {
        let rust_ctx = ctx.to_rust();
        let reqs: Vec<crate::kernel::UnlinkRequest> = paths
            .into_iter()
            .map(|path| crate::kernel::UnlinkRequest {
                path,
                recursive: false,
            })
            .collect();
        let results = py.detach(|| self.inner.sys_unlink(&reqs, &rust_ctx));
        Ok(results
            .into_iter()
            .map(|r| {
                let r = match r {
                    Ok(r) => r,
                    Err(_) => crate::kernel::SysUnlinkResult {
                        hit: false,
                        entry_type: 0,
                        post_hook_needed: false,
                        path: String::new(),
                        content_id: None,
                        size: 0,
                    },
                };
                PySysUnlinkResult {
                    hit: r.hit,
                    entry_type: r.entry_type,
                    post_hook_needed: r.post_hook_needed,
                    path: r.path,
                    content_id: r.content_id,
                    size: r.size,
                }
            })
            .collect())
    }

    // ── Zone revision counter (§10 A2) ──────────────────────────────

    /// Increment zone revision (called after successful metastore write).
    fn increment_zone_revision(&self, zone_id: &str) -> u64 {
        self.inner.increment_zone_revision(zone_id)
    }

    /// Notify a specific zone revision (monotonic update).
    fn notify_zone_revision(&self, zone_id: &str, revision: u64) {
        self.inner.notify_zone_revision(zone_id, revision)
    }

    /// Get current zone revision (0 if unknown).
    fn get_zone_revision(&self, zone_id: &str) -> u64 {
        self.inner.get_zone_revision(zone_id)
    }

    /// Wait until zone revision >= min_revision, or timeout.
    /// Releases GIL during condvar wait (pure Rust blocking).
    fn wait_zone_revision(
        &self,
        py: Python<'_>,
        zone_id: &str,
        min_revision: u64,
        timeout_ms: u64,
    ) -> bool {
        let zone = zone_id.to_string();
        py.detach(|| {
            self.inner
                .wait_zone_revision(&zone, min_revision, timeout_ms)
        })
    }

    // ── File watch registry (§10 A3) ────────────────────────────────

    /// Register a glob pattern watch. Returns watch ID.
    fn register_watch(&self, pattern: &str) -> u64 {
        self.inner.register_watch(pattern)
    }

    /// Unregister a file watch by ID.
    fn unregister_watch(&self, watch_id: u64) -> bool {
        self.inner.unregister_watch(watch_id)
    }

    /// Match a path against all registered watch patterns.
    fn match_watches(&self, path: &str) -> Vec<u64> {
        self.inner.match_watches(path)
    }

    // ── Agent registry (§10 B1) ─────────────────────────────────────

    /// Sub-handle exposing the kernel `AgentRegistry` SSOT to
    /// Python callers. Wraps the kernel `Arc<AgentRegistry>`
    /// — every call returns a fresh wrapper sharing the same
    /// state, so Python code can write
    /// `kernel.agent_registry.spawn(...)` without manual Arc
    /// management.
    #[getter]
    fn agent_registry(&self) -> crate::agent_registry_py::PyAgentRegistry {
        crate::agent_registry_py::from_kernel(&self.inner)
    }

    /// Register a new agent. Returns true if inserted (pid was new).
    #[pyo3(signature = (pid, name, kind, owner_id, zone_id, created_at_ms, parent_pid=None, connection_id=None))]
    #[allow(clippy::too_many_arguments)]
    fn agent_register(
        &self,
        pid: &str,
        name: &str,
        kind: &str,
        owner_id: &str,
        zone_id: &str,
        created_at_ms: u64,
        parent_pid: Option<&str>,
        connection_id: Option<&str>,
    ) -> bool {
        use crate::core::agents::registry::{AgentDescriptor, AgentKind};
        let kind = AgentKind::from_str(kind).unwrap_or(AgentKind::Worker);
        self.inner.agent_registry.register(AgentDescriptor {
            pid: pid.to_string(),
            name: name.to_string(),
            kind,
            owner_id: owner_id.to_string(),
            zone_id: zone_id.to_string(),
            created_at_ms,
            updated_at_ms: created_at_ms,
            parent_pid: parent_pid.map(|s| s.to_string()),
            connection_id: connection_id.map(|s| s.to_string()),
            ..Default::default()
        })
    }

    /// Unregister an agent by pid.
    fn agent_unregister(&self, pid: &str) -> bool {
        self.inner.agent_registry.unregister(pid).is_some()
    }

    /// Get agent descriptor as dict (legacy; new callers should
    /// use `kernel.agent_registry.get(pid)`).
    fn agent_get<'py>(&self, py: Python<'py>, pid: &str) -> PyResult<Option<Bound<'py, PyDict>>> {
        match self.inner.agent_registry.get(pid) {
            Some(desc) => {
                let dict = PyDict::new(py);
                dict.set_item("pid", &desc.pid)?;
                dict.set_item("name", &desc.name)?;
                dict.set_item("kind", desc.kind.as_str().to_ascii_lowercase())?;
                dict.set_item("state", desc.state.as_str().to_ascii_lowercase())?;
                dict.set_item("owner_id", &desc.owner_id)?;
                dict.set_item("zone_id", &desc.zone_id)?;
                dict.set_item("created_at_ms", desc.created_at_ms)?;
                dict.set_item("updated_at_ms", desc.updated_at_ms)?;
                dict.set_item("exit_code", desc.exit_code)?;
                dict.set_item("generation", desc.generation)?;
                dict.set_item("cwd", &desc.cwd)?;
                dict.set_item("root", &desc.root)?;
                dict.set_item("parent_pid", desc.parent_pid.as_deref())?;
                dict.set_item("connection_id", desc.connection_id.as_deref())?;
                dict.set_item("last_heartbeat_ms", desc.last_heartbeat_ms)?;
                dict.set_item("children", desc.children.clone())?;
                dict.set_item("labels", desc.labels.clone())?;
                Ok(Some(dict))
            }
            None => Ok(None),
        }
    }

    /// Update agent state. Returns true if found, false if pid
    /// missing. Raises ValueError on invalid transitions.
    fn agent_update_state(&self, pid: &str, new_state: &str) -> PyResult<bool> {
        use crate::core::agents::registry::AgentState;
        let state = AgentState::from_str(new_state).ok_or_else(|| {
            pyo3::exceptions::PyValueError::new_err(format!("unknown agent state: {new_state}"))
        })?;
        self.inner
            .agent_registry
            .update_state(pid, state)
            .map_err(|e| pyo3::exceptions::PyValueError::new_err(e.to_string()))
    }

    /// List all agents. Returns list of dicts.
    #[pyo3(signature = (zone_id=None, owner_id=None, state=None, kind=None))]
    fn agent_list<'py>(
        &self,
        py: Python<'py>,
        zone_id: Option<&str>,
        owner_id: Option<&str>,
        state: Option<&str>,
        kind: Option<&str>,
    ) -> PyResult<Vec<Bound<'py, PyDict>>> {
        use crate::core::agents::registry::{AgentKind, AgentState};
        let state_filter = state.and_then(AgentState::from_str);
        let kind_filter = kind.and_then(AgentKind::from_str);
        let agents = self.inner.agent_registry.list(
            zone_id,
            owner_id,
            kind_filter.as_ref(),
            state_filter.as_ref(),
        );
        let mut result = Vec::with_capacity(agents.len());
        for desc in agents {
            let dict = PyDict::new(py);
            dict.set_item("pid", &desc.pid)?;
            dict.set_item("name", &desc.name)?;
            dict.set_item("kind", desc.kind.as_str().to_ascii_lowercase())?;
            dict.set_item("state", desc.state.as_str().to_ascii_lowercase())?;
            dict.set_item("owner_id", &desc.owner_id)?;
            dict.set_item("zone_id", &desc.zone_id)?;
            dict.set_item("created_at_ms", desc.created_at_ms)?;
            result.push(dict);
        }
        Ok(result)
    }

    /// Update heartbeat timestamp for an agent (raw stamp; no
    /// kind/info validation — see ``AgentRegistry.heartbeat`` for
    /// the validating variant).
    fn agent_heartbeat(&self, pid: &str, timestamp_ms: u64) -> bool {
        self.inner.agent_registry.heartbeat_at(pid, timestamp_ms)
    }

    /// Get number of registered agents.
    fn agent_count(&self) -> usize {
        self.inner.agent_registry.count()
    }

    /// Block (GIL-free) until agent `pid` reaches `target_state` or timeout.
    ///
    /// Returns the state string on success. Raises `RuntimeError` on
    /// timeout ("timeout") or unknown pid ("not_found").
    fn agent_wait(
        &self,
        py: Python<'_>,
        pid: &str,
        target_state: &str,
        timeout_ms: u64,
    ) -> PyResult<String> {
        use crate::core::agents::registry::AgentState;
        let target = AgentState::from_str(target_state).ok_or_else(|| {
            pyo3::exceptions::PyValueError::new_err(format!("unknown agent state: {target_state}"))
        })?;
        let pid = pid.to_string();
        let registry = std::sync::Arc::clone(&self.inner.agent_registry);
        py.detach(|| {
            registry
                .wait_for_state(&pid, &target, timeout_ms)
                .map_err(pyo3::exceptions::PyRuntimeError::new_err)
        })
    }

    // ── Service registry ─────────────────────────────────────────────

    /// Register a service. Wraps the Python instance in PyServiceLifecycle.
    #[pyo3(signature = (name, instance, exports=vec![], allow_overwrite=false))]
    fn service_enlist(
        &self,
        _py: Python<'_>,
        name: &str,
        instance: &Bound<'_, PyAny>,
        exports: Vec<String>,
        allow_overwrite: bool,
    ) -> PyResult<()> {
        // Validate exports on the Python side before wrapping
        for exp in &exports {
            if !instance.hasattr(exp.as_str())? {
                return Err(pyo3::exceptions::PyValueError::new_err(format!(
                    "services: {name:?} declares exports not found on instance: [{exp}]"
                )));
            }
        }
        let lifecycle = Box::new(PyServiceLifecycle(instance.clone().unbind()));
        self.inner
            .register_managed_service(name, lifecycle, exports, allow_overwrite)
            .map_err(pyo3::exceptions::PyValueError::new_err)
    }

    /// Look up a service by name. Returns the raw Python instance or None.
    fn service_lookup(&self, py: Python<'_>, name: &str) -> Option<Py<PyAny>> {
        let lifecycle = self.inner.service_lookup_managed(name)?;
        let adapter = (&*lifecycle as &dyn std::any::Any).downcast_ref::<PyServiceLifecycle>()?;
        Some(adapter.0.clone_ref(py))
    }

    /// Hot-swap a service: drain → replace → rehook.
    #[pyo3(signature = (name, new_instance, exports=vec![], timeout_ms=10000))]
    fn service_swap(
        &self,
        _py: Python<'_>,
        name: &str,
        new_instance: &Bound<'_, PyAny>,
        exports: Vec<String>,
        timeout_ms: u64,
    ) -> PyResult<()> {
        // Validate exports
        for exp in &exports {
            if !new_instance.hasattr(exp.as_str())? {
                return Err(pyo3::exceptions::PyValueError::new_err(format!(
                    "services: {name:?} replacement declares invalid exports: [{exp}]"
                )));
            }
        }
        let lifecycle = Box::new(PyServiceLifecycle(new_instance.clone().unbind()));
        self.inner
            .swap_managed_service(name, lifecycle, exports, timeout_ms)
            .map_err(pyo3::exceptions::PyKeyError::new_err)
    }

    /// Unregister a service.
    fn service_unregister(&self, name: &str) -> bool {
        self.inner.unregister_service(name)
    }

    /// Start all BackgroundService instances.
    #[pyo3(signature = (timeout_ms=30000))]
    fn service_start_all(&self, timeout_ms: u64) -> PyResult<Vec<String>> {
        self.inner
            .service_start_all(timeout_ms as f64 / 1000.0)
            .map_err(pyo3::exceptions::PyRuntimeError::new_err)
    }

    /// Stop all BackgroundService instances.
    #[pyo3(signature = (timeout_ms=10000))]
    fn service_stop_all(&self, timeout_ms: u64) -> PyResult<Vec<String>> {
        self.inner
            .service_stop_all(timeout_ms as f64 / 1000.0)
            .map_err(pyo3::exceptions::PyRuntimeError::new_err)
    }

    /// Close all services with close() method.
    fn service_close_all(&self) {
        self.inner.service_close_all()
    }

    /// Mark bootstrap complete.
    fn service_mark_bootstrapped(&self) {
        self.inner.service_mark_bootstrapped()
    }

    /// Diagnostic snapshot: list of dicts.
    fn service_snapshot<'py>(&self, py: Python<'py>) -> PyResult<Vec<Bound<'py, PyDict>>> {
        let entries = self.inner.service_snapshot();
        let mut result = Vec::with_capacity(entries.len());
        for (name, type_name, exports) in entries {
            let dict = PyDict::new(py);
            dict.set_item("name", &name)?;
            dict.set_item("type", &type_name)?;
            let exports_list: Vec<&str> = exports.iter().map(|s| s.as_str()).collect();
            dict.set_item("exports", exports_list)?;
            result.push(dict);
        }
        Ok(result)
    }
}

// ── Private: hook dispatch (wrapper-only) ───────────────────────────────

impl PyKernel {
    /// Internal pre-hook dispatch (used by Tier 1 syscalls).
    fn dispatch_pre_hooks_inner(&self, op: &str, hook_ctx: &Py<PyAny>) -> PyResult<()> {
        let hooks = self.hooks.read_unconditional();
        let impls = hooks.get_pre_hook_impls(op);
        for hook in impls {
            match op {
                "read" => hook.on_pre_read(hook_ctx)?,
                "write" => hook.on_pre_write(hook_ctx)?,
                "delete" => hook.on_pre_delete(hook_ctx)?,
                "rename" => hook.on_pre_rename(hook_ctx)?,
                "mkdir" => hook.on_pre_mkdir(hook_ctx)?,
                "rmdir" => hook.on_pre_rmdir(hook_ctx)?,
                "copy" => hook.on_pre_copy(hook_ctx)?,
                "stat" => hook.on_pre_stat(hook_ctx)?,
                "access" => hook.on_pre_access(hook_ctx)?,
                "write_batch" => hook.on_pre_write_batch(hook_ctx)?,
                _ => {}
            }
        }
        Ok(())
    }

    /// Dispatch sync post-hooks via Rust InterceptHook trait.
    ///
    /// Runs sync post-hooks serially, fault-isolated (fire-and-forget).
    /// Async hooks are NOT handled here — Python thin wrapper schedules
    /// them via asyncio.gather (codegen or manual).
    fn dispatch_post_hooks_sync(&self, op: &str, hook_ctx: &Py<PyAny>) {
        let hooks = self.hooks.read_unconditional();
        let impls = hooks.get_post_hook_impls(op);
        for hook in impls {
            match op {
                "read" => hook.on_post_read(hook_ctx),
                "write" => hook.on_post_write(hook_ctx),
                "delete" => hook.on_post_delete(hook_ctx),
                "rename" => hook.on_post_rename(hook_ctx),
                "mkdir" => hook.on_post_mkdir(hook_ctx),
                "rmdir" => hook.on_post_rmdir(hook_ctx),
                "copy" => hook.on_post_copy(hook_ctx),
                "stat" => hook.on_post_stat(hook_ctx),
                "access" => hook.on_post_access(hook_ctx),
                "write_batch" => hook.on_post_write_batch(hook_ctx),
                _ => {}
            }
        }
    }
}
