//! Daemon-side handlers for the `MAILBOX-CREATE` and `MAILBOX-DELETE`
//! verbs of the `AIMX/1` UDS protocol.
//!
//! `aimx mailboxes create` / `delete` route through the daemon over UDS so
//! the in-memory `Config` is hot-swapped under a `RwLock<Arc<Config>>`
//! whenever `config.toml` on disk changes. Inbound mail to a just-created
//! mailbox routes correctly on the very next SMTP session. No restart,
//! no "silent fall through to catchall" surprise.
//!
//! Correctness model:
//!
//! 1. Validate the mailbox name using the same `mailbox::validate_mailbox_name`
//!    rules every caller already trusts.
//! 2. Load the *current* `Config` snapshot through the shared
//!    [`ConfigHandle`]. Re-derive the new snapshot in memory (either
//!    inserting or removing the stanza). Fail fast on duplicates /
//!    missing / attempts to delete catchall / non-empty directories.
//! 3. Write the new `Config` to disk **atomically** via
//!    `write-temp-then-rename`. Either we see the old snapshot or the
//!    new one, never a partial write.
//! 4. Only after the rename succeeds, swap the in-memory `Config` via
//!    [`ConfigHandle::store`]. This ordering matters: if we swapped
//!    first and the rename failed, we would be handing inbound mail a
//!    routing table that doesn't match disk.
//!
//! Locking: acquired in the order the
//! [`crate::mailbox_locks`] module documents. The **outer** per-mailbox
//! `tokio::sync::Mutex<()>` (shared with inbound ingest and MARK-*) is
//! taken for the duration of the write so no other writer on the same
//! mailbox can see the file system half-created / half-deleted. The
//! **inner** process-wide [`CONFIG_WRITE_LOCK`] serializes the
//! `load → modify → write → store` critical section across *all* mailbox
//! names. Without it, two concurrent `MAILBOX-CREATE alice` +
//! `MAILBOX-CREATE bob` requests hold disjoint per-mailbox locks and can
//! interleave their loads + writes, clobbering one stanza on disk and in
//! memory. Always outer → inner.

use std::os::unix::fs::MetadataExt;
use std::path::{Path, PathBuf};
use std::sync::Mutex;

use crate::auth::{Action, AuthError, authorize};
use crate::config::{Config, ConfigHandle, MailboxConfig};
use crate::ownership::chown_as_owner;
use crate::send_protocol::{AckResponse, ErrCode, MailboxLifecycleRequest};
use crate::state_handler::StateContext;
use crate::uds_authz::{Caller, log_decision, peer_username};

/// Process-wide mutex around the `config.toml` read-modify-write critical
/// section. Symmetric in spirit to `send_handler::SENT_WRITE_LOCK`: a
/// single short-lived `std::sync::Mutex` that serializes *only* the
/// load-modify-write-store sequence so two concurrent
/// `MAILBOX-CREATE`/`DELETE` requests on different mailbox names cannot
/// clobber each other. Held only for the duration of the critical
/// section (not across the full request lifetime).
pub(crate) static CONFIG_WRITE_LOCK: Mutex<()> = Mutex::new(());

/// Shared context for MAILBOX-CREATE / MAILBOX-DELETE. Kept separate from
/// `StateContext` so the state-lock map is not accidentally bypassed (the
/// handler calls through `state_ctx.lock_for(...)` before touching disk).
pub struct MailboxContext {
    /// Location of `config.toml` on disk. Injected rather than taken from
    /// `crate::config::config_path()` so tests can point at a tempdir
    /// without mutating the process-wide `AIMX_CONFIG_DIR` env var.
    pub config_path: PathBuf,
    /// Live handle to the daemon's `Config`. Written through on success.
    pub config_handle: ConfigHandle,
}

impl MailboxContext {
    pub fn new(config_path: PathBuf, config_handle: ConfigHandle) -> Self {
        Self {
            config_path,
            config_handle,
        }
    }
}

/// Dispatch a `MAILBOX-CREATE` or `MAILBOX-DELETE` request to the matching
/// handler. Holds the same per-mailbox lock the MARK-* path takes so a
/// concurrent MARK on the same mailbox cannot observe a half-created or
/// half-deleted file system.
pub async fn handle_mailbox_crud(
    state_ctx: &StateContext,
    mb_ctx: &MailboxContext,
    req: &MailboxLifecycleRequest,
    caller: &Caller,
) -> AckResponse {
    if let Err(e) = crate::mailbox::validate_mailbox_name(&req.name) {
        return AckResponse::Err {
            code: ErrCode::Validation,
            reason: e,
        };
    }

    let verb = if req.create {
        "MAILBOX-CREATE"
    } else {
        "MAILBOX-DELETE"
    };

    // CREATE owner resolution runs *before* the lock acquisition so the
    // orphan-uid failure path doesn't take and immediately drop a
    // per-mailbox lock for nothing. Root callers honor `req.owner`
    // exactly as today (cross-uid creates remain operator-only because
    // no other surface accepts an `Owner:` parameter from non-root).
    // Non-root callers have `req.owner` ignored on the floor — the
    // owner is synthesized from `SO_PEERCRED` so privilege escalation
    // is structurally impossible.
    let resolved_owner: Option<String> = if req.create {
        if caller.is_root() {
            // Root path: the dispatcher already enforces that
            // MailboxLifecycleRequest carries `owner: Some(...)` for
            // CREATE; defense-in-depth check below.
            match req.owner.as_deref() {
                Some(o) => Some(o.to_string()),
                None => {
                    return AckResponse::Err {
                        code: ErrCode::Validation,
                        reason: "MAILBOX-CREATE requires an Owner: header".into(),
                    };
                }
            }
        } else {
            // Non-root path: ignore any wire-supplied owner; the daemon
            // is the authoritative source. `peer_username` errs (does
            // not panic) on orphan uids — surface as Validation so the
            // wire response shape stays consistent with the rest of
            // the validation failures.
            match peer_username(caller.uid) {
                Ok(name) => Some(name),
                Err(reason) => {
                    log_decision(
                        verb,
                        caller,
                        Some(&req.name),
                        crate::uds_authz::LogDecision::Reject,
                        Some(&reason),
                    );
                    return AckResponse::Err {
                        code: ErrCode::Validation,
                        reason,
                    };
                }
            }
        }
    } else {
        None
    };

    let state = state_ctx.lock_for(&req.name);
    let _guard = state.lock.lock().await;

    // Serialize the load → modify → write → store critical section across
    // *all* mailbox names. The per-mailbox lock above keeps MARK-* on the
    // same mailbox from racing; this global lock keeps
    // `MAILBOX-CREATE alice` + `MAILBOX-CREATE bob` from losing one of
    // the two stanzas via interleaved load-write-store sequences.
    let _config_guard = CONFIG_WRITE_LOCK
        .lock()
        .unwrap_or_else(|poisoned| poisoned.into_inner());

    if req.create {
        // Predicate sanity check: for non-root callers the synthesized
        // owner_uid is always the caller's own uid, so the predicate
        // trivially passes; for root callers the bypass passes
        // unconditionally. The check is kept so future refactors
        // can't decouple the wire owner from the authorized owner
        // without tripping the central predicate.
        if let Err(e) = authorize(
            caller.uid,
            Action::MailboxCreate {
                owner_uid: caller.uid,
            },
            None,
        ) {
            log_decision(
                verb,
                caller,
                Some(&req.name),
                crate::uds_authz::LogDecision::Reject,
                Some(&format!("{e}")),
            );
            return AckResponse::Err {
                code: ErrCode::Eaccess,
                reason: not_authorized_wire_reason(),
            };
        }
        let owner = resolved_owner.expect("CREATE path resolved an owner above");
        log_decision(
            verb,
            caller,
            Some(&req.name),
            if caller.is_root() {
                crate::uds_authz::LogDecision::RootBypass
            } else {
                crate::uds_authz::LogDecision::Accept
            },
            None,
        );
        handle_create(state_ctx, mb_ctx, &req.name, &owner)
    } else {
        // DELETE owner check: non-root callers must own the target
        // mailbox. Resolve against the live `Arc<Config>` snapshot;
        // the lock above keeps the snapshot stable for the rest of
        // this critical section. Both the "no such mailbox" and
        // "mailbox exists but caller doesn't own it" paths must
        // produce a string-identical wire response so the daemon
        // doesn't leak whether the mailbox exists.
        let current = mb_ctx.config_handle.load();
        let mb_cfg = current.mailboxes.get(&req.name);
        if let Err(e) = authorize(
            caller.uid,
            Action::MailboxDelete {
                mailbox: req.name.clone(),
            },
            mb_cfg,
        ) {
            // Map both `NotOwner` and `NoSuchMailbox` to the same wire
            // response — the no-such-mailbox text the existing
            // handler already returns. NFR2: no information leak.
            let reason = match e {
                AuthError::NotOwner { .. } | AuthError::NoSuchMailbox => {
                    no_such_mailbox_reason(&req.name)
                }
                // `OwnerMismatch` only fires from `MailboxCreate`, so it
                // is unreachable on the DELETE path. Kept exhaustive so
                // a future variant addition surfaces here as a compile
                // error rather than silently mapping to the catch-all.
                AuthError::OwnerMismatch { .. } | AuthError::NotRoot => {
                    not_authorized_wire_reason()
                }
            };
            log_decision(
                verb,
                caller,
                Some(&req.name),
                crate::uds_authz::LogDecision::Reject,
                Some(&reason),
            );
            return AckResponse::Err {
                code: ErrCode::Mailbox,
                reason,
            };
        }
        // Drop the borrow into `current` before calling handle_delete,
        // which re-acquires its own snapshot under the same lock.
        drop(current);
        log_decision(
            verb,
            caller,
            Some(&req.name),
            if caller.is_root() {
                crate::uds_authz::LogDecision::RootBypass
            } else {
                crate::uds_authz::LogDecision::Accept
            },
            None,
        );
        handle_delete(state_ctx, mb_ctx, &req.name, req.force)
    }
}

/// Canonical "no such mailbox" reason. Used by both the daemon's
/// genuine no-such-mailbox path in [`handle_delete`] and the non-root
/// "mailbox exists but caller doesn't own it" path so the wire
/// response is byte-identical for both. NFR2: the daemon never leaks
/// whether a mailbox exists to a caller who is not its owner.
fn no_such_mailbox_reason(name: &str) -> String {
    format!("mailbox '{name}' does not exist")
}

/// Canonical opaque "not authorized" wire reason. Used by the few
/// truly-not-authorized paths that aren't already routed through the
/// no-such-mailbox shape.
fn not_authorized_wire_reason() -> String {
    "not authorized".to_string()
}

/// Resolve `owner` via the same helpers used for `MailboxConfig`.
/// Using [`MailboxConfig::owner_uid`] / [`MailboxConfig::owner_gid`] keeps
/// the reserved-name short-circuits (`root` → uid 0 without hitting
/// `getpwnam`, `aimx-catchall` → routed through the resolver) in one code
/// path, so the validation rules the rest of the daemon trusts apply
/// identically here. The regex gate from `validate_run_as` also rejects
/// malformed inputs at this seam rather than at the `getpwnam` layer.
/// Returns an `EINVAL`-ready message on miss so the caller maps it
/// straight into `ErrCode::Validation`.
fn resolve_owner_ids(owner: &str) -> Result<(u32, u32), String> {
    let mb = MailboxConfig {
        address: String::new(),
        owner: owner.to_string(),
        hooks: vec![],
        trust: None,
        trusted_senders: None,
        allow_root_catchall: false,
    };
    let uid = mb
        .owner_uid()
        .map_err(|_| format!("unknown owner '{owner}'"))?;
    let gid = mb
        .owner_gid()
        .map_err(|_| format!("unknown owner '{owner}'"))?;
    Ok((uid, gid))
}

// The repeated `AckResponse::Err { code: ErrCode::Io, reason:
// format!(...) }` pattern inside `handle_create` is intentionally not
// collapsed into a small `io_err(path, op, e)` helper: the six Io-error
// sites carry four distinct operation verbs (`failed to create`,
// `failed to chown`, `failed to write`, `stat … failed`) and two of
// them include site-specific guidance (the chown site names the
// syscall, the rename site names the config path). A shared helper
// would either force the verbs into a uniform template (losing the
// distinct framing the messages currently carry) or accept a `&str`
// verb argument that reintroduces the same `format!` chain at every
// call site. The spelled-out pattern stays more scannable.
fn handle_create(
    state_ctx: &StateContext,
    mb_ctx: &MailboxContext,
    name: &str,
    owner: &str,
) -> AckResponse {
    let current = mb_ctx.config_handle.load();

    if current.mailboxes.contains_key(name) {
        return AckResponse::Err {
            code: ErrCode::Mailbox,
            reason: format!("mailbox '{name}' already exists"),
        };
    }

    // Resolve the owner uid/gid up front so we can both chown newly-
    // created dirs and compare against an already-existing dir's
    // uid/gid for the idempotent re-create path.
    let (uid, gid) = match resolve_owner_ids(owner) {
        Ok(ids) => ids,
        Err(reason) => {
            return AckResponse::Err {
                code: ErrCode::Validation,
                reason,
            };
        }
    };

    let data_dir = &state_ctx.data_dir;
    let inbox = data_dir.join("inbox").join(name);
    let sent = data_dir.join("sent").join(name);

    // Track whether each dir existed before this call so the rollback
    // branches below never `remove_dir` an operator-created directory.
    let inbox_existed = inbox.exists();
    let sent_existed = sent.exists();

    // Pre-existence check: if either dir already exists, the only safe
    // outcome is "already correct" (no-op) or "conflict" (wrong owner/
    // mode). A silent chown-over would mask an operator misconfig.
    if let Some(err) = check_preexisting_dirs(&inbox, &sent, uid, gid) {
        return err;
    }

    if let Err(e) = std::fs::create_dir_all(&inbox) {
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to create {}: {e}", inbox.display()),
        };
    }
    if let Err(e) = std::fs::create_dir_all(&sent) {
        // Roll back the inbox dir so we never leave half a mailbox behind.
        // Only remove if WE created it in this call.
        if !inbox_existed {
            let _ = std::fs::remove_dir(&inbox);
        }
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to create {}: {e}", sent.display()),
        };
    }

    // Build the new Config with the mailbox stanza inserted. The
    // resolved-side chown uses this MailboxConfig so a future
    // `config.toml` reload would still agree with the on-disk owner.
    let mut new_config: Config = (*current).clone();
    let address = format!("{name}@{}", new_config.domain);
    let mb_cfg = MailboxConfig {
        address,
        owner: owner.to_string(),
        hooks: vec![],
        trust: None,
        trusted_senders: None,
        allow_root_catchall: false,
    };
    new_config
        .mailboxes
        .insert(name.to_string(), mb_cfg.clone());

    // Chown + chmod the freshly-created dirs. Mode 0o700 means only
    // the owning user (and root) can traverse; this is the isolation
    // invariant from PRD §7.1. If either chown fails, roll back the
    // dirs we just created so we never leave half a mailbox with the
    // wrong owner. Dirs that existed before this call are left alone —
    // the idempotent-matching-owner path reaches here too, and removing
    // a pre-existing dir would delete an operator-created artifact.
    if let Err(e) = chown_as_owner(&inbox, &mb_cfg, 0o700) {
        rollback_self_created_dirs(&inbox, inbox_existed, &sent, sent_existed);
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to chown {}: {e}", inbox.display()),
        };
    }
    if let Err(e) = chown_as_owner(&sent, &mb_cfg, 0o700) {
        rollback_self_created_dirs(&inbox, inbox_existed, &sent, sent_existed);
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to chown {}: {e}", sent.display()),
        };
    }

    if let Err(e) = write_config_atomic(&mb_ctx.config_path, &new_config) {
        // Rename failed; leave the in-memory Config untouched so the
        // daemon continues running against the pre-call state on both
        // disk and memory. Best-effort clean up of the freshly-created
        // directories so the operator can retry cleanly. Dirs that
        // pre-existed are preserved.
        rollback_self_created_dirs(&inbox, inbox_existed, &sent, sent_existed);
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to write {}: {e}", mb_ctx.config_path.display()),
        };
    }

    mb_ctx.config_handle.store(new_config);
    AckResponse::Ok
}

/// Best-effort removal of dirs the handler itself created in this call.
/// Any dir that pre-existed (operator-created or left over from an
/// earlier idempotent run) is left untouched — see the callers in the
/// chown-failure and rename-failure rollback branches for the contract.
fn rollback_self_created_dirs(inbox: &Path, inbox_existed: bool, sent: &Path, sent_existed: bool) {
    if !inbox_existed {
        let _ = std::fs::remove_dir(inbox);
    }
    if !sent_existed {
        let _ = std::fs::remove_dir(sent);
    }
}

/// Returns `Some(err)` if either directory pre-exists with a uid/gid or
/// mode that doesn't match the expected `(uid, gid, 0o700)` triple. The
/// re-create path is idempotent: matching-uid/gid dirs are treated as
/// "already claimed" and bypass the creation and chown steps.
fn check_preexisting_dirs(inbox: &Path, sent: &Path, uid: u32, gid: u32) -> Option<AckResponse> {
    for dir in [inbox, sent] {
        match std::fs::symlink_metadata(dir) {
            Ok(meta) => {
                if !meta.is_dir() {
                    return Some(AckResponse::Err {
                        code: ErrCode::Conflict,
                        reason: format!(
                            "{} exists but is not a directory; remove it before creating the mailbox",
                            dir.display()
                        ),
                    });
                }
                let current_uid = meta.uid();
                let current_gid = meta.gid();
                if current_uid != uid || current_gid != gid {
                    return Some(AckResponse::Err {
                        code: ErrCode::Conflict,
                        reason: format!(
                            "{} already exists with owner {current_uid}:{current_gid}, \
                             expected {uid}:{gid}; run `chown -R {uid}:{gid} {path}` \
                             (and the matching sibling) and retry",
                            dir.display(),
                            path = dir.display()
                        ),
                    });
                }
            }
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
                // Not an error — this is the happy path, dir will be created.
            }
            Err(e) => {
                return Some(AckResponse::Err {
                    code: ErrCode::Io,
                    reason: format!("stat {} failed: {e}", dir.display()),
                });
            }
        }
    }
    None
}

fn handle_delete(
    state_ctx: &StateContext,
    mb_ctx: &MailboxContext,
    name: &str,
    force: bool,
) -> AckResponse {
    if name == "catchall" {
        return AckResponse::Err {
            code: ErrCode::Mailbox,
            reason: "cannot delete the catchall mailbox".into(),
        };
    }

    let current = mb_ctx.config_handle.load();
    if !current.mailboxes.contains_key(name) {
        return AckResponse::Err {
            code: ErrCode::Mailbox,
            reason: no_such_mailbox_reason(name),
        };
    }
    drop(current);

    let data_dir = &state_ctx.data_dir;
    let inbox = data_dir.join("inbox").join(name);
    let sent = data_dir.join("sent").join(name);

    // `force=true`: wipe inbox + sent contents under the same per-mailbox
    // lock that already guards the stanza removal. This eliminates the
    // data-destruction race window the client-side wipe used to leave
    // open (daemon dies between wipe and DELETE submit → contents gone,
    // config still references the mailbox). The wipe leaves the
    // directories themselves in place; the post-wipe NONEMPTY check
    // below sees zero files and lets the delete proceed.
    if force && let Err(e) = wipe_mailbox_dirs(&inbox, &sent) {
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to wipe mailbox '{name}': {e}"),
        };
    }

    let inbox_files = count_files_if_exists(&inbox);
    let sent_files = count_files_if_exists(&sent);
    if inbox_files + sent_files > 0 {
        return AckResponse::Err {
            code: ErrCode::NonEmpty,
            reason: format!(
                "mailbox '{name}' has {} files ({} in inbox, {} in sent); \
                 archive or remove them first",
                inbox_files + sent_files,
                inbox_files,
                sent_files
            ),
        };
    }

    let current = mb_ctx.config_handle.load();

    let mut new_config: Config = (*current).clone();
    new_config.mailboxes.remove(name);

    if let Err(e) = write_config_atomic(&mb_ctx.config_path, &new_config) {
        return AckResponse::Err {
            code: ErrCode::Io,
            reason: format!("failed to write {}: {e}", mb_ctx.config_path.display()),
        };
    }

    mb_ctx.config_handle.store(new_config);
    // Directories are left on disk (operator owns cleanup). The non-empty
    // check above guarantees they're empty, so the operator only ever has
    // to `rmdir` empty directories if they want to tidy up.
    AckResponse::Ok
}

/// Count the entries directly inside `dir`. Treats missing directory as
/// zero files. Does not recurse; the mailbox layout only writes at the
/// top level (flat `.md` files) and one level deep (bundle directories);
/// a non-empty bundle still trips the top-level count.
fn count_files_if_exists(dir: &Path) -> usize {
    match std::fs::read_dir(dir) {
        Ok(entries) => entries.filter_map(|e| e.ok()).count(),
        Err(_) => 0,
    }
}

/// Wipe both the inbox and sent directories of a mailbox, removing
/// every entry but leaving the directories themselves in place. Called
/// from the `MAILBOX-DELETE force=true` path so the wipe and the
/// stanza removal happen atomically under the per-mailbox lock — no
/// data-destruction race window if the daemon dies mid-flight.
/// Reuses [`crate::mailbox::wipe_mailbox_contents`] so the CLI
/// fallback path and the daemon path share the same wipe contract
/// (dotfile preservation, symlink handling, missing-dir tolerance).
fn wipe_mailbox_dirs(inbox: &Path, sent: &Path) -> Result<(), Box<dyn std::error::Error>> {
    crate::mailbox::wipe_mailbox_contents(inbox)?;
    crate::mailbox::wipe_mailbox_contents(sent)?;
    Ok(())
}

/// Atomically replace `path` with the TOML serialization of `config`.
///
/// Writes to a sibling `<path>.tmp.<pid>` file first, syncs it, then
/// renames over the target. On POSIX `rename(2)` is atomic for same-
/// filesystem targets, so readers see either the old snapshot or the new
/// one; never a truncated file. On failure the temp file is cleaned up
/// best-effort so subsequent retries don't trip over stale state.
///
/// **Unknown-key / comment behaviour (v1):** this function re-serializes
/// the `Config` struct through `toml::to_string_pretty`, which means any
/// TOML fields the operator added that are **not** modeled in the `Config`
/// struct are dropped on rewrite, and any human-authored comments are
/// erased. This is symmetric with the pre-existing `Config::save` path
/// and matches the v1 assumption that `config.toml` is machine-authored
/// (the only supported edits are through `aimx setup` / `aimx mailboxes
/// create|delete`). Adopting a preserving editor (e.g. `toml_edit`) is
/// tracked for v2; see the test `unknown_stanza_is_dropped_on_rewrite`
/// below for the contract check.
pub(crate) fn write_config_atomic(path: &Path, config: &Config) -> std::io::Result<()> {
    // Shared implementation lives in `config::write_atomic` so the
    // daemon path (this function) and the CLI fallback path
    // (`Config::save` — used by `mailbox create/delete` and
    // `hooks create/delete`) share one durability contract.
    crate::config::write_atomic(path, config)
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::config::{Config, MailboxConfig};
    use std::collections::HashMap;
    use tempfile::TempDir;

    fn base_config(data_dir: &Path) -> Config {
        let mut mailboxes = HashMap::new();
        mailboxes.insert(
            "catchall".to_string(),
            MailboxConfig {
                address: "*@example.com".to_string(),
                owner: "aimx-catchall".to_string(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        Config {
            domain: "example.com".to_string(),
            data_dir: data_dir.to_path_buf(),
            dkim_selector: "aimx".to_string(),
            trust: "none".to_string(),
            trusted_senders: vec![],
            mailboxes,
            verify_host: None,
            enable_ipv6: false,
            signature: None,
            upgrade: None,
        }
    }

    fn contexts(tmp: &TempDir) -> (StateContext, MailboxContext) {
        let config = base_config(tmp.path());
        let handle = ConfigHandle::new(config);
        let state_ctx = StateContext::new(tmp.path().to_path_buf(), handle.clone());
        let config_path = tmp.path().join("config.toml");
        // Seed the on-disk file so write-temp-then-rename has a real
        // target to replace.
        write_config_atomic(&config_path, &handle.load()).unwrap();
        let mb_ctx = MailboxContext::new(config_path, handle);
        (state_ctx, mb_ctx)
    }

    /// Install a test resolver that maps the literal username
    /// `"testowner"` (and `"root"`) to the current test process's
    /// uid/gid. Tests use `owner: Some("testowner".into())` so the
    /// handler's post-create chown is a no-op on the non-root CI runner
    /// (chown to your own uid/gid is a zero-effect syscall the kernel
    /// accepts for every user).
    fn install_tester_resolver() -> crate::user_resolver::test_resolver::ResolverOverride {
        fn fake(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "testowner" || name == "root" {
                let uid = unsafe { libc::geteuid() };
                let gid = unsafe { libc::getegid() };
                Some(crate::user_resolver::ResolvedUser {
                    name: name.to_string(),
                    uid,
                    gid,
                })
            } else {
                None
            }
        }
        crate::user_resolver::set_test_resolver(fake)
    }

    /// Install a test resolver that *additionally* recognizes the
    /// current Linux username (the one `peer_username(geteuid())`
    /// returns) and maps it to the current uid/gid. Sprint 1's
    /// non-root CREATE path synthesizes the owner string from
    /// `peer_username(caller.uid)`, so the chown step needs that
    /// real-user name to resolve via the `resolve_user` seam too.
    /// `"testowner"` and `"root"` continue to map for the existing
    /// root-caller tests.
    fn install_tester_resolver_with_current_user()
    -> crate::user_resolver::test_resolver::ResolverOverride {
        // Snapshot the current username so the resolver fn (which
        // can't capture environment) reads from a static lookup.
        // Statics in Rust can't run arbitrary code at init, so the
        // resolver fn does the lookup itself on each call — fine
        // because `peer_username` is just a `getpwuid` wrapper.
        fn fake(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            let uid = unsafe { libc::geteuid() };
            let gid = unsafe { libc::getegid() };
            // Resolve the current Linux user's actual name so this
            // resolver answers for whatever the daemon's
            // `peer_username` synthesized.
            // Per-call `lookup_username` (a thin `getpwuid` wrapper) is
            // intentional and cheap — do not memoize: tests rely on
            // each invocation re-reading passwd so resolver overrides
            // installed/torn down between calls stay isolated.
            let current_user = crate::uds_authz::lookup_username(uid);
            if name == "testowner" || name == "root" || current_user.as_deref() == Some(name) {
                Some(crate::user_resolver::ResolvedUser {
                    name: name.to_string(),
                    uid,
                    gid,
                })
            } else {
                None
            }
        }
        crate::user_resolver::set_test_resolver(fake)
    }

    /// Construct a "non-root" Caller for use in tests. The Caller's
    /// uid is the current process euid (so `peer_username` succeeds
    /// against the host's passwd file), but `is_root()` is gated on
    /// `uid == 0` — the test only runs on hosts where that is false.
    /// Returns `None` and lets the caller skip when the runner is
    /// uid 0 (root CI lane). Tests calling this **must** check the
    /// `Option` and short-circuit the test body when running as root.
    fn nonroot_caller_or_skip() -> Option<Caller> {
        let uid = unsafe { libc::geteuid() };
        if uid == 0 {
            return None;
        }
        let gid = unsafe { libc::getegid() };
        Some(Caller::new(uid, gid, None))
    }

    #[tokio::test]
    async fn create_mailbox_creates_dirs_writes_config_swaps_handle() {
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        // Dirs exist
        assert!(tmp.path().join("inbox").join("alice").is_dir());
        assert!(tmp.path().join("sent").join("alice").is_dir());

        // config.toml reloads the new mailbox
        let reloaded = Config::load_ignore_warnings(&mb_ctx.config_path).unwrap();
        assert!(reloaded.mailboxes.contains_key("alice"));
        assert_eq!(reloaded.mailboxes["alice"].address, "alice@example.com");

        // In-memory handle reflects the swap immediately.
        assert!(mb_ctx.config_handle.load().mailboxes.contains_key("alice"));
    }

    #[tokio::test]
    async fn create_duplicate_mailbox_returns_mailbox_error() {
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        // Seed a non-reserved mailbox first.
        let seed = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &seed, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let dup = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &dup, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Mailbox);
                assert!(reason.contains("already exists"), "{reason}");
            }
            other => panic!("expected Err(MAILBOX), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn create_rejects_reserved_name_catchall() {
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        for name in ["catchall", "aimx-catchall"] {
            let req = MailboxLifecycleRequest {
                name: name.into(),
                create: true,
                owner: Some("testowner".into()),
                force: false,
            };
            match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
                AckResponse::Err { code, reason } => {
                    assert_eq!(code, ErrCode::Validation, "name {name:?}");
                    assert!(
                        reason.contains("reserved"),
                        "name {name:?}: expected `reserved` in reason, got {reason}"
                    );
                }
                other => panic!("name {name:?}: expected Err(VALIDATION), got {other:?}"),
            }
        }
    }

    #[tokio::test]
    async fn create_rejects_reserved_name_before_authz() {
        // Defense-in-depth: reserved-name validation runs before the
        // root-only authz check, so the rejection text is the same
        // regardless of caller (root or non-root).
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let attacker = Caller::with_username(99999, 99999, "attacker");
        let req = MailboxLifecycleRequest {
            name: "catchall".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &attacker).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Validation);
                assert!(reason.contains("reserved"), "{reason}");
            }
            other => panic!("expected Err(VALIDATION), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn create_rejects_invalid_name_empty() {
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::Validation),
            other => panic!("expected Err(VALIDATION), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn create_rejects_invalid_name_path_traversal() {
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        for bad in ["../etc", "foo/bar", "foo\\bar", "a\0b"] {
            let req = MailboxLifecycleRequest {
                name: bad.into(),
                create: true,
                owner: Some("testowner".into()),
                force: false,
            };
            match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
                AckResponse::Err { code, .. } => {
                    assert_eq!(code, ErrCode::Validation, "name {bad:?}")
                }
                other => panic!("{bad:?}: expected Err(VALIDATION), got {other:?}"),
            }
        }
    }

    #[tokio::test]
    async fn delete_mailbox_removes_stanza_and_swaps_handle() {
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        // Pre-create
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        // Delete with empty dirs should succeed.
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: false,
            owner: None,
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        let reloaded = Config::load_ignore_warnings(&mb_ctx.config_path).unwrap();
        assert!(!reloaded.mailboxes.contains_key("alice"));
        assert!(!mb_ctx.config_handle.load().mailboxes.contains_key("alice"));
    }

    #[tokio::test]
    async fn delete_refuses_nonempty_mailbox() {
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        // Create
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        // Write a stray file into inbox
        let inbox = tmp.path().join("inbox").join("alice");
        std::fs::write(inbox.join("2025-06-01-001.md"), "content").unwrap();

        // Delete must be refused
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: false,
            owner: None,
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::NonEmpty);
                assert!(
                    reason.contains("alice") && reason.contains("1 files"),
                    "{reason}"
                );
            }
            other => panic!("expected Err(NONEMPTY), got {other:?}"),
        }

        // Stanza still present
        assert!(mb_ctx.config_handle.load().mailboxes.contains_key("alice"));
    }

    #[tokio::test]
    async fn delete_catchall_forbidden() {
        // The catchall name is now reserved at `validate_mailbox_name`,
        // so the rejection comes back as `Validation` ("reserved")
        // before ever reaching the catchall-specific delete guard. Both
        // the validation rejection and the deeper guard in
        // `handle_delete` agree the catchall must not be deleted via
        // UDS — the layer that fires first changed.
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "catchall".into(),
            create: false,
            owner: None,
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Validation);
                assert!(reason.contains("reserved"), "{reason}");
            }
            other => panic!("expected Err(VALIDATION), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn delete_nonexistent_returns_mailbox_error() {
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "ghost".into(),
            create: false,
            owner: None,
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Mailbox);
                assert!(reason.contains("does not exist"), "{reason}");
            }
            other => panic!("expected Err(MAILBOX), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn create_failure_at_disk_write_leaves_handle_and_disk_unchanged() {
        let _r = install_tester_resolver();
        // This test used to force failure by pointing config_path
        // at a non-existent parent directory, which tripped
        // `File::create` before the temp write even started, so the
        // rename-rollback branch was never exercised. The rewritten form
        // below makes the temp write succeed (parent is writable) and the
        // subsequent `rename(2)` fail because the target path is a
        // **non-empty directory**. `rename("foo.tmp", "config.toml/")`
        // with a non-empty directory as the target returns `EISDIR` /
        // `ENOTEMPTY` on Linux + macOS, which is what the real failure
        // mode looks like in production (e.g. a filesystem error mid-rename).
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        // `config.toml` as a non-empty *directory*: the temp write to
        // `.config.toml.tmp.<pid>` succeeds, then rename-over fails.
        let dir_path = tmp.path().join("configdir").join("config.toml");
        std::fs::create_dir_all(&dir_path).unwrap();
        // Put a file inside so the directory is non-empty (rename semantics
        // over an empty directory differ on some kernels).
        std::fs::write(dir_path.join("blocker"), "x").unwrap();
        let bad_ctx = MailboxContext::new(dir_path.clone(), mb_ctx.config_handle.clone());

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &bad_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => assert_eq!(code, ErrCode::Io),
            other => panic!("expected Err(IO), got {other:?}"),
        }

        // Config stays unchanged on both disk and memory.
        assert!(!bad_ctx.config_handle.load().mailboxes.contains_key("alice"));
        // Rollback cleaned up the freshly-created dirs so we don't leave
        // half a mailbox behind.
        assert!(!tmp.path().join("inbox").join("alice").exists());
        assert!(!tmp.path().join("sent").join("alice").exists());

        // No `.config.toml.tmp.<pid>` was left behind; the rollback path
        // cleans up the temp file after a rename failure.
        let parent = dir_path.parent().unwrap();
        let strays: Vec<_> = std::fs::read_dir(parent)
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.file_name().to_string_lossy().to_string())
            .filter(|n| n.contains(".tmp."))
            .collect();
        assert!(
            strays.is_empty(),
            "no stray temp file should remain after rename failure: {strays:?}"
        );
    }

    #[tokio::test]
    async fn unknown_stanza_is_dropped_on_rewrite() {
        // This pins the **documented** v1 behaviour of
        // `write_config_atomic`. The rewrite goes through
        // `toml::to_string_pretty(&Config)`, so any stanza the operator
        // added that isn't modeled in `Config` is not preserved. If this
        // test starts failing because the behaviour changed (e.g. we
        // adopted `toml_edit`), update the doc comment on
        // `write_config_atomic` and flip this assertion accordingly.
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("config.toml");

        let base = base_config(tmp.path());
        let serialized = toml::to_string_pretty(&base).unwrap();
        let with_unknown = format!(
            "# operator-added comment\n\
             {serialized}\n\
             [experimental.extra]\n\
             opaque_value = 42\n"
        );
        std::fs::write(&path, with_unknown).unwrap();

        // Rewrite through the canonical path.
        write_config_atomic(&path, &base).unwrap();

        let after = std::fs::read_to_string(&path).unwrap();
        assert!(
            !after.contains("operator-added comment"),
            "v1 documented behaviour: rewrite erases human comments. Got: {after:?}"
        );
        assert!(
            !after.contains("experimental"),
            "v1 documented behaviour: rewrite drops unknown stanzas. Got: {after:?}"
        );
    }

    #[tokio::test]
    async fn write_config_atomic_replaces_existing_file() {
        let tmp = TempDir::new().unwrap();
        let path = tmp.path().join("config.toml");
        let c1 = base_config(tmp.path());
        write_config_atomic(&path, &c1).unwrap();
        assert!(path.exists());

        // Overwrite with a different Config
        let mut c2 = c1.clone();
        c2.mailboxes.insert(
            "alice".to_string(),
            MailboxConfig {
                address: "alice@example.com".into(),
                owner: "ops".into(),
                hooks: vec![],
                trust: None,
                trusted_senders: None,
                allow_root_catchall: false,
            },
        );
        write_config_atomic(&path, &c2).unwrap();
        let reloaded = Config::load_ignore_warnings(&path).unwrap();
        assert!(reloaded.mailboxes.contains_key("alice"));

        // No `.tmp.*` file left behind
        let entries: Vec<_> = std::fs::read_dir(tmp.path())
            .unwrap()
            .filter_map(|e| e.ok())
            .map(|e| e.file_name().to_string_lossy().to_string())
            .collect();
        let has_tmp = entries.iter().any(|n| n.contains(".tmp."));
        assert!(
            !has_tmp,
            "no stray tmp file left after atomic rename: {entries:?}"
        );
    }

    #[tokio::test]
    async fn mailbox_crud_create_then_mark_works_on_same_mailbox() {
        let _r = install_tester_resolver();
        // Verify the `MAILBOX-CREATE` → `MARK-READ` chain works in-process
        // without a restart: the new mailbox is routable by MARK-*.
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(matches!(
            handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
            AckResponse::Ok
        ));

        // Drop a pre-existing email into the newly-created inbox and
        // mark it read via the same StateContext.
        use crate::frontmatter::InboundFrontmatter;
        let meta = InboundFrontmatter {
            id: "2025-06-01-001".into(),
            message_id: "<msg@example.com>".into(),
            thread_id: "0123456789abcdef".into(),
            from: "sender@example.com".into(),
            to: "alice@example.com".into(),
            cc: None,
            reply_to: None,
            delivered_to: "alice@example.com".into(),
            subject: "Hi".into(),
            date: "2025-06-01T12:00:00Z".into(),
            received_at: "2025-06-01T12:00:01Z".into(),
            received_from_ip: None,
            size_bytes: 10,
            in_reply_to: None,
            references: None,
            attachments: vec![],
            list_id: None,
            auto_submitted: None,
            dkim: "none".into(),
            spf: "none".into(),
            dmarc: "none".into(),
            trusted: "none".into(),
            mailbox: "alice".into(),
            read: false,
            read_at: None,
            labels: vec![],
        };
        let inbox = tmp.path().join("inbox").join("alice");
        let toml = toml::to_string(&meta).unwrap();
        let content = format!("+++\n{toml}+++\n\nbody\n");
        std::fs::write(inbox.join("2025-06-01-001.md"), content).unwrap();

        let mark_req = crate::send_protocol::MarkRequest {
            mailbox: "alice".into(),
            id: "2025-06-01-001".into(),
            read: true,
        };
        let resp = crate::state_handler::handle_mark(
            &state_ctx,
            &mark_req,
            &crate::uds_authz::Caller::internal_root(),
        )
        .await;
        assert!(matches!(resp, AckResponse::Ok), "{resp:?}");

        let reread = std::fs::read_to_string(inbox.join("2025-06-01-001.md")).unwrap();
        assert!(reread.contains("read = true"));
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn concurrent_create_different_names_both_stanzas_survive() {
        let _r = install_tester_resolver();
        // Regression test for the lost-update race on `config.toml`:
        // two concurrent `MAILBOX-CREATE` calls on *different* names
        // held disjoint per-mailbox locks and could interleave their
        // load-modify-write sequences, clobbering one stanza. The
        // process-wide `CONFIG_WRITE_LOCK` closes the race. Every
        // stanza must survive on disk and in the live handle.
        //
        // We fan out to many concurrent names to give the scheduler a
        // realistic chance of interleaving without the lock; a
        // single-pair test is too easy for tokio's multi-thread runtime
        // to serialize accidentally.
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        let state_ctx = std::sync::Arc::new(state_ctx);
        let mb_ctx = std::sync::Arc::new(mb_ctx);

        let names: Vec<String> = (0..16).map(|i| format!("mbox{i:02}")).collect();
        let mut handles = Vec::with_capacity(names.len());
        for name in &names {
            let s = state_ctx.clone();
            let m = mb_ctx.clone();
            let n = name.clone();
            handles.push(tokio::spawn(async move {
                let req = MailboxLifecycleRequest {
                    name: n,
                    create: true,
                    owner: Some("testowner".into()),
                    force: false,
                };
                handle_mailbox_crud(&s, &m, &req, &Caller::internal_root()).await
            }));
        }
        for h in handles {
            assert!(matches!(h.await.unwrap(), AckResponse::Ok));
        }

        // Every stanza survives on disk.
        let reloaded = Config::load_ignore_warnings(&mb_ctx.config_path).unwrap();
        for name in &names {
            assert!(
                reloaded.mailboxes.contains_key(name),
                "{name} missing on disk: {:?}",
                reloaded.mailboxes.keys().collect::<Vec<_>>()
            );
        }

        // Every stanza survives in the live handle.
        let in_mem = mb_ctx.config_handle.load();
        for name in &names {
            assert!(
                in_mem.mailboxes.contains_key(name),
                "{name} missing in handle: {:?}",
                in_mem.mailboxes.keys().collect::<Vec<_>>()
            );
        }

        // Every directory pair exists.
        for name in &names {
            assert!(tmp.path().join("inbox").join(name).is_dir());
            assert!(tmp.path().join("sent").join(name).is_dir());
        }
    }

    // ----- Owner:-header behaviour --------------------------------------

    #[tokio::test]
    async fn create_missing_owner_header_is_validation_error() {
        // The parser requires Owner: on CREATE, but the dispatcher also
        // defends in depth: if somehow a `None` reaches the handler, the
        // response must be EINVAL-shaped (ErrCode::Validation).
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: None,
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Validation);
                assert!(reason.contains("Owner:"), "{reason}");
            }
            other => panic!("expected Err(VALIDATION), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn create_unknown_owner_returns_validation_and_leaves_disk_clean() {
        // The tester resolver only resolves `testowner`; any other name
        // returns None. The handler must return EINVAL and NOT touch
        // disk so a bad owner doesn't leave orphan directories behind.
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("ghost".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Validation);
                assert!(reason.contains("ghost"), "{reason}");
            }
            other => panic!("expected Err(VALIDATION), got {other:?}"),
        }
        assert!(!tmp.path().join("inbox").join("alice").exists());
        assert!(!tmp.path().join("sent").join("alice").exists());
    }

    #[tokio::test]
    async fn create_existing_dirs_with_wrong_owner_return_conflict() {
        // Pre-existing dirs owned by a uid/gid other than the requested
        // owner land on ECONFLICT. We simulate "wrong owner" by using
        // the default-create uid (current user) and then pointing the
        // resolver at a different uid (test-only: uid 65534 / gid
        // 65534, the traditional `nobody` uids).
        let _r_init = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        std::fs::create_dir_all(tmp.path().join("inbox").join("alice")).unwrap();
        std::fs::create_dir_all(tmp.path().join("sent").join("alice")).unwrap();

        // Swap to a resolver that returns an improbable uid (65534) so
        // the pre-existing dir's "owned by current user" state definitely
        // doesn't match the requested owner.
        drop(_r_init);
        fn nobody(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "nobody" {
                Some(crate::user_resolver::ResolvedUser {
                    name: "nobody".into(),
                    uid: 65534,
                    gid: 65534,
                })
            } else {
                None
            }
        }
        let _r = crate::user_resolver::set_test_resolver(nobody);

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("nobody".into()),
            force: false,
        };
        // Note: the test harness may actually be running as uid 65534 in
        // some exotic CI, in which case the conflict check flips to
        // "already correct" and the create succeeds. Accept either
        // Ok or Conflict so the test is portable.
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Conflict, "{reason}");
                assert!(
                    reason.contains("chown"),
                    "fix hint must be present: {reason}"
                );
            }
            AckResponse::Ok => {
                let current_uid = unsafe { libc::geteuid() };
                assert_eq!(
                    current_uid, 65534,
                    "Ok response only acceptable when the test runner IS uid 65534"
                );
            }
        }
    }

    #[tokio::test]
    async fn create_conflict_on_preexisting_dir_with_wrong_owner_preserves_operator_data() {
        // This test pins the `Conflict` short-circuit
        // in `check_preexisting_dirs`, NOT the chown-failure rollback
        // path. By pre-creating `inbox/alice` owned by the current uid
        // while asking the handler to chown for `nobody` (uid 65534),
        // the preexistence check refuses with `Conflict` BEFORE any
        // chown is attempted. The rollback invariant we pin here is:
        // operator-created artifacts survive, no config stanza is
        // written, and the handler-created sibling dir does not linger.
        //
        // The chown-failure rollback branch is exercised separately by
        // `create_rollback_on_chown_failure_removes_only_self_created_dirs`.
        let is_root = unsafe { libc::geteuid() == 0 };
        if is_root {
            return;
        }
        fn nobody(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "nobody" {
                Some(crate::user_resolver::ResolvedUser {
                    name: "nobody".into(),
                    uid: 65534,
                    gid: 65534,
                })
            } else {
                None
            }
        }
        let _r = crate::user_resolver::set_test_resolver(nobody);

        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let inbox = tmp.path().join("inbox").join("alice");
        let sent = tmp.path().join("sent").join("alice");
        std::fs::create_dir_all(&inbox).unwrap();
        // Seed an operator-authored file inside the pre-existing inbox
        // dir so `remove_dir` (which only removes empty dirs) cannot
        // touch it even if rollback *did* try to remove it.
        std::fs::write(inbox.join("do-not-delete"), b"operator data").unwrap();

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("nobody".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, .. } => {
                assert_eq!(
                    code,
                    ErrCode::Conflict,
                    "pre-existing dir with wrong owner must short-circuit with Conflict, \
                     got {code:?}"
                );
            }
            other => panic!("expected Err, got {other:?}"),
        }

        // Operator's pre-existing inbox dir and sentinel file survive.
        assert!(
            inbox.join("do-not-delete").exists(),
            "operator's data must survive rollback"
        );
        assert!(
            inbox.is_dir(),
            "pre-existing inbox dir must survive rollback"
        );
        // Sent dir was never created (early Conflict return from the
        // preexistence check on inbox).
        assert!(
            !sent.exists(),
            "handler must not create the sent dir when inbox pre-exists with wrong owner"
        );
        // Config stanza did not get written.
        assert!(!mb_ctx.config_handle.load().mailboxes.contains_key("alice"));
    }

    #[tokio::test]
    async fn create_rollback_on_chown_failure_removes_only_self_created_dirs() {
        // When the post-create chown
        // fails the handler must roll back the directories IT created
        // in this call. To deterministically hit the chown-failure
        // branch (NOT the preexistence Conflict short-circuit), we
        // leave both `inbox/alice` and `sent/alice` absent so the
        // handler creates them fresh, then resolve the owner to uid
        // 65534 (nobody) so the subsequent `libc::chown` syscall fails
        // with EPERM on an unprivileged CI runner.
        //
        // Root runners skip the test since the chown syscall succeeds
        // and we can't provoke the failure without elevated privilege
        // tricks we don't want in the test suite.
        let is_root = unsafe { libc::geteuid() == 0 };
        if is_root {
            return;
        }
        fn nobody(name: &str) -> Option<crate::user_resolver::ResolvedUser> {
            if name == "nobody" {
                Some(crate::user_resolver::ResolvedUser {
                    name: "nobody".into(),
                    uid: 65534,
                    gid: 65534,
                })
            } else {
                None
            }
        }
        let _r = crate::user_resolver::set_test_resolver(nobody);

        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let inbox = tmp.path().join("inbox").join("alice");
        let sent = tmp.path().join("sent").join("alice");
        // Neither dir pre-exists. The handler creates both fresh, then
        // the chown for `nobody` fails (EPERM under unprivileged euid),
        // and rollback removes both self-created dirs.
        assert!(!inbox.exists());
        assert!(!sent.exists());

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("nobody".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(
                    code,
                    ErrCode::Io,
                    "chown failure must surface as Io, got {code:?}: {reason}"
                );
                assert!(
                    reason.contains("failed to chown"),
                    "reason must name the chown op: {reason}"
                );
            }
            other => panic!("expected Err, got {other:?}"),
        }

        // Both handler-created dirs were rolled back.
        assert!(
            !inbox.exists(),
            "handler-created inbox dir must not survive chown failure"
        );
        assert!(
            !sent.exists(),
            "handler-created sent dir must not survive chown failure"
        );
        // Config stanza did not get written.
        assert!(!mb_ctx.config_handle.load().mailboxes.contains_key("alice"));
    }

    #[tokio::test]
    async fn create_idempotent_when_dirs_already_owned_correctly() {
        // Pre-create dirs owned by the current (tester) uid/gid, then
        // submit a CREATE with owner=testowner (mapped to the same
        // uid/gid). The handler should succeed without touching the
        // existing directories' ownership.
        let _r = install_tester_resolver();
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        let inbox = tmp.path().join("inbox").join("alice");
        let sent = tmp.path().join("sent").join("alice");
        std::fs::create_dir_all(&inbox).unwrap();
        std::fs::create_dir_all(&sent).unwrap();

        let req = MailboxLifecycleRequest {
            name: "alice".into(),
            create: true,
            owner: Some("testowner".into()),
            force: false,
        };
        assert!(
            matches!(
                handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &Caller::internal_root()).await,
                AckResponse::Ok
            ),
            "idempotent re-create should succeed"
        );
        // Config stanza was still written.
        assert!(
            mb_ctx.config_handle.load().mailboxes.contains_key("alice"),
            "config must register the mailbox on idempotent re-create"
        );
    }

    // ----- Sprint 1 (S1-4) privilege-escalation negative tests --------
    //
    // These tests pin the structural defense that the daemon never
    // trusts the wire `Owner:` header from a non-root caller and never
    // leaks information about mailboxes a non-root caller does not own.
    // A failure here probably means a future refactor reopened the
    // exact privilege-escalation hole this sprint closed.

    #[tokio::test]
    async fn nonroot_create_ignores_wire_owner_and_uses_so_peercred() {
        // The headline NFR1 test: a non-root caller submits
        // `Owner: root` over the wire; the daemon must drop the wire
        // value on the floor and synthesize the owner from the
        // `SO_PEERCRED` uid via `peer_username`.
        let _r = install_tester_resolver_with_current_user();
        let Some(caller) = nonroot_caller_or_skip() else {
            return;
        };
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "x".into(),
            create: true,
            // The attacker tries to claim a root-owned mailbox.
            owner: Some("root".into()),
            force: false,
        };
        let resp = handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &caller).await;
        assert!(matches!(resp, AckResponse::Ok), "{resp:?}");

        let cfg = mb_ctx.config_handle.load();
        let stanza = cfg
            .mailboxes
            .get("x")
            .expect("mailbox stanza must be present after Ok");

        let expected_owner =
            crate::uds_authz::peer_username(caller.uid).expect("current uid must resolve");
        assert_eq!(
            stanza.owner, expected_owner,
            "Owner: header was honored — privilege-escalation hole reopened. \
             Expected owner '{expected_owner}' (from SO_PEERCRED), got '{}'.",
            stanza.owner,
        );
        assert_ne!(
            stanza.owner, "root",
            "non-root caller created a root-owned mailbox via Owner: header — \
             this must never happen"
        );
    }

    #[tokio::test]
    async fn nonroot_delete_unowned_returns_no_such_mailbox_string_match() {
        // NFR2: the wire response for "mailbox exists but you don't
        // own it" must be byte-identical to "no such mailbox" so the
        // daemon doesn't help an attacker enumerate which mailboxes
        // exist.
        let _r = install_tester_resolver();
        let Some(nonroot) = nonroot_caller_or_skip() else {
            return;
        };
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        // Inject a mailbox owned by root directly into the in-memory
        // config snapshot. We bypass `MAILBOX-CREATE` because the
        // create path's chown to uid 0 would fail for a non-root test
        // runner; only the *delete* path's auth check is what this
        // test exercises.
        {
            let current = mb_ctx.config_handle.load();
            let mut new_config: Config = (*current).clone();
            new_config.mailboxes.insert(
                "opsbox".into(),
                MailboxConfig {
                    address: "opsbox@example.com".into(),
                    owner: "root".into(),
                    hooks: vec![],
                    trust: None,
                    trusted_senders: None,
                    allow_root_catchall: false,
                },
            );
            mb_ctx.config_handle.store(new_config);
        }

        // The "missing" reason a non-root delete of a never-existed
        // mailbox produces. Compute it from the same helper to make
        // the string-match contract obvious.
        let canonical_missing = no_such_mailbox_reason("nope");

        // Non-root tries to delete the root-owned mailbox: the wire
        // response must equal what they'd get for a non-existent name.
        let unowned_delete = MailboxLifecycleRequest {
            name: "opsbox".into(),
            create: false,
            owner: None,
            force: false,
        };
        let unowned_resp =
            handle_mailbox_crud(&state_ctx, &mb_ctx, &unowned_delete, &nonroot).await;

        // Same caller, mailbox that never existed: this is the
        // baseline "no such mailbox" string the unowned response must
        // match (modulo the mailbox name).
        let missing_delete = MailboxLifecycleRequest {
            name: "nope".into(),
            create: false,
            owner: None,
            force: false,
        };
        let missing_resp =
            handle_mailbox_crud(&state_ctx, &mb_ctx, &missing_delete, &nonroot).await;

        match (unowned_resp, missing_resp) {
            (
                AckResponse::Err {
                    code: c1,
                    reason: r1,
                },
                AckResponse::Err {
                    code: c2,
                    reason: r2,
                },
            ) => {
                assert_eq!(c1, c2, "ErrCode must match: unowned={c1:?} missing={c2:?}");
                assert_eq!(c1, ErrCode::Mailbox);
                // The two reasons differ only in the mailbox name
                // suffix; both match the no-such-mailbox template.
                assert_eq!(r1, no_such_mailbox_reason("opsbox"));
                assert_eq!(r2, canonical_missing);
                // Final guarantee: the unowned reason must NOT mention
                // ownership, owners, the actual owner uid, the caller,
                // or anything else that would let an attacker tell the
                // two cases apart by inspection.
                for leak in &[
                    "owner",
                    "Owner",
                    "permission",
                    "denied",
                    "authorized",
                    "root",
                ] {
                    assert!(
                        !r1.contains(leak),
                        "unowned-delete reason leaked '{leak}': {r1:?}"
                    );
                }
            }
            (a, b) => panic!("expected two Err responses, got {a:?} and {b:?}"),
        }

        // The mailbox must still exist on disk — the unowned-delete
        // attempt must not delete or modify state it didn't authorize.
        assert!(mb_ctx.config_handle.load().mailboxes.contains_key("opsbox"));
    }

    #[tokio::test]
    async fn nonroot_create_idempotent_for_owned_mailbox() {
        // Re-creating a mailbox the non-root caller already owns is a
        // no-op success, mirroring the existing root-side idempotent
        // path.
        let _r = install_tester_resolver_with_current_user();
        let Some(caller) = nonroot_caller_or_skip() else {
            return;
        };
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let req = MailboxLifecycleRequest {
            name: "agent".into(),
            create: true,
            // Owner is ignored for non-root; included to prove that.
            owner: Some("ignored-by-daemon".into()),
            force: false,
        };
        let r1 = handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &caller).await;
        assert!(matches!(r1, AckResponse::Ok), "first create: {r1:?}");

        // Second create with the exact same name + caller: idempotent.
        // The handler returns ERR(Mailbox, "already exists") — the
        // pre-Sprint-1 behavior on a duplicate. This pin keeps the
        // idempotent-on-disk semantics ("dirs already owned correctly
        // bypass the chown") working for non-root callers too.
        let r2 = handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &caller).await;
        match r2 {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Mailbox);
                assert!(
                    reason.contains("already exists"),
                    "second create reason: {reason}"
                );
            }
            other => panic!("expected Err(MAILBOX, already exists), got {other:?}"),
        }
    }

    #[tokio::test]
    async fn nonroot_create_delete_create_cycle_succeeds() {
        // The lifecycle invariant: a non-root caller can create, then
        // delete, then re-create the same mailbox name, all three
        // succeeding. This catches any state that leaks across the
        // delete (config stanza, lock map, in-memory handle) that
        // would block the second create.
        let _r = install_tester_resolver_with_current_user();
        let Some(caller) = nonroot_caller_or_skip() else {
            return;
        };
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);

        let create_req = MailboxLifecycleRequest {
            name: "task42".into(),
            create: true,
            owner: None,
            force: false,
        };
        let delete_req = MailboxLifecycleRequest {
            name: "task42".into(),
            create: false,
            owner: None,
            force: false,
        };

        let r1 = handle_mailbox_crud(&state_ctx, &mb_ctx, &create_req, &caller).await;
        assert!(matches!(r1, AckResponse::Ok), "create #1: {r1:?}");
        assert!(mb_ctx.config_handle.load().mailboxes.contains_key("task42"));

        let r2 = handle_mailbox_crud(&state_ctx, &mb_ctx, &delete_req, &caller).await;
        assert!(matches!(r2, AckResponse::Ok), "delete: {r2:?}");
        assert!(!mb_ctx.config_handle.load().mailboxes.contains_key("task42"));

        let r3 = handle_mailbox_crud(&state_ctx, &mb_ctx, &create_req, &caller).await;
        assert!(matches!(r3, AckResponse::Ok), "create #2: {r3:?}");
        assert!(mb_ctx.config_handle.load().mailboxes.contains_key("task42"));
    }

    #[tokio::test]
    async fn nonroot_create_orphan_uid_returns_validation_with_exact_message() {
        // A `SO_PEERCRED` uid that has no `/etc/passwd` entry → the
        // daemon must surface `ErrCode::Validation` with the exact
        // message `"uid <N> has no passwd entry"`. This is the only
        // wire-text dependency this sprint introduces; pinning it
        // keeps the behavior visible to operators reading the
        // structured log.
        //
        // Discover an unused high uid at runtime rather than hardcoding
        // one — `0xFFFFFFFE` maps to `nobody` on some images (notably
        // Alpine). We probe downward until `getpwuid` returns `None`,
        // then use that uid; if the entire upper range is mapped on
        // this host (vanishingly unlikely), the test is skipped with
        // a diagnostic.
        let Some(orphan_uid) = find_orphan_uid() else {
            eprintln!(
                "skipping nonroot_create_orphan_uid_returns_validation_with_exact_message: \
                 no unmapped high uid available on this host"
            );
            return;
        };
        let tmp = TempDir::new().unwrap();
        let (state_ctx, mb_ctx) = contexts(&tmp);
        let orphan = Caller::new(orphan_uid, orphan_uid, None);

        let req = MailboxLifecycleRequest {
            name: "orph".into(),
            create: true,
            owner: Some("anything".into()),
            force: false,
        };
        match handle_mailbox_crud(&state_ctx, &mb_ctx, &req, &orphan).await {
            AckResponse::Err { code, reason } => {
                assert_eq!(code, ErrCode::Validation);
                assert_eq!(reason, format!("uid {orphan_uid} has no passwd entry"));
            }
            other => panic!("expected Err(VALIDATION, orphan-uid), got {other:?}"),
        }

        // Defense in depth: nothing got written.
        assert!(!tmp.path().join("inbox").join("orph").exists());
        assert!(!tmp.path().join("sent").join("orph").exists());
        assert!(!mb_ctx.config_handle.load().mailboxes.contains_key("orph"));
    }

    /// Walk down from `0xFFFFFFFE` until we find a uid `lookup_username`
    /// reports as unmapped. The bounded probe depth keeps the test
    /// fast even on hosts with many high-uid passwd entries.
    fn find_orphan_uid() -> Option<u32> {
        const PROBE_DEPTH: u32 = 64;
        let start: u32 = 0xFFFF_FFFE;
        for offset in 0..PROBE_DEPTH {
            let candidate = start - offset;
            if crate::uds_authz::lookup_username(candidate).is_none() {
                return Some(candidate);
            }
        }
        None
    }
}
