Dockerfile

@@ -36,6 +36,25 @@ ENV INBUXA_REVISION=${INBUXA_REVISION}
 RUN RUSTFLAGS="$(cat /flags.txt)" cargo build --target "$(cat /target.txt)" --release -p inbuxa --no-default-features --features "sqlite postgres mysql rocks s3 redis azure nats"
 RUN mv "/build/target/$(cat /target.txt)/release" "/output"
 
+# inbuxa: migrations spec, MG-4: the inbuxa-migrate the server runs, a
+# release pinned by its checksums
+FROM --platform=$BUILDPLATFORM mirror.gcr.io/library/debian:trixie-slim AS migrate
+ARG TARGETPLATFORM
+ARG MIGRATE_VERSION=27.1-rc.8
+ARG MIGRATE_SHA256_AMD64=a8a880111c583fd190f70852ed64fc2794cbc085ef3623f90193e168d7cbc75d
+ARG MIGRATE_SHA256_ARM64=05b7402a61e0203b3bb4a5e967cf3da4d1ace64c8fafe78eaa64157d2abea2c8
+RUN export DEBIAN_FRONTEND=noninteractive && \
+    apt-get update && \
+    apt-get install -yq --no-install-recommends ca-certificates curl && \
+    case "${TARGETPLATFORM}" in \
+    "linux/arm64") arch=arm64 sum="${MIGRATE_SHA256_ARM64}" ;; \
+    "linux/amd64") arch=amd64 sum="${MIGRATE_SHA256_AMD64}" ;; \
+    *) exit 1 ;; \
+    esac && \
+    curl -fsSL -o /migrate.tar.gz "https://github.com/inbuxa/inbuxa-migrate/releases/download/v${MIGRATE_VERSION}/inbuxa-migrate-linux-${arch}.tar.gz" && \
+    echo "${sum}  /migrate.tar.gz" | sha256sum -c - && \
+    mkdir /migrate && tar -xzf /migrate.tar.gz -C /migrate
+
 FROM mirror.gcr.io/library/debian:trixie-slim
 RUN export DEBIAN_FRONTEND=noninteractive && \
     apt-get update && \
@@ -46,6 +65,9 @@ RUN export DEBIAN_FRONTEND=noninteractive && \
     mkdir -p /etc/inbuxa /var/lib/inbuxa && \
     chown inbuxa:inbuxa /etc/inbuxa /var/lib/inbuxa
 COPY --from=builder --chmod=0755 /output/inbuxa /usr/local/bin/inbuxa
+COPY --from=migrate --chmod=0755 /migrate/inbuxa-migrate /usr/local/bin/inbuxa-migrate
+COPY --from=migrate /migrate/LICENSES /usr/share/doc/inbuxa-migrate/LICENSES
+COPY --from=migrate /migrate/README.md /usr/share/doc/inbuxa-migrate/README.md
 RUN setcap 'cap_net_bind_service=+ep' /usr/local/bin/inbuxa
 USER inbuxa
 WORKDIR /var/lib/inbuxa

crates/common/src/audit.rs

@@ -180,7 +180,9 @@ impl Server {
     pub async fn audit_sign_in(&self, req: &AuthRequest, token: &AccessToken) {
         let via = token.origin();
         let (actor, target) = match via {
-            None | Some(Via::OAuth { .. }) | Some(Via::Directory) => return,
+            None | Some(Via::OAuth { .. }) | Some(Via::Directory) | Some(Via::Migration { .. }) => {
+                return;
+            }
             Some(Via::Master { account_id, name }) => {
                 let target_id = token.account_id();
                 (

crates/common/src/auth/authentication.rs

@@ -351,6 +351,25 @@ impl Server {
                 }
             }
             Credentials::Bearer { username, token } => {
+                // inbuxa: migrations spec, MG-10: a migration's run token,
+                // from this node's loopback only
+                if token.starts_with(crate::migration_token::PREFIX) {
+                    let Some(holder) = crate::migration_token::holder(token, req.remote_ip) else {
+                        return Err(trc::AuthEvent::Failed
+                            .into_err()
+                            .reason("Not a live migration run token, or not from loopback"));
+                    };
+                    return self
+                        .access_token(holder.account_id)
+                        .await
+                        .and_then(|token| AccessToken::new(token, req.remote_ip))
+                        .map(|token| {
+                            token.with_origin(Via::Migration {
+                                id: holder.migration_id,
+                            })
+                        });
+                }
+
                 // Handle API key authentication
                 if let Some(key) = ApiKey::parse(token) {
                     return self

crates/common/src/auth/permissions.rs

@@ -374,6 +374,11 @@ impl Default for DefaultPermissions {
                 | Permission::SysRelayQueueUpdate => {
                     default.superuser.push(permission);
                 }
+                // inbuxa: migrations spec, MG-10: a migration writes into
+                // the whole server, so only its administrators run one
+                Permission::SysMigrationRun => {
+                    default.superuser.push(permission);
+                }
                 // inbuxa: AL-12: tenant administrators lock and delegate
                 // within their tenant
                 Permission::SysAccountLockGet

crates/common/src/lib.rs

@@ -74,6 +74,7 @@ pub mod hold; // inbuxa: legal holds (audit-hold-lock spec, LH)
 pub mod i18n;
 pub mod ipc;
 pub mod manager;
+pub mod migration_token; // inbuxa: migrations' run tokens (migrations spec, MG-10)
 pub mod network;
 pub mod privacy; // inbuxa: the personal-data catalog, evaluated
 pub mod reachability; // inbuxa: whether the outside world reaches each node's ports

crates/common/src/manager/granted_permissions.rs

@@ -33,8 +33,9 @@ use types::id::Id;
 /// locks and legal holds (audit-hold-lock spec, AU-9, AL-12, LH-13), and
 /// the data inventory (personal-data catalog spec), accepting security
 /// to-do items (security to-do list spec), the deliverability check
-/// (deliverability spec), scheduled reports (scheduled-reports spec), and
-/// inbuxa Relays (relay spec).
+/// (deliverability spec), scheduled reports (scheduled-reports spec),
+/// inbuxa Relays (relay spec), and running migrations (migrations spec,
+/// MG-10).
 const ADMIN_GRANTS: &[Permission] = &[
     Permission::SysAiExplain,
     Permission::SysAuditGet,
@@ -67,6 +68,7 @@ const ADMIN_GRANTS: &[Permission] = &[
     Permission::SysRelayUpdate,
     Permission::SysRelayQueueGet,
     Permission::SysRelayQueueUpdate,
+    Permission::SysMigrationRun,
 ];
 
 /// Granted to the server-level Compliance Officer role once it exists:

crates/common/src/migration_token.rs

@@ -0,0 +1,94 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! inbuxa: the run tokens a migration's inbuxa-migrate signs in with
+//! (migrations spec, MG-10). A run gets one when it starts and loses it when
+//! it ends. It acts as the administrator who started the migration, from
+//! this node's loopback only.
+//!
+//! Tokens are held in this node's memory and nowhere else, not even hashed
+//! in the store: the child runs here, so no other node needs them, and a
+//! restart ends them along with the runs they were for.
+
+use base64::{Engine, engine::general_purpose::URL_SAFE_NO_PAD};
+use parking_lot::Mutex;
+use sha2::{Digest, Sha256};
+use std::{collections::HashMap, net::IpAddr, sync::LazyLock};
+
+/// What a run token starts with, so authentication knows it at a glance.
+pub const PREFIX: &str = "MIG_";
+
+/// Whom a run token acts as, and for which migration.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct Holder {
+    pub account_id: u32,
+    pub migration_id: u32,
+}
+
+/// By the token's hash: a lookup compares digests, never the token.
+static TOKENS: LazyLock<Mutex<HashMap<[u8; 32], Holder>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+fn digest(token: &str) -> [u8; 32] {
+    Sha256::digest(token.as_bytes()).into()
+}
+
+/// A new token for a run of `migration_id`, acting as `account_id`.
+pub fn issue(account_id: u32, migration_id: u32) -> String {
+    let token = format!(
+        "{PREFIX}{}",
+        URL_SAFE_NO_PAD.encode(store::rand::random::<[u8; 32]>())
+    );
+    TOKENS.lock().insert(
+        digest(&token),
+        Holder {
+            account_id,
+            migration_id,
+        },
+    );
+    token
+}
+
+/// Ends a token; it signs nothing in after this.
+pub fn revoke(token: &str) {
+    TOKENS.lock().remove(&digest(token));
+}
+
+/// Whom `token` acts as, if it's a live run token and the request came
+/// from this node's loopback.
+pub fn holder(token: &str, remote_ip: IpAddr) -> Option<Holder> {
+    if !remote_ip.to_canonical().is_loopback() {
+        return None;
+    }
+    TOKENS.lock().get(&digest(token)).copied()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::net::{Ipv4Addr, Ipv6Addr};
+
+    #[test]
+    fn a_token_works_from_loopback_until_revoked() {
+        let token = issue(7, 3);
+        assert!(token.starts_with(PREFIX));
+        let local = IpAddr::V4(Ipv4Addr::LOCALHOST);
+        let expected = Some(Holder {
+            account_id: 7,
+            migration_id: 3,
+        });
+        assert_eq!(holder(&token, local), expected);
+        assert_eq!(holder(&token, IpAddr::V6(Ipv6Addr::LOCALHOST)), expected);
+        assert_eq!(
+            holder(&token, "::ffff:127.0.0.1".parse().unwrap()),
+            expected
+        );
+        assert_eq!(holder(&token, "192.0.2.1".parse().unwrap()), None);
+        assert_eq!(holder(&format!("{token}x"), local), None);
+        revoke(&token);
+        assert_eq!(holder(&token, local), None);
+    }
+}

crates/features/src/audit/record.rs

@@ -99,6 +99,11 @@ pub enum Via {
     },
     /// The recovery administrator from the server's own configuration.
     Recovery,
+    /// A migration's run, acting for the administrator who started it
+    /// (migrations spec, MG-10).
+    Migration {
+        id: u32,
+    },
 }
 
 /// What kind of thing happened.

crates/features/src/lib.rs

@@ -27,6 +27,7 @@ pub mod journal;
 pub mod lock;
 pub mod mailflow;
 pub mod masked_email;
+pub mod migration; // inbuxa: migrations into this server (not a rebuild)
 pub mod privacy;
 pub mod relay; // inbuxa: the downstream side of inbuxa Relay (not a rebuild)
 pub mod scheduled_reports; // inbuxa: scheduled reports and the weekly digest (not a rebuild)

crates/features/src/migration/mod.rs

@@ -0,0 +1,610 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Migrations into this server (docs/spec/features/migrations.md): the
+//! `inbuxa:Migration` objects. The server runs the bundled inbuxa-migrate
+//! on a migration's answers; this module keeps what a migration is and how
+//! far it got, never a secret.
+//!
+//! Kept in the fork's subspace (`store::SUBSPACE_INBUXA`), never in the
+//! registry. Every key starts with `I` (import), then one byte for the kind:
+//!
+//! - `m` + migration id (u32): a migration ([`Migration`]), as JSON.
+//!
+//! Numbers are big-endian. There are few migrations, so they're read whole.
+
+use serde::{Deserialize as SerdeDeserialize, Serialize as SerdeSerialize, de::DeserializeOwned};
+use serde_json::{Map, Value};
+use store::{
+    Deserialize, IterateParams, SUBSPACE_INBUXA, Serialize, Store, ValueKey,
+    write::{AnyClass, BatchBuilder, ValueClass, assert::AssertValue},
+};
+use trc::AddContext;
+
+const FEATURE: u8 = b'I';
+const KIND_MIGRATION: u8 = b'm';
+const CREATE_ATTEMPTS: usize = 5;
+
+/// MG-3: at most this many accounts in one migration.
+pub const MAX_ACCOUNTS: usize = 10_000;
+/// MG-9: the failures a migration keeps, newest last.
+pub const MAX_FAILURES: usize = 200;
+const MAX_TEXT: usize = 2_000;
+
+/// MG-2: where the mail is now. Maildir is left out: its folders would have
+/// to be on this node.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "lowercase")]
+pub enum SourceKind {
+    Imap,
+    Jmap,
+    Microsoft,
+    Google,
+}
+
+impl SourceKind {
+    pub fn parse(value: &str) -> Option<Self> {
+        Some(match value {
+            "imap" => SourceKind::Imap,
+            "jmap" => SourceKind::Jmap,
+            "microsoft" => SourceKind::Microsoft,
+            "google" => SourceKind::Google,
+            _ => return None,
+        })
+    }
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            SourceKind::Imap => "imap",
+            SourceKind::Jmap => "jmap",
+            SourceKind::Microsoft => "microsoft",
+            SourceKind::Google => "google",
+        }
+    }
+}
+
+/// MG-2: the source, as the guide asks for it, without its secret.
+#[derive(Debug, Clone, Default, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase", default)]
+pub struct Source {
+    pub url: String,
+    /// IMAP: the login for each account, such as `{email}*master`. JMAP:
+    /// an administrator, or empty for an API key.
+    pub login: String,
+    /// IMAP: credentials on `imap://` without STARTTLS.
+    pub cleartext: bool,
+    /// Microsoft 365: the Entra tenant and the app's client id.
+    pub tenant: String,
+    pub client_id: String,
+}
+
+/// Which run a migration is doing, or did last (MG-5).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum RunKind {
+    /// A dry run: signs in for each account and counts what would move.
+    Preview,
+    /// The move. Run again, it resumes.
+    Move,
+    /// The move again with `--again`, at cutover.
+    FinalPass,
+}
+
+impl RunKind {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            RunKind::Preview => "preview",
+            RunKind::Move => "move",
+            RunKind::FinalPass => "finalPass",
+        }
+    }
+}
+
+/// How the last run ended (MG-6).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum RunResult {
+    /// Every account finished.
+    Done,
+    /// Some account didn't finish; the report says which and why.
+    Unfinished,
+    /// The run couldn't start, or the engine stopped with an error.
+    Failed,
+    /// An administrator stopped it.
+    Stopped,
+    /// The server stopped while it ran. Resume carries on; the secret has
+    /// to be given again (MG-12).
+    Interrupted,
+}
+
+impl RunResult {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            RunResult::Done => "done",
+            RunResult::Unfinished => "unfinished",
+            RunResult::Failed => "failed",
+            RunResult::Stopped => "stopped",
+            RunResult::Interrupted => "interrupted",
+        }
+    }
+}
+
+/// Where a migration is in its life (MG-5, MG-14).
+#[derive(Debug, Clone, Copy, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub enum Status {
+    /// Nothing is running.
+    Idle,
+    Running,
+    /// Closed: its work directory, archives included, is gone. It stays
+    /// listed.
+    Closed,
+}
+
+impl Status {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Status::Idle => "idle",
+            Status::Running => "running",
+            Status::Closed => "closed",
+        }
+    }
+}
+
+/// What a run has done so far, from the engine's progress (MG-8).
+#[derive(Debug, Clone, Default, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase", default)]
+pub struct Progress {
+    /// The accounts this run has.
+    pub total: u64,
+    /// The accounts it finished with, however they ended.
+    pub finished: u64,
+    /// Of those, the ones that didn't finish.
+    pub unfinished: u64,
+    /// Items read from the source, summed over the accounts so far.
+    pub read: u64,
+    /// Items written into this server: created and updated.
+    pub written: u64,
+    /// Items that failed, on either side.
+    pub failed: u64,
+}
+
+/// An account that didn't finish, and why (MG-9).
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Failure {
+    pub email: String,
+    pub step: String,
+    pub detail: String,
+}
+
+/// A migration into this server (MG-1).
+#[derive(Debug, Clone, PartialEq, Eq, SerdeSerialize, SerdeDeserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct Migration {
+    pub id: u32,
+    #[serde(default)]
+    pub description: String,
+    pub source_kind: SourceKind,
+    #[serde(default)]
+    pub source: Source,
+    pub accounts: Vec<String>,
+    /// Create on this server the accounts it doesn't have yet.
+    #[serde(default)]
+    pub create_accounts: bool,
+    pub parallel: u32,
+    pub status: Status,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub run: Option<RunKind>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub result: Option<RunResult>,
+    /// Why a run failed, in the engine's words.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub error: Option<String>,
+    #[serde(default)]
+    pub progress: Progress,
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub failures: Vec<Failure>,
+    /// The node the migration's work directory is on (MG-11). Set by the
+    /// first run.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub node_id: Option<u64>,
+    /// When the running node last said it was still at it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub heartbeat: Option<u64>,
+    /// Whether a move has finished at least once, so a final pass makes
+    /// sense.
+    #[serde(default)]
+    pub moved: bool,
+    /// The work directory's size, archives included, after the last run.
+    #[serde(default)]
+    pub disk_used: u64,
+    /// Whether new accounts' passwords are waiting to be downloaded once
+    /// (MG-13).
+    #[serde(default)]
+    pub passwords_waiting: bool,
+    /// The last run's report, as a blob of whoever started the migration.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub report_blob_id: Option<String>,
+    pub created_at: u64,
+    pub created_by: String,
+    /// Whom the runs act as (MG-10).
+    pub created_by_id: u32,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub started_at: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub finished_at: Option<u64>,
+}
+
+impl Migration {
+    /// The answers `inbuxa-migrate plan` reads, for a target this server
+    /// gives it (MG-4). The secret isn't among them.
+    pub fn answers(&self, target_url: &str, allow_invalid_certs: bool) -> Value {
+        let mut source = Map::new();
+        source.insert("kind".into(), self.source_kind.as_str().into());
+        source.insert("url".into(), self.source.url.clone().into());
+        source.insert("login".into(), self.source.login.clone().into());
+        source.insert("cleartext".into(), self.source.cleartext.into());
+        source.insert("tenant".into(), self.source.tenant.clone().into());
+        source.insert("clientId".into(), self.source.client_id.clone().into());
+        serde_json::json!({
+            "source": source,
+            "accounts": self.accounts,
+            "create": self.create_accounts,
+            "target": {
+                "url": target_url,
+                "stayOnUrl": true,
+                "allowInvalidCerts": allow_invalid_certs,
+            },
+            "parallel": self.parallel,
+        })
+    }
+
+    /// Adds a failure, keeping the newest [`MAX_FAILURES`].
+    pub fn add_failure(&mut self, failure: Failure) {
+        self.failures.retain(|f| f.email != failure.email);
+        self.failures.push(failure);
+        let over = self.failures.len().saturating_sub(MAX_FAILURES);
+        self.failures.drain(..over);
+    }
+}
+
+/// Why a migration's answers were refused, for the console.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct Invalid {
+    pub property: &'static str,
+    pub reason: String,
+}
+
+fn invalid(property: &'static str, reason: impl Into<String>) -> Result<(), Invalid> {
+    Err(Invalid {
+        property,
+        reason: reason.into(),
+    })
+}
+
+/// An address, as the engine takes it: lowercase, one `@`, no spaces.
+pub fn account_email(value: &str) -> Option<String> {
+    let email = value.trim().to_ascii_lowercase();
+    let (local, domain) = email.split_once('@')?;
+    (!local.is_empty()
+        && domain.contains('.')
+        && !domain.contains('@')
+        && email.len() <= 320
+        && !email.chars().any(|c| c.is_whitespace() || c.is_control()))
+    .then_some(email)
+}
+
+fn names_account(template: &str) -> bool {
+    template.contains("{email}") || template.contains("{local}")
+}
+
+/// MG-2, MG-3: what a migration needs before anything runs. The engine
+/// checks the rest when it writes the plans.
+pub fn validate(migration: &Migration) -> Result<(), Invalid> {
+    let s = &migration.source;
+    for (property, text) in [
+        ("description", &migration.description),
+        ("source", &s.url),
+        ("source", &s.login),
+        ("source", &s.tenant),
+        ("source", &s.client_id),
+    ] {
+        if text.len() > MAX_TEXT || text.chars().any(char::is_control) {
+            return invalid(property, "Too long, or holds a control character.");
+        }
+    }
+    match migration.source_kind {
+        SourceKind::Imap => {
+            if !(s.url.starts_with("imaps://") || s.url.starts_with("imap://")) {
+                return invalid(
+                    "source",
+                    "Give the IMAP server as imaps://host or imap://host.",
+                );
+            }
+            if !names_account(&s.login) {
+                return invalid(
+                    "source",
+                    "The login for each account has to name it: {email} or {local}.",
+                );
+            }
+        }
+        SourceKind::Jmap => {
+            if !(s.url.starts_with("https://") || s.url.starts_with("http://")) {
+                return invalid("source", "Give the JMAP server's address, from https://.");
+            }
+        }
+        SourceKind::Microsoft => {
+            if s.tenant.trim().is_empty() || s.client_id.trim().is_empty() {
+                return invalid(
+                    "source",
+                    "Give the Microsoft 365 tenant and the app's client id.",
+                );
+            }
+        }
+        SourceKind::Google => {}
+    }
+    if migration.accounts.is_empty() {
+        return invalid("accounts", "List the accounts to move.");
+    }
+    if migration.accounts.len() > MAX_ACCOUNTS {
+        return invalid(
+            "accounts",
+            format!("At most {MAX_ACCOUNTS} accounts in one migration; split it."),
+        );
+    }
+    for email in &migration.accounts {
+        if account_email(email).as_deref() != Some(email.as_str()) {
+            return invalid("accounts", format!("{email} isn't an address."));
+        }
+    }
+    if !(1..=16).contains(&migration.parallel) {
+        return invalid("parallel", "Between 1 and 16 accounts at once.");
+    }
+    Ok(())
+}
+
+struct Json<T>(T);
+
+impl<T: SerdeSerialize> Serialize for Json<&T> {
+    fn serialize(&self) -> trc::Result<Vec<u8>> {
+        serde_json::to_vec(self.0).map_err(|err| {
+            trc::StoreEvent::UnexpectedError
+                .into_err()
+                .details("Could not serialize a migration")
+                .reason(err)
+        })
+    }
+}
+
+impl<T: DeserializeOwned + Sync + Send> Deserialize for Json<T> {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        serde_json::from_slice(bytes).map(Json).map_err(|err| {
+            trc::StoreEvent::DataCorruption
+                .into_err()
+                .details("Invalid migration")
+                .reason(err)
+        })
+    }
+}
+
+fn class(id: u32) -> ValueClass {
+    let mut key = Vec::with_capacity(6);
+    key.push(FEATURE);
+    key.push(KIND_MIGRATION);
+    key.extend_from_slice(&id.to_be_bytes());
+    ValueClass::Any(AnyClass {
+        subspace: SUBSPACE_INBUXA,
+        key,
+    })
+}
+
+pub async fn get(data: &Store, id: u32) -> trc::Result<Option<Migration>> {
+    Ok(data
+        .get_value::<Json<Migration>>(ValueKey::from(class(id)))
+        .await
+        .caused_by(trc::location!())?
+        .map(|Json(migration)| migration))
+}
+
+/// Every migration, oldest first.
+pub async fn all(data: &Store) -> trc::Result<Vec<Migration>> {
+    let mut migrations = Vec::new();
+    data.iterate(
+        IterateParams::new(ValueKey::from(class(0)), ValueKey::from(class(u32::MAX))),
+        |_, value| {
+            if let Ok(Json(migration)) = Json::<Migration>::deserialize(value) {
+                migrations.push(migration);
+            }
+            Ok(true)
+        },
+    )
+    .await
+    .caused_by(trc::location!())?;
+    migrations.sort_by_key(|m| m.id);
+    Ok(migrations)
+}
+
+/// Writes a new migration under the next free id, which it returns. Two
+/// nodes creating migrations at once can't take the same id.
+pub async fn create(data: &Store, migration: &Migration) -> trc::Result<u32> {
+    let mut attempt = 0;
+    loop {
+        attempt += 1;
+        let id = all(data).await?.iter().map(|m| m.id).max().unwrap_or(0) + 1;
+        let stored = Migration {
+            id,
+            ..migration.clone()
+        };
+        let mut batch = BatchBuilder::new();
+        batch.assert_value(class(id), AssertValue::None);
+        batch.set(class(id), Json(&stored).serialize()?);
+        match data.write(batch.build_all()).await {
+            Ok(_) => return Ok(id),
+            Err(err)
+                if attempt < CREATE_ATTEMPTS
+                    && matches!(
+                        err.as_ref(),
+                        trc::EventType::Store(trc::StoreEvent::AssertValueFailed)
+                    ) => {}
+            Err(err) => return Err(err.caused_by(trc::location!())),
+        }
+    }
+}
+
+/// A migration as stored, with the version [`replace`] checks: a hash of
+/// the stored bytes.
+struct Versioned(Migration, u64);
+
+impl Deserialize for Versioned {
+    fn deserialize(bytes: &[u8]) -> trc::Result<Self> {
+        let Json(migration) = Json::<Migration>::deserialize(bytes)?;
+        Ok(Versioned(migration, xxhash_rust::xxh3::xxh3_64(bytes)))
+    }
+}
+
+pub async fn get_versioned(data: &Store, id: u32) -> trc::Result<Option<(Migration, u64)>> {
+    Ok(data
+        .get_value::<Versioned>(ValueKey::from(class(id)))
+        .await
+        .caused_by(trc::location!())?
+        .map(|Versioned(migration, version)| (migration, version)))
+}
+
+/// Replaces a stored migration if it is still at `version`. False if
+/// something else changed it first, such as the running node saving
+/// progress while an administrator stops it: read it again and retry.
+pub async fn replace(data: &Store, migration: &Migration, version: u64) -> trc::Result<bool> {
+    let mut batch = BatchBuilder::new();
+    batch.assert_value(class(migration.id), AssertValue::Hash(version));
+    batch.set(class(migration.id), Json(migration).serialize()?);
+    match data.write(batch.build_all()).await {
+        Ok(_) => Ok(true),
+        Err(err)
+            if matches!(
+                err.as_ref(),
+                trc::EventType::Store(trc::StoreEvent::AssertValueFailed)
+            ) =>
+        {
+            Ok(false)
+        }
+        Err(err) => Err(err.caused_by(trc::location!())),
+    }
+}
+
+/// Reads a migration, changes it with `change`, and writes it back, again
+/// if something else wrote it in between. `change` returns false to leave
+/// it as it is. None if there's no such migration.
+pub async fn update<F>(data: &Store, id: u32, mut change: F) -> trc::Result<Option<Migration>>
+where
+    F: FnMut(&mut Migration) -> bool,
+{
+    loop {
+        let Some((mut migration, version)) = get_versioned(data, id).await? else {
+            return Ok(None);
+        };
+        if !change(&mut migration) {
+            return Ok(Some(migration));
+        }
+        if replace(data, &migration, version).await? {
+            return Ok(Some(migration));
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn migration(kind: SourceKind) -> Migration {
+        Migration {
+            id: 1,
+            description: String::new(),
+            source_kind: kind,
+            source: Source {
+                url: "imaps://old.example.org".into(),
+                login: "{email}*master".into(),
+                ..Source::default()
+            },
+            accounts: vec!["[email protected]".into()],
+            create_accounts: true,
+            parallel: 4,
+            status: Status::Idle,
+            run: None,
+            result: None,
+            error: None,
+            progress: Progress::default(),
+            failures: Vec::new(),
+            node_id: None,
+            heartbeat: None,
+            moved: false,
+            disk_used: 0,
+            passwords_waiting: false,
+            report_blob_id: None,
+            created_at: 0,
+            created_by: "admin".into(),
+            created_by_id: 1,
+            started_at: None,
+            finished_at: None,
+        }
+    }
+
+    #[test]
+    fn answers_name_this_server_and_stay_on_its_address() {
+        let answers = migration(SourceKind::Imap).answers("http://127.0.0.1:8080", false);
+        assert_eq!(answers["source"]["kind"], "imap");
+        assert_eq!(answers["source"]["login"], "{email}*master");
+        assert_eq!(answers["target"]["url"], "http://127.0.0.1:8080");
+        assert_eq!(answers["target"]["stayOnUrl"], true);
+        assert_eq!(answers["create"], true);
+        assert!(answers["target"].get("token").is_none());
+        assert!(answers["source"].get("secret").is_none());
+    }
+
+    #[test]
+    fn what_a_migration_needs() {
+        assert_eq!(validate(&migration(SourceKind::Imap)), Ok(()));
+
+        let mut m = migration(SourceKind::Imap);
+        m.source.login = "master".into();
+        assert_eq!(validate(&m).unwrap_err().property, "source");
+
+        let mut m = migration(SourceKind::Microsoft);
+        assert_eq!(validate(&m).unwrap_err().property, "source");
+        m.source.tenant = "contoso.onmicrosoft.com".into();
+        m.source.client_id = "1234".into();
+        assert_eq!(validate(&m), Ok(()));
+
+        let mut m = migration(SourceKind::Google);
+        m.accounts.clear();
+        assert_eq!(validate(&m).unwrap_err().property, "accounts");
+        m.accounts = vec!["[email protected]".into()];
+        assert_eq!(validate(&m).unwrap_err().property, "accounts");
+        m.accounts = vec!["[email protected]".into()];
+        m.parallel = 0;
+        assert_eq!(validate(&m).unwrap_err().property, "parallel");
+    }
+
+    #[test]
+    fn failures_keep_one_per_account_and_the_newest() {
+        let mut m = migration(SourceKind::Imap);
+        for n in 0..(MAX_FAILURES + 5) {
+            m.add_failure(Failure {
+                email: format!("u{n}@example.org"),
+                step: "import".into(),
+                detail: "no".into(),
+            });
+        }
+        m.add_failure(Failure {
+            email: format!("u{}@example.org", MAX_FAILURES + 4),
+            step: "export".into(),
+            detail: "again".into(),
+        });
+        assert_eq!(m.failures.len(), MAX_FAILURES);
+        assert_eq!(m.failures.last().unwrap().step, "export");
+        assert_eq!(m.failures[0].email, "[email protected]");
+    }
+}

crates/http/src/auth/authenticate.rs

@@ -106,7 +106,10 @@ impl Authenticator for Server {
                     .min(lifetime),
                 None => lifetime,
             };
-            if !lifetime.is_zero() {
+            // inbuxa: migrations spec, MG-10: a run token is checked on every
+            // request, so it ends with its run and stays loopback-only
+            let is_run_token = token.starts_with(common::migration_token::PREFIX);
+            if !lifetime.is_zero() && !is_run_token {
                 self.inner.cache.http_auth.insert(
                     token.into(),
                     HttpAuthCache {

crates/jmap-proto/src/object/inbuxa_migration.rs

@@ -0,0 +1,268 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:Migration/get` and `/set` under `urn:inbuxa:jmap`: moving
+//! accounts into this server (docs/spec/features/migrations.md). A create
+//! records what to move; an update's `action` runs it. The set call's
+//! `reason` argument, if given, goes into the audit log with the change.
+
+use crate::{
+    object::{AnyId, JmapObject, JmapObjectId},
+    request::deserialize::DeserializeArguments,
+};
+use jmap_tools::{Element, Key, Property};
+use std::{borrow::Cow, str::FromStr};
+use types::id::Id;
+
+#[derive(Debug, Clone, Default)]
+pub struct Migration;
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum MigrationProperty {
+    Id,
+    Description,
+    /// `imap`, `jmap`, `microsoft` or `google` (MG-2).
+    SourceKind,
+    /// The source's address and login, without its secret.
+    Source,
+    Accounts,
+    /// Create the accounts this server doesn't have yet.
+    CreateAccounts,
+    /// Accounts moved at once.
+    Parallel,
+    /// `idle`, `running` or `closed`.
+    Status,
+    /// `preview`, `move` or `finalPass`: the run going on, or the last one.
+    Run,
+    /// How the last run ended (MG-6).
+    Result,
+    Error,
+    Progress,
+    /// The accounts that didn't finish, and why (MG-9).
+    Failures,
+    /// The node its work directory is on (MG-11).
+    NodeId,
+    Heartbeat,
+    /// Whether a move has finished once, so a final pass makes sense.
+    Moved,
+    DiskUsed,
+    /// Whether new accounts' passwords wait to be downloaded, once (MG-13).
+    PasswordsWaiting,
+    /// The last run's report, as CSV, for whoever started the migration.
+    ReportBlobId,
+    /// Update only: `preview`, `move`, `finalPass`, `stop` or `passwords` (MG-5).
+    Action,
+    /// Update only: the source's password, API key, client secret or key, kept in memory (MG-12).
+    Secret,
+    /// In a `passwords` update's response only.
+    PasswordsBlobId,
+    CreatedAt,
+    CreatedBy,
+    StartedAt,
+    FinishedAt,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
+pub enum MigrationValue {
+    Id(Id),
+}
+
+impl Property for MigrationProperty {
+    fn try_parse(parent: Option<&Key<'_, Self>>, value: &str) -> Option<Self> {
+        // Keys inside source and progress stay plain keys
+        match parent {
+            None => MigrationProperty::parse(value),
+            Some(_) => None,
+        }
+    }
+
+    fn to_cow(&self) -> Cow<'static, str> {
+        match self {
+            MigrationProperty::Id => "id",
+            MigrationProperty::Description => "description",
+            MigrationProperty::SourceKind => "sourceKind",
+            MigrationProperty::Source => "source",
+            MigrationProperty::Accounts => "accounts",
+            MigrationProperty::CreateAccounts => "createAccounts",
+            MigrationProperty::Parallel => "parallel",
+            MigrationProperty::Status => "status",
+            MigrationProperty::Run => "run",
+            MigrationProperty::Result => "result",
+            MigrationProperty::Error => "error",
+            MigrationProperty::Progress => "progress",
+            MigrationProperty::Failures => "failures",
+            MigrationProperty::NodeId => "nodeId",
+            MigrationProperty::Heartbeat => "heartbeat",
+            MigrationProperty::Moved => "moved",
+            MigrationProperty::DiskUsed => "diskUsed",
+            MigrationProperty::PasswordsWaiting => "passwordsWaiting",
+            MigrationProperty::ReportBlobId => "reportBlobId",
+            MigrationProperty::Action => "action",
+            MigrationProperty::Secret => "secret",
+            MigrationProperty::PasswordsBlobId => "passwordsBlobId",
+            MigrationProperty::CreatedAt => "createdAt",
+            MigrationProperty::CreatedBy => "createdBy",
+            MigrationProperty::StartedAt => "startedAt",
+            MigrationProperty::FinishedAt => "finishedAt",
+        }
+        .into()
+    }
+}
+
+impl MigrationProperty {
+    fn parse(value: &str) -> Option<Self> {
+        hashify::tiny_map!(value.as_bytes(),
+            b"id" => MigrationProperty::Id,
+            b"description" => MigrationProperty::Description,
+            b"sourceKind" => MigrationProperty::SourceKind,
+            b"source" => MigrationProperty::Source,
+            b"accounts" => MigrationProperty::Accounts,
+            b"createAccounts" => MigrationProperty::CreateAccounts,
+            b"parallel" => MigrationProperty::Parallel,
+            b"status" => MigrationProperty::Status,
+            b"run" => MigrationProperty::Run,
+            b"result" => MigrationProperty::Result,
+            b"error" => MigrationProperty::Error,
+            b"progress" => MigrationProperty::Progress,
+            b"failures" => MigrationProperty::Failures,
+            b"nodeId" => MigrationProperty::NodeId,
+            b"heartbeat" => MigrationProperty::Heartbeat,
+            b"moved" => MigrationProperty::Moved,
+            b"diskUsed" => MigrationProperty::DiskUsed,
+            b"passwordsWaiting" => MigrationProperty::PasswordsWaiting,
+            b"reportBlobId" => MigrationProperty::ReportBlobId,
+            b"action" => MigrationProperty::Action,
+            b"secret" => MigrationProperty::Secret,
+            b"passwordsBlobId" => MigrationProperty::PasswordsBlobId,
+            b"createdAt" => MigrationProperty::CreatedAt,
+            b"createdBy" => MigrationProperty::CreatedBy,
+            b"startedAt" => MigrationProperty::StartedAt,
+            b"finishedAt" => MigrationProperty::FinishedAt,
+        )
+    }
+}
+
+impl FromStr for MigrationProperty {
+    type Err = ();
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        MigrationProperty::parse(s).ok_or(())
+    }
+}
+
+impl Element for MigrationValue {
+    type Property = MigrationProperty;
+
+    fn try_parse<P>(key: &Key<'_, Self::Property>, value: &str) -> Option<Self> {
+        match key {
+            Key::Property(MigrationProperty::Id) => {
+                Id::from_str(value).ok().map(MigrationValue::Id)
+            }
+            _ => None,
+        }
+    }
+
+    fn to_cow(&self) -> Cow<'static, str> {
+        match self {
+            MigrationValue::Id(id) => id.to_string().into(),
+        }
+    }
+}
+
+/// The set call's own argument: why, for the audit log.
+#[derive(Debug, Clone, Default)]
+pub struct MigrationSetArguments {
+    pub reason: Option<String>,
+}
+
+impl<'de> DeserializeArguments<'de> for MigrationSetArguments {
+    fn deserialize_argument<A>(&mut self, key: &str, map: &mut A) -> Result<(), A::Error>
+    where
+        A: serde::de::MapAccess<'de>,
+    {
+        if key == "reason" {
+            self.reason = map.next_value()?;
+        } else {
+            let _ = map.next_value::<serde::de::IgnoredAny>()?;
+        }
+        Ok(())
+    }
+}
+
+impl JmapObject for Migration {
+    type Property = MigrationProperty;
+
+    type Element = MigrationValue;
+
+    type Id = Id;
+
+    type Filter = ();
+
+    type Comparator = ();
+
+    type GetArguments = ();
+
+    type SetArguments<'de> = MigrationSetArguments;
+
+    type QueryArguments = ();
+
+    type CopyArguments = ();
+
+    type ParseArguments = ();
+
+    const ID_PROPERTY: Self::Property = MigrationProperty::Id;
+}
+
+impl From<Id> for MigrationValue {
+    fn from(id: Id) -> Self {
+        MigrationValue::Id(id)
+    }
+}
+
+impl JmapObjectId for MigrationValue {
+    fn as_id(&self) -> Option<Id> {
+        match self {
+            MigrationValue::Id(id) => Some(*id),
+        }
+    }
+
+    fn as_any_id(&self) -> Option<AnyId> {
+        match self {
+            MigrationValue::Id(id) => Some(AnyId::Id(*id)),
+        }
+    }
+
+    fn as_id_ref(&self) -> Option<&str> {
+        None
+    }
+
+    fn try_set_id(&mut self, new_id: AnyId) -> bool {
+        if let AnyId::Id(id) = new_id {
+            *self = MigrationValue::Id(id);
+            true
+        } else {
+            false
+        }
+    }
+}
+
+impl JmapObjectId for MigrationProperty {
+    fn as_id(&self) -> Option<Id> {
+        None
+    }
+
+    fn as_any_id(&self) -> Option<AnyId> {
+        None
+    }
+
+    fn as_id_ref(&self) -> Option<&str> {
+        None
+    }
+
+    fn try_set_id(&mut self, _: AnyId) -> bool {
+        false
+    }
+}

crates/jmap-proto/src/object/mod.rs

@@ -41,6 +41,7 @@ pub mod inbuxa_journal_entry; // inbuxa: journaling, search and export
 pub mod inbuxa_legal_hold; // inbuxa: legal hold
 pub mod inbuxa_log_settings; // inbuxa: personal-data catalog, D1
 pub mod inbuxa_mail_rule; // inbuxa: DLP and mail flow rules
+pub mod inbuxa_migration; // inbuxa: migrations into this server
 pub mod inbuxa_protocol_policy; // inbuxa: legacy protocols off
 pub mod inbuxa_relay; // inbuxa: inbuxa Relay
 pub mod inbuxa_relay_message; // inbuxa: a relay's queues

crates/jmap-proto/src/references/eval.rs

@@ -88,6 +88,9 @@ impl Response<'_> {
                         GetResponseMethod::MailRule(response) => {
                             response.eval_jptr(path, &mut results)
                         }
+                        GetResponseMethod::Migration(response) => {
+                            response.eval_jptr(path, &mut results)
+                        }
                         GetResponseMethod::Relay(response) => {
                             response.eval_jptr(path, &mut results)
                         }

crates/jmap-proto/src/references/resolve.rs

@@ -55,6 +55,7 @@ impl Response<'_> {
                 GetRequestMethod::AccountLock(request) => request.resolve_references(self)?,
                 GetRequestMethod::LegalHold(request) => request.resolve_references(self)?,
                 GetRequestMethod::MailRule(request) => request.resolve_references(self)?,
+                GetRequestMethod::Migration(request) => request.resolve_references(self)?,
                 GetRequestMethod::Relay(request) => request.resolve_references(self)?,
                 GetRequestMethod::RelayMessage(request) => request.resolve_references(self)?,
                 GetRequestMethod::SecurityAcceptance(request) => {
@@ -151,6 +152,9 @@ impl Response<'_> {
                 SetRequestMethod::MailRule(request) => {
                     request.resolve_references(self, 1, false)?
                 }
+                SetRequestMethod::Migration(request) => {
+                    request.resolve_references(self, 1, false)?
+                }
                 SetRequestMethod::Relay(request) => request.resolve_references(self, 1, false)?,
                 SetRequestMethod::RelayMessage(request) => {
                     request.resolve_references(self, 1, false)?

crates/jmap-proto/src/request/method.rs

@@ -68,6 +68,8 @@ pub enum MethodObject {
     ProtocolPolicy,
     // inbuxa: DLP and mail flow rules
     MailRule,
+    // inbuxa: migrations into this server
+    Migration,
     Relay,
     RelayMessage,
     // inbuxa: accepted security to-do items
@@ -126,6 +128,7 @@ impl MethodObject {
             | MethodObject::LegalHold
             | MethodObject::HoldExport
             | MethodObject::MailRule
+            | MethodObject::Migration
             | MethodObject::Relay
             | MethodObject::RelayMessage
             | MethodObject::SecurityAcceptance
@@ -340,6 +343,8 @@ impl MethodName {
             (MethodFunction::Set, MethodObject::LegalHold) => "inbuxa:LegalHold/set",
             (MethodFunction::Get, MethodObject::MailRule) => "inbuxa:MailRule/get",
             (MethodFunction::Set, MethodObject::MailRule) => "inbuxa:MailRule/set",
+            (MethodFunction::Get, MethodObject::Migration) => "inbuxa:Migration/get",
+            (MethodFunction::Set, MethodObject::Migration) => "inbuxa:Migration/set",
             (MethodFunction::Get, MethodObject::Relay) => "inbuxa:Relay/get",
             (MethodFunction::Set, MethodObject::Relay) => "inbuxa:Relay/set",
             (MethodFunction::Get, MethodObject::RelayMessage) => "inbuxa:RelayMessage/get",
@@ -543,6 +548,8 @@ impl MethodName {
             "inbuxa:LegalHold/set" => (MethodObject::LegalHold, MethodFunction::Set),
             "inbuxa:MailRule/get" => (MethodObject::MailRule, MethodFunction::Get),
             "inbuxa:MailRule/set" => (MethodObject::MailRule, MethodFunction::Set),
+            "inbuxa:Migration/get" => (MethodObject::Migration, MethodFunction::Get),
+            "inbuxa:Migration/set" => (MethodObject::Migration, MethodFunction::Set),
             "inbuxa:Relay/get" => (MethodObject::Relay, MethodFunction::Get),
             "inbuxa:Relay/set" => (MethodObject::Relay, MethodFunction::Set),
             "inbuxa:RelayMessage/get" => (MethodObject::RelayMessage, MethodFunction::Get),
@@ -644,6 +651,7 @@ impl Display for MethodObject {
             MethodObject::AccountLock => "inbuxa:AccountLock",
             MethodObject::LegalHold => "inbuxa:LegalHold",
             MethodObject::MailRule => "inbuxa:MailRule",
+            MethodObject::Migration => "inbuxa:Migration",
             MethodObject::Relay => "inbuxa:Relay",
             MethodObject::RelayMessage => "inbuxa:RelayMessage",
             MethodObject::SecurityAcceptance => "inbuxa:SecurityAcceptance",

crates/jmap-proto/src/request/mod.rs

@@ -127,6 +127,7 @@ pub enum GetRequestMethod {
     AccountLock(Box<GetRequest<crate::object::inbuxa_account_lock::AccountLock>>),
     LegalHold(Box<GetRequest<crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<GetRequest<crate::object::inbuxa_mail_rule::MailRule>>),
+    Migration(Box<GetRequest<crate::object::inbuxa_migration::Migration>>),
     Relay(Box<GetRequest<crate::object::inbuxa_relay::Relay>>),
     RelayMessage(Box<GetRequest<crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(
@@ -185,6 +186,7 @@ pub enum SetRequestMethod<'x> {
     AccountLock(Box<SetRequest<'x, crate::object::inbuxa_account_lock::AccountLock>>),
     LegalHold(Box<SetRequest<'x, crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<SetRequest<'x, crate::object::inbuxa_mail_rule::MailRule>>),
+    Migration(Box<SetRequest<'x, crate::object::inbuxa_migration::Migration>>),
     Relay(Box<SetRequest<'x, crate::object::inbuxa_relay::Relay>>),
     RelayMessage(Box<SetRequest<'x, crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(

crates/jmap-proto/src/request/parser.rs

@@ -689,6 +689,21 @@ impl<'de> Visitor<'de> for CallVisitor {
                     return Err(de::Error::invalid_length(1, &self));
                 }
             },
+            // inbuxa: migrations into this server
+            (MethodFunction::Get, MethodObject::Migration) => match seq.next_element() {
+                Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::Migration(value)),
+                Err(err) => RequestMethod::invalid(err),
+                Ok(None) => {
+                    return Err(de::Error::invalid_length(1, &self));
+                }
+            },
+            (MethodFunction::Set, MethodObject::Migration) => match seq.next_element() {
+                Ok(Some(value)) => RequestMethod::Set(SetRequestMethod::Migration(value)),
+                Err(err) => RequestMethod::invalid(err),
+                Ok(None) => {
+                    return Err(de::Error::invalid_length(1, &self));
+                }
+            },
             // inbuxa: inbuxa Relay
             (MethodFunction::Get, MethodObject::Relay) => match seq.next_element() {
                 Ok(Some(value)) => RequestMethod::Get(GetRequestMethod::Relay(value)),

crates/jmap-proto/src/response/mod.rs

@@ -112,6 +112,7 @@ pub enum GetResponseMethod {
     AccountLock(GetResponse<crate::object::inbuxa_account_lock::AccountLock>),
     LegalHold(GetResponse<crate::object::inbuxa_legal_hold::LegalHold>),
     MailRule(GetResponse<crate::object::inbuxa_mail_rule::MailRule>),
+    Migration(GetResponse<crate::object::inbuxa_migration::Migration>),
     Relay(GetResponse<crate::object::inbuxa_relay::Relay>),
     RelayMessage(GetResponse<crate::object::inbuxa_relay_message::RelayMessage>),
     SecurityAcceptance(GetResponse<crate::object::inbuxa_security_acceptance::SecurityAcceptance>),
@@ -168,6 +169,7 @@ pub enum SetResponseMethod {
     AccountLock(Box<SetResponse<crate::object::inbuxa_account_lock::AccountLock>>),
     LegalHold(Box<SetResponse<crate::object::inbuxa_legal_hold::LegalHold>>),
     MailRule(Box<SetResponse<crate::object::inbuxa_mail_rule::MailRule>>),
+    Migration(Box<SetResponse<crate::object::inbuxa_migration::Migration>>),
     Relay(Box<SetResponse<crate::object::inbuxa_relay::Relay>>),
     RelayMessage(Box<SetResponse<crate::object::inbuxa_relay_message::RelayMessage>>),
     SecurityAcceptance(
@@ -1037,6 +1039,19 @@ impl<'x> From<SetResponse<crate::object::inbuxa_mail_rule::MailRule>> for Respon
     }
 }
 
+// inbuxa: migrations into this server
+impl<'x> From<GetResponse<crate::object::inbuxa_migration::Migration>> for ResponseMethod<'x> {
+    fn from(value: GetResponse<crate::object::inbuxa_migration::Migration>) -> Self {
+        ResponseMethod::Get(GetResponseMethod::Migration(value))
+    }
+}
+
+impl<'x> From<SetResponse<crate::object::inbuxa_migration::Migration>> for ResponseMethod<'x> {
+    fn from(value: SetResponse<crate::object::inbuxa_migration::Migration>) -> Self {
+        ResponseMethod::Set(SetResponseMethod::Migration(Box::new(value)))
+    }
+}
+
 impl<'x> From<GetResponse<crate::object::inbuxa_relay::Relay>> for ResponseMethod<'x> {
     fn from(value: GetResponse<crate::object::inbuxa_relay::Relay>) -> Self {
         ResponseMethod::Get(GetResponseMethod::Relay(value))

crates/jmap/src/api/auth.rs

@@ -116,6 +116,8 @@ impl JmapAuthorization for AccessToken {
                         Permission::SysDlpPolicyGet
                     }
                 }
+                // inbuxa: migrations into this server (MG-10)
+                GetRequestMethod::Migration(_) => Permission::SysMigrationRun,
                 // inbuxa: inbuxa Relay
                 GetRequestMethod::Relay(_) => Permission::SysRelayGet,
                 GetRequestMethod::RelayMessage(_) => Permission::SysRelayQueueGet,
@@ -313,6 +315,14 @@ impl JmapAuthorization for AccessToken {
                                 .details("You are not authorized to change mail rules"))
                         }
                     }
+                    // inbuxa: migrations into this server (MG-10)
+                    SetRequestMethod::Migration(s) => validate_set(
+                        s,
+                        self,
+                        Permission::SysMigrationRun,
+                        Permission::SysMigrationRun,
+                        Permission::SysMigrationRun,
+                    ),
                     // inbuxa: inbuxa Relay; a new join token is an update
                     SetRequestMethod::Relay(s) => validate_set(
                         s,
@@ -585,6 +595,7 @@ impl JmapAuthorization for AccessToken {
                 | MethodObject::LegalHold
                 | MethodObject::HoldExport
                 | MethodObject::MailRule
+                | MethodObject::Migration
                 | MethodObject::Relay
                 | MethodObject::RelayMessage
                 | MethodObject::SecurityAcceptance

crates/jmap/src/api/request.rs

@@ -290,6 +290,9 @@ impl RequestHandler for Server {
                                     SetResponseMethod::MailRule(set_response) => {
                                         set_response.update_created_ids(&mut response);
                                     }
+                                    SetResponseMethod::Migration(set_response) => {
+                                        set_response.update_created_ids(&mut response);
+                                    }
                                     SetResponseMethod::Relay(set_response) => {
                                         set_response.update_created_ids(&mut response);
                                     }
@@ -562,6 +565,13 @@ impl RequestHandler for Server {
                         .await?
                         .into()
                 }
+                // inbuxa: migrations into this server
+                GetRequestMethod::Migration(mut req) => {
+                    resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
+                    crate::inbuxa::migration::get(self, access_token, *req)
+                        .await?
+                        .into()
+                }
                 // inbuxa: inbuxa Relay
                 GetRequestMethod::Relay(mut req) => {
                     resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
@@ -1139,6 +1149,22 @@ impl RequestHandler for Server {
                     .await?
                     .into()
                 }
+                SetRequestMethod::Migration(mut req) => {
+                    resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
+                    let reason = req.arguments.reason.clone();
+                    crate::inbuxa::audit::recorded(
+                        self,
+                        access_token,
+                        session,
+                        &method_name.obj.to_string(),
+                        None,
+                        reason,
+                        *req,
+                        |req| Box::pin(crate::inbuxa::migration::set(self, access_token, req)),
+                    )
+                    .await?
+                    .into()
+                }
                 SetRequestMethod::Relay(mut req) => {
                     resolve_account_id(&mut req.account_id, method_name.obj, access_token)?;
                     let reason = req.arguments.reason.clone();

crates/jmap/src/changes/get.rs

@@ -431,6 +431,7 @@ impl IntermediateChangesResponse {
             | MethodObject::LegalHold
             | MethodObject::HoldExport
             | MethodObject::MailRule
+            | MethodObject::Migration
             | MethodObject::Relay
             | MethodObject::RelayMessage
             | MethodObject::SecurityAcceptance

crates/jmap/src/inbuxa/audit.rs

@@ -488,6 +488,29 @@ async fn fork_current(server: &Server, object: &str, id: &MaybeInvalid<Id>) -> O
             }
             MaybeInvalid::Invalid(_) => None,
         },
+        // A migration's answers and where it is; it holds no secret
+        "inbuxa:Migration" => match id {
+            MaybeInvalid::Value(id) => {
+                inbuxa_features::migration::get(data, u32::try_from(id.id()).ok()?)
+                    .await
+                    .ok()?
+                    .map(|m| {
+                        serde_json::json!({
+                            "name": if m.description.is_empty() {
+                                format!("Migration {}", m.id)
+                            } else {
+                                m.description.clone()
+                            },
+                            "sourceKind": m.source_kind.as_str(),
+                            "source": m.source,
+                            "accounts": m.accounts.len(),
+                            "createAccounts": m.create_accounts,
+                            "status": m.status.as_str(),
+                        })
+                    })
+            }
+            MaybeInvalid::Invalid(_) => None,
+        },
         // A relay as an administrator sets it: never its keys or token
         "inbuxa:Relay" => match id {
             MaybeInvalid::Value(id) => {

crates/jmap/src/inbuxa/migration.rs

@@ -0,0 +1,476 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:Migration` (migrations spec): what to move into this server,
+//! and the runs that move it. A create records a migration; an update's
+//! `action` previews, moves, makes the final pass, stops, hands over the
+//! new passwords or closes it. Destroy is refused: a closed migration stays
+//! listed. Server-wide, so a tenant is refused (MG-10).
+
+use super::relay::json_to_value;
+use common::{Server, auth::AccessToken};
+use inbuxa_features::{
+    audit::Via,
+    migration::{self as model, Migration as Stored, RunKind, Source, SourceKind, Status},
+};
+use jmap_proto::{
+    error::set::SetError,
+    method::{
+        get::{GetRequest, GetResponse},
+        set::{SetRequest, SetResponse},
+    },
+    object::inbuxa_migration::{Migration, MigrationProperty as P, MigrationValue},
+    request::IntoValid,
+    types::date::UTCDate,
+};
+use jmap_tools::{Key, Map, Property, Value};
+use serde_json::json;
+use services::inbuxa_migrations::{self as runs, Refusal};
+use store::write::now;
+use types::id::Id;
+
+type MValue = Value<'static, P, MigrationValue>;
+
+const ALL: &[P] = &[
+    P::Id,
+    P::Description,
+    P::SourceKind,
+    P::Source,
+    P::Accounts,
+    P::CreateAccounts,
+    P::Parallel,
+    P::Status,
+    P::Run,
+    P::Result,
+    P::Error,
+    P::Progress,
+    P::Failures,
+    P::NodeId,
+    P::Heartbeat,
+    P::Moved,
+    P::DiskUsed,
+    P::PasswordsWaiting,
+    P::ReportBlobId,
+    P::CreatedAt,
+    P::CreatedBy,
+    P::StartedAt,
+    P::FinishedAt,
+];
+
+/// What a client may write; everything else the server sets.
+const WRITABLE: &[P] = &[
+    P::Description,
+    P::SourceKind,
+    P::Source,
+    P::Accounts,
+    P::CreateAccounts,
+    P::Parallel,
+];
+
+/// The accounts moved at once unless the console says otherwise.
+const DEFAULT_PARALLEL: u32 = 4;
+
+fn server_level(access_token: &AccessToken) -> trc::Result<()> {
+    if access_token.tenant_id().is_some() {
+        Err(trc::JmapEvent::Forbidden
+            .into_err()
+            .details("A migration writes into the whole server, not a tenant."))
+    } else if matches!(access_token.origin(), Some(Via::Migration { .. })) {
+        Err(trc::JmapEvent::Forbidden
+            .into_err()
+            .details("A migration's run can't manage migrations."))
+    } else {
+        Ok(())
+    }
+}
+
+fn date(seconds: u64) -> serde_json::Value {
+    json!(UTCDate::from_timestamp(seconds as i64).to_string())
+}
+
+fn to_value(m: &Stored, properties: &[P]) -> MValue {
+    let mut out = Map::with_capacity(properties.len());
+    for property in properties {
+        let json = match property {
+            P::Id => {
+                out.insert_unchecked(
+                    Key::Property(P::Id),
+                    Value::Element(MigrationValue::Id(Id::from(m.id))),
+                );
+                continue;
+            }
+            P::Description => json!(m.description),
+            P::SourceKind => json!(m.source_kind.as_str()),
+            P::Source => serde_json::to_value(&m.source).unwrap_or_default(),
+            P::Accounts => json!(m.accounts),
+            P::CreateAccounts => json!(m.create_accounts),
+            P::Parallel => json!(m.parallel),
+            P::Status => json!(m.status.as_str()),
+            P::Run => json!(m.run.map(|r| r.as_str())),
+            P::Result => json!(m.result.map(|r| r.as_str())),
+            P::Error => json!(m.error),
+            P::Progress => serde_json::to_value(&m.progress).unwrap_or_default(),
+            P::Failures => serde_json::to_value(&m.failures).unwrap_or_default(),
+            P::NodeId => json!(m.node_id.map(|n| n.to_string())),
+            P::Heartbeat => m.heartbeat.map_or(serde_json::Value::Null, date),
+            P::Moved => json!(m.moved),
+            P::DiskUsed => json!(m.disk_used),
+            P::PasswordsWaiting => json!(m.passwords_waiting),
+            P::ReportBlobId => json!(m.report_blob_id),
+            P::CreatedAt => date(m.created_at),
+            P::CreatedBy => json!(m.created_by),
+            P::StartedAt => m.started_at.map_or(serde_json::Value::Null, date),
+            P::FinishedAt => m.finished_at.map_or(serde_json::Value::Null, date),
+            // Write-only, or only in a response
+            P::Action | P::Secret | P::PasswordsBlobId => continue,
+        };
+        out.insert_unchecked(Key::Property(property.clone()), json_to_value(json));
+    }
+    Value::Object(out)
+}
+
+/// `inbuxa:Migration/get`: every migration, newest first.
+pub async fn get(
+    server: &Server,
+    access_token: &AccessToken,
+    mut request: GetRequest<Migration>,
+) -> trc::Result<GetResponse<Migration>> {
+    server_level(access_token)?;
+    let properties = request.unwrap_properties(ALL);
+    let (ids, not_found) = request.unwrap_ids(server.core.jmap.get_max_objects)?;
+    let mut response = GetResponse {
+        account_id: request.account_id.into(),
+        state: None,
+        list: Vec::new(),
+        not_found,
+    };
+    let mut migrations = model::all(server.store()).await?;
+    migrations.reverse();
+    match ids {
+        None => response
+            .list
+            .extend(migrations.iter().map(|m| to_value(m, &properties))),
+        Some(ids) => {
+            for id in ids {
+                match migrations.iter().find(|m| u64::from(m.id) == id.id()) {
+                    Some(m) => response.list.push(to_value(m, &properties)),
+                    None => response.push_not_found(id),
+                }
+            }
+        }
+    }
+    Ok(response)
+}
+
+fn invalid(property: P, why: impl Into<String>) -> SetError<P> {
+    SetError::invalid_properties()
+        .with_property(property)
+        .with_description(why.into())
+}
+
+/// What a create or an update sent.
+#[derive(Default)]
+struct Sent {
+    fields: serde_json::Map<String, serde_json::Value>,
+    action: Option<String>,
+    secret: Option<String>,
+}
+
+fn sent(value: Value<'_, P, MigrationValue>) -> Result<Sent, SetError<P>> {
+    let mut sent = Sent::default();
+    for (key, value) in value.into_expanded_object() {
+        match &key {
+            Key::Property(P::Action) => match value.as_str() {
+                Some(action) => sent.action = Some(action.to_string()),
+                None => return Err(invalid(P::Action, "Name the action.")),
+            },
+            Key::Property(P::Secret) => match value.as_str() {
+                Some(secret) => sent.secret = Some(secret.to_string()),
+                None => return Err(invalid(P::Secret, "The secret is text.")),
+            },
+            Key::Property(p) if WRITABLE.contains(p) => {
+                sent.fields
+                    .insert(p.to_cow().into_owned(), serde_json::Value::from(value));
+            }
+            Key::Property(p) => return Err(invalid(p.clone(), "The server sets this.")),
+            _ => {
+                return Err(SetError::invalid_properties()
+                    .with_property(key.into_owned())
+                    .with_description("Unknown property."));
+            }
+        }
+    }
+    Ok(sent)
+}
+
+/// Applies the writable fields to a migration, checked (MG-2, MG-3).
+fn apply(
+    m: &mut Stored,
+    fields: &serde_json::Map<String, serde_json::Value>,
+) -> Result<(), SetError<P>> {
+    for (name, value) in fields {
+        match name.as_str() {
+            "description" => match value.as_str() {
+                Some(text) => m.description = text.trim().to_string(),
+                None => return Err(invalid(P::Description, "The description is text.")),
+            },
+            "sourceKind" => match value.as_str().and_then(SourceKind::parse) {
+                Some(kind) => m.source_kind = kind,
+                None => {
+                    return Err(invalid(
+                        P::SourceKind,
+                        "imap, jmap, microsoft or google. Maildir isn't offered here: the \
+                         folders would have to be on the server.",
+                    ));
+                }
+            },
+            "source" => match serde_json::from_value::<Source>(value.clone()) {
+                Ok(source) => m.source = source,
+                Err(err) => return Err(invalid(P::Source, err.to_string())),
+            },
+            "accounts" => {
+                let Some(items) = value.as_array() else {
+                    return Err(invalid(P::Accounts, "A list of addresses."));
+                };
+                let mut accounts = Vec::with_capacity(items.len());
+                for item in items {
+                    let Some(email) = item.as_str().and_then(model::account_email) else {
+                        return Err(invalid(P::Accounts, format!("{item} isn't an address.")));
+                    };
+                    if !accounts.contains(&email) {
+                        accounts.push(email);
+                    }
+                }
+                m.accounts = accounts;
+            }
+            "createAccounts" => match value.as_bool() {
+                Some(create) => m.create_accounts = create,
+                None => return Err(invalid(P::CreateAccounts, "true or false.")),
+            },
+            "parallel" => match value.as_u64().and_then(|n| u32::try_from(n).ok()) {
+                Some(parallel) => m.parallel = parallel,
+                None => return Err(invalid(P::Parallel, "A number of accounts.")),
+            },
+            _ => {}
+        }
+    }
+    model::validate(m).map_err(|err| {
+        let property = match err.property {
+            "description" => P::Description,
+            "accounts" => P::Accounts,
+            "parallel" => P::Parallel,
+            _ => P::Source,
+        };
+        invalid(property, err.reason)
+    })
+}
+
+fn refused(refusal: Refusal) -> SetError<P> {
+    let error = match refusal {
+        Refusal::NotFound => return SetError::not_found(),
+        Refusal::SecretNeeded => invalid(P::Secret, refusal.describe()),
+        _ => SetError::forbidden(),
+    };
+    error.with_description(refusal.describe())
+}
+
+/// `inbuxa:Migration/set`. The request layer records it, the secret
+/// redacted (AU-3).
+pub async fn set(
+    server: &Server,
+    access_token: &AccessToken,
+    mut request: SetRequest<'_, Migration>,
+) -> trc::Result<SetResponse<Migration>> {
+    server_level(access_token)?;
+    let mut response = SetResponse::from_request(&request, server.core.jmap.set_max_objects)?;
+    let data = server.store();
+    let actor = server.audit_actor(access_token).await;
+
+    for (client_id, value) in request.unwrap_create() {
+        let sent = match sent(value) {
+            Ok(sent) => sent,
+            Err(error) => {
+                response.not_created.append(client_id, error);
+                continue;
+            }
+        };
+        if sent.action.is_some() || sent.secret.is_some() {
+            response.not_created.append(
+                client_id,
+                invalid(
+                    P::Action,
+                    "Create the migration first, then run it with an update.",
+                ),
+            );
+            continue;
+        }
+        let Some(created_by_id) = actor.account_id else {
+            response.not_created.append(
+                client_id,
+                SetError::forbidden().with_description("Sign in as a person to migrate."),
+            );
+            continue;
+        };
+        let mut migration = Stored {
+            id: 0,
+            description: String::new(),
+            source_kind: SourceKind::Imap,
+            source: Source::default(),
+            accounts: Vec::new(),
+            create_accounts: false,
+            parallel: DEFAULT_PARALLEL,
+            status: Status::Idle,
+            run: None,
+            result: None,
+            error: None,
+            progress: Default::default(),
+            failures: Vec::new(),
+            node_id: None,
+            heartbeat: None,
+            moved: false,
+            disk_used: 0,
+            passwords_waiting: false,
+            report_blob_id: None,
+            created_at: now(),
+            created_by: actor.name.clone(),
+            created_by_id,
+            started_at: None,
+            finished_at: None,
+        };
+        if !sent.fields.contains_key("sourceKind") {
+            response.not_created.append(
+                client_id,
+                invalid(P::SourceKind, "Say where the mail is now."),
+            );
+            continue;
+        }
+        if let Err(error) = apply(&mut migration, &sent.fields) {
+            response.not_created.append(client_id, error);
+            continue;
+        }
+        let id = model::create(data, &migration).await?;
+        let mut out = Map::with_capacity(1);
+        out.insert_unchecked(
+            Key::Property(P::Id),
+            Value::Element(MigrationValue::Id(Id::from(id))),
+        );
+        response.created.insert(client_id, Value::Object(out));
+    }
+
+    for (id, value) in request.unwrap_update().into_valid() {
+        let sent = match sent(value) {
+            Ok(sent) => sent,
+            Err(error) => {
+                response.not_updated.append(id, error);
+                continue;
+            }
+        };
+        let Ok(migration_id) = u32::try_from(id.id()) else {
+            response.not_updated.append(id, SetError::not_found());
+            continue;
+        };
+
+        // The answers first: only while nothing runs and before a move
+        if !sent.fields.is_empty() {
+            let mut error = None;
+            let found = model::update(data, migration_id, |m| {
+                error = if m.status != Status::Idle {
+                    Some(SetError::forbidden().with_description(
+                        "Change a migration while nothing runs, and before it's closed.",
+                    ))
+                } else if m.moved
+                    && sent
+                        .fields
+                        .keys()
+                        .any(|k| k == "sourceKind" || k == "source")
+                {
+                    Some(invalid(
+                        P::Source,
+                        "The accounts were moved from this source; start a new migration \
+                         for another.",
+                    ))
+                } else {
+                    let mut next = m.clone();
+                    match apply(&mut next, &sent.fields) {
+                        Ok(()) => {
+                            *m = next;
+                            None
+                        }
+                        Err(err) => Some(err),
+                    }
+                };
+                error.is_none()
+            })
+            .await?;
+            match (found, error) {
+                (None, _) => {
+                    response.not_updated.append(id, SetError::not_found());
+                    continue;
+                }
+                (_, Some(error)) => {
+                    response.not_updated.append(id, error);
+                    continue;
+                }
+                _ => {}
+            }
+        }
+
+        let outcome = match sent.action.as_deref() {
+            None if sent.secret.is_some() => Err(invalid(
+                P::Secret,
+                "Give the secret with the action that needs it.",
+            )),
+            None => Ok(None),
+            Some("preview") => runs::start(server, migration_id, RunKind::Preview, sent.secret)
+                .await
+                .map(|_| None)
+                .map_err(refused),
+            Some("move") => runs::start(server, migration_id, RunKind::Move, sent.secret)
+                .await
+                .map(|_| None)
+                .map_err(refused),
+            Some("finalPass") => runs::start(server, migration_id, RunKind::FinalPass, sent.secret)
+                .await
+                .map(|_| None)
+                .map_err(refused),
+            Some("stop") => runs::stop(server, migration_id)
+                .await
+                .map(|_| None)
+                .map_err(refused),
+            Some("close") => runs::close(server, migration_id)
+                .await
+                .map(|_| None)
+                .map_err(refused),
+            Some("passwords") => runs::take_passwords(server, migration_id)
+                .await
+                .map(|blob| {
+                    let mut out = Map::with_capacity(1);
+                    out.insert_unchecked(
+                        Key::Property(P::PasswordsBlobId),
+                        Value::Str(blob.into()),
+                    );
+                    Some(Value::Object(out))
+                })
+                .map_err(refused),
+            Some(_) => Err(invalid(
+                P::Action,
+                "preview, move, finalPass, stop, passwords or close.",
+            )),
+        };
+        match outcome {
+            Ok(value) => response.updated.append(id, value),
+            Err(error) => response.not_updated.append(id, error),
+        }
+    }
+
+    for id in request.unwrap_destroy().into_valid() {
+        response.not_destroyed.append(
+            id,
+            SetError::forbidden().with_description("Close the migration instead; it stays listed."),
+        );
+    }
+    Ok(response)
+}

crates/jmap/src/inbuxa/mod.rs

@@ -30,6 +30,7 @@ pub mod log_settings;
 pub mod mail_rule;
 pub mod masked_email;
 pub mod masked_secrets; // inbuxa: "****" is never saved as a secret
+pub mod migration; // inbuxa: migrations into this server
 pub mod oidc_directory; // inbuxa: http-directory-services-13
 pub mod protocol_policy;
 pub mod relay; // inbuxa: inbuxa Relay

crates/registry/src/schema/enums.rs

@@ -1789,6 +1789,8 @@ pub enum Permission {
     SysRelayUpdate = 691,
     SysRelayQueueGet = 692,
     SysRelayQueueUpdate = 693,
+    // inbuxa: migrations into this server
+    SysMigrationRun = 694,
     SysAccountGet = 219,
     SysAccountCreate = 220,
     SysAccountUpdate = 221,

crates/registry/src/schema/enums_impl.rs

@@ -7170,6 +7170,7 @@ impl EnumImpl for Permission {
             b"sysRelayUpdate" => Permission::SysRelayUpdate,
             b"sysRelayQueueGet" => Permission::SysRelayQueueGet,
             b"sysRelayQueueUpdate" => Permission::SysRelayQueueUpdate,
+            b"sysMigrationRun" => Permission::SysMigrationRun,
             b"sysAccountGet" => Permission::SysAccountGet,
             b"sysAccountCreate" => Permission::SysAccountCreate,
             b"sysAccountUpdate" => Permission::SysAccountUpdate,
@@ -7880,6 +7881,7 @@ impl EnumImpl for Permission {
             Permission::SysRelayUpdate => "sysRelayUpdate",
             Permission::SysRelayQueueGet => "sysRelayQueueGet",
             Permission::SysRelayQueueUpdate => "sysRelayQueueUpdate",
+            Permission::SysMigrationRun => "sysMigrationRun",
             Permission::SysAccountGet => "sysAccountGet",
             Permission::SysAccountCreate => "sysAccountCreate",
             Permission::SysAccountUpdate => "sysAccountUpdate",
@@ -8583,6 +8585,7 @@ impl EnumImpl for Permission {
             691 => Some(Permission::SysRelayUpdate),
             692 => Some(Permission::SysRelayQueueGet),
             693 => Some(Permission::SysRelayQueueUpdate),
+            694 => Some(Permission::SysMigrationRun),
             219 => Some(Permission::SysAccountGet),
             220 => Some(Permission::SysAccountCreate),
             221 => Some(Permission::SysAccountUpdate),
@@ -9027,7 +9030,7 @@ impl EnumImpl for Permission {
         }
     }
 
-    const COUNT: usize = 694;
+    const COUNT: usize = 695;
 }
 
 impl serde::Serialize for Permission {

crates/services/Cargo.toml

@@ -19,7 +19,7 @@ jmap-tools = { version = "0.1" }
 registry = { path =  "../registry" }
 inbuxa-features = { path = "../features" }
 smtp-proto = { version = "0.2", features = ["rkyv", "serde"] }
-tokio = { version = "1.53", features = ["rt", "net", "time", "sync", "macros"] }
+tokio = { version = "1.53", features = ["rt", "net", "time", "sync", "macros", "process", "fs", "io-util"] }
 mail-parser = { version = "0.11", features = ["full_encoding", "rkyv"] }
 mail-builder = { version = "1.0" }
 aho-corasick = { version = "1.1" }

crates/services/src/inbuxa_migrations.rs

@@ -0,0 +1,888 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! Running migrations (migrations spec, MG-4 to MG-14): the bundled
+//! inbuxa-migrate, as a child of this node, on a migration's answers.
+//!
+//! A run writes the answers into the migration's work directory, has
+//! `inbuxa-migrate plan` turn them into plans, and runs `migrate` on one with
+//! `--progress json`, reading its events to keep the migration's record up
+//! to date. The child writes into this server through this node's own HTTP
+//! listener, on loopback, with a run token that acts as the administrator
+//! who started the migration (MG-10).
+//!
+//! The source's secret is given to the child through its environment and is
+//! held in this node's memory between runs, never written anywhere (MG-12).
+//! A restart forgets it, and the next run asks for it again.
+
+use common::{BuildServer, Inner, Server, migration_token};
+use inbuxa_features::migration::{self as model, Failure, Migration, RunKind, RunResult, Status};
+use registry::schema::{
+    enums::NetworkListenerProtocol,
+    prelude::ObjectType,
+    structs::{NetworkListener, SystemSettings},
+};
+use serde_json::Value;
+use std::{
+    collections::HashMap,
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+    path::{Path, PathBuf},
+    process::Stdio,
+    sync::{Arc, LazyLock},
+    time::Duration,
+};
+use store::{registry::RegistryQuery, write::now};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncReadExt, BufReader},
+    process::Command,
+    sync::{Mutex, Notify},
+};
+use types::id::Id;
+
+/// MG-8: progress is saved at least this often while a run goes on.
+const SAVE_EVERY: Duration = Duration::from_secs(30);
+/// MG-11: a run whose node hasn't said anything for this long is taken to
+/// have stopped with its node.
+const STALE_AFTER: u64 = 5 * 60;
+/// How much of the engine's stderr is kept to say why a run failed.
+const KEEP_STDERR: usize = 4_000;
+/// The variables the child is given; everything else named like them is
+/// cleared, so it sees only these secrets.
+const SOURCE_SECRET_VAR: &str = "MIGRATE_SOURCE_SECRET";
+const TARGET_TOKEN_VAR: &str = "MIGRATE_INBUXA_TOKEN";
+/// The child's `PATH`: the rest of its environment is cleared.
+const SAFE_PATH: &str = "/usr/local/bin:/usr/bin:/bin";
+
+/// The secrets this node holds, by migration (MG-12).
+static SECRETS: LazyLock<Mutex<HashMap<u32, String>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+/// The runs going on on this node, each with what stops it.
+static RUNS: LazyLock<Mutex<HashMap<u32, Arc<Notify>>>> =
+    LazyLock::new(|| Mutex::new(HashMap::new()));
+
+/// Why a run couldn't start, or a migration couldn't be stopped or closed,
+/// in the console's words.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum Refusal {
+    NotFound,
+    Closed,
+    Running,
+    NotRunning,
+    /// MG-12: no secret was given and this node doesn't hold one.
+    SecretNeeded,
+    /// MG-11: the migration's work directory is on another node.
+    OtherNode(u64),
+    /// MG-5: a final pass comes after a move.
+    NotMoved,
+    NoPasswords,
+    Failed(String),
+}
+
+impl Refusal {
+    pub fn describe(&self) -> String {
+        match self {
+            Refusal::NotFound => "No such migration.".into(),
+            Refusal::Closed => "This migration is closed; start a new one.".into(),
+            Refusal::Running => "A run is going on; stop it first, or wait for it.".into(),
+            Refusal::NotRunning => "Nothing is running.".into(),
+            Refusal::SecretNeeded => {
+                "Give the source's password, API key, client secret or key again: \
+                 the server doesn't keep it across a restart."
+                    .into()
+            }
+            Refusal::OtherNode(node) => format!(
+                "This migration's files are on node {node}. Its runs happen there; \
+                 open the console through that node."
+            ),
+            Refusal::NotMoved => "Move the accounts first; the final pass comes after.".into(),
+            Refusal::NoPasswords => "There are no new passwords waiting.".into(),
+            Refusal::Failed(why) => why.clone(),
+        }
+    }
+}
+
+fn node_id(server: &Server) -> u64 {
+    server.core.network.node_id
+}
+
+/// MG-4: where the migrations' work directories are: `$INBUXA_MIGRATIONS_DIR`,
+/// or `migrations` in the server's working directory, which the image makes
+/// its data volume.
+pub fn migrations_dir() -> PathBuf {
+    types::branding::env_var("MIGRATIONS_DIR")
+        .ok()
+        .filter(|dir| !dir.is_empty())
+        .map(PathBuf::from)
+        .unwrap_or_else(|| {
+            std::env::current_dir()
+                .unwrap_or_else(|_| PathBuf::from("."))
+                .join("migrations")
+        })
+}
+
+fn work_dir(id: u32) -> PathBuf {
+    migrations_dir().join(id.to_string())
+}
+
+/// MG-4: the inbuxa-migrate the image carries, or `$INBUXA_MIGRATE_BIN`.
+fn engine() -> PathBuf {
+    types::branding::env_var("MIGRATE_BIN")
+        .ok()
+        .filter(|bin| !bin.is_empty())
+        .map(PathBuf::from)
+        .unwrap_or_else(|| PathBuf::from("/usr/local/bin/inbuxa-migrate"))
+}
+
+/// MG-10: an address on this node's loopback that its HTTP listeners
+/// answer, plain HTTP first, and whether it's TLS. A listener bound to one
+/// address other than loopback is left out: the run token is only taken
+/// from loopback. So is one that expects the PROXY protocol from loopback.
+async fn local_url(server: &Server) -> Result<(String, bool), String> {
+    if let Ok(url) = types::branding::env_var("MIGRATIONS_URL")
+        && !url.is_empty()
+    {
+        return Ok((
+            url.trim_end_matches('/').to_string(),
+            url.starts_with("https:"),
+        ));
+    }
+    let registry = server.registry();
+    let system = registry
+        .object::<SystemSettings>(Id::singleton())
+        .await
+        .map_err(|err| err.to_string())?
+        .unwrap_or_default();
+    let ids = registry
+        .query::<Vec<Id>>(RegistryQuery::new(ObjectType::NetworkListener))
+        .await
+        .map_err(|err| err.to_string())?;
+    let mut candidates = Vec::new();
+    for id in ids {
+        let Ok(Some(listener)) = registry.object::<NetworkListener>(id).await else {
+            continue;
+        };
+        if listener.protocol != NetworkListenerProtocol::Http {
+            continue;
+        }
+        let proxies = if listener.override_proxy_trusted_networks.is_empty() {
+            system.proxy_trusted_networks.as_slice()
+        } else {
+            listener.override_proxy_trusted_networks.as_slice()
+        };
+        for bind in listener.bind.iter() {
+            let bind = bind.0;
+            let ip = match bind.ip() {
+                IpAddr::V4(ip) if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
+                IpAddr::V6(ip) if ip.is_unspecified() => IpAddr::V6(Ipv6Addr::LOCALHOST),
+                ip if ip.is_loopback() => ip,
+                _ => continue,
+            };
+            if proxies.iter().any(|network| network.matches(&ip)) {
+                continue;
+            }
+            // HTTP is TLS from the first byte or not at all
+            candidates.push((listener.tls_implicit, SocketAddr::new(ip, bind.port())));
+        }
+    }
+    // Plain HTTP first; a node whose role leaves a listener off doesn't
+    // answer on it
+    candidates.sort_by_key(|(tls, addr)| (*tls, addr.is_ipv6()));
+    for (tls, addr) in candidates {
+        let answers =
+            tokio::time::timeout(Duration::from_secs(2), tokio::net::TcpStream::connect(addr))
+                .await
+                .is_ok_and(|connected| connected.is_ok());
+        if answers {
+            let scheme = if tls { "https" } else { "http" };
+            return Ok((format!("{scheme}://{addr}"), tls));
+        }
+    }
+    Err(
+        "No HTTP listener answers on this node's loopback. Add one, or set \
+         INBUXA_MIGRATIONS_URL."
+            .into(),
+    )
+}
+
+/// Takes the migration for a run on this node, or says why not.
+async fn claim(server: &Server, id: u32, kind: RunKind) -> Result<Migration, Refusal> {
+    let node = node_id(server);
+    let mut refusal = None;
+    let claimed = model::update(server.store(), id, |m| {
+        refusal = match (m.status, m.node_id) {
+            (Status::Closed, _) => Some(Refusal::Closed),
+            (Status::Running, _) => Some(Refusal::Running),
+            (_, Some(other)) if other != node => Some(Refusal::OtherNode(other)),
+            _ if kind == RunKind::FinalPass && !m.moved => Some(Refusal::NotMoved),
+            _ => None,
+        };
+        if refusal.is_some() {
+            return false;
+        }
+        m.status = Status::Running;
+        m.run = Some(kind);
+        m.result = None;
+        m.error = None;
+        m.node_id = Some(node);
+        m.heartbeat = Some(now());
+        m.progress = Default::default();
+        m.failures.clear();
+        m.started_at = Some(now());
+        m.finished_at = None;
+        true
+    })
+    .await
+    .map_err(|err| Refusal::Failed(err.to_string()))?;
+    match (claimed, refusal) {
+        (None, _) => Err(Refusal::NotFound),
+        (_, Some(refusal)) => Err(refusal),
+        (Some(m), None) => Ok(m),
+    }
+}
+
+/// MG-5: starts a run of `kind`. `secret`, if given, replaces the one this
+/// node holds.
+pub async fn start(
+    server: &Server,
+    id: u32,
+    kind: RunKind,
+    secret: Option<String>,
+) -> Result<(), Refusal> {
+    let secret = {
+        let mut secrets = SECRETS.lock().await;
+        if let Some(secret) = secret.filter(|s| !s.is_empty()) {
+            secrets.insert(id, secret);
+        }
+        secrets.get(&id).cloned()
+    };
+    let Some(secret) = secret else {
+        return match model::get(server.store(), id).await {
+            Ok(None) => Err(Refusal::NotFound),
+            _ => Err(Refusal::SecretNeeded),
+        };
+    };
+    let migration = claim(server, id, kind).await?;
+    let stop = Arc::new(Notify::new());
+    RUNS.lock().await.insert(id, stop.clone());
+    let server = server.clone();
+    tokio::spawn(async move {
+        let ended = run(&server, &migration, kind, &secret, &stop).await;
+        RUNS.lock().await.remove(&id);
+        if let Err(err) = finish(&server, &migration, kind, ended).await {
+            trc::error!(err.details("Failed to save a migration's result"));
+        }
+    });
+    Ok(())
+}
+
+/// MG-5: stops the run going on.
+pub async fn stop(server: &Server, id: u32) -> Result<(), Refusal> {
+    if let Some(stop) = RUNS.lock().await.get(&id) {
+        stop.notify_one();
+        return Ok(());
+    }
+    let node = node_id(server);
+    let mut refusal = None;
+    let found = model::update(server.store(), id, |m| {
+        refusal = match (m.status, m.node_id) {
+            (Status::Running, Some(other)) if other != node => Some(Refusal::OtherNode(other)),
+            // Said to be running here, but nothing is: it stopped with an
+            // earlier start of this node
+            (Status::Running, _) => None,
+            _ => Some(Refusal::NotRunning),
+        };
+        if refusal.is_some() {
+            return false;
+        }
+        m.status = Status::Idle;
+        m.result = Some(RunResult::Interrupted);
+        m.finished_at = Some(now());
+        true
+    })
+    .await
+    .map_err(|err| Refusal::Failed(err.to_string()))?;
+    match (found, refusal) {
+        (None, _) => Err(Refusal::NotFound),
+        (_, Some(refusal)) => Err(refusal),
+        _ => Ok(()),
+    }
+}
+
+/// MG-14: closes a migration: its work directory, archives and all, goes,
+/// and so does the secret. It stays listed.
+pub async fn close(server: &Server, id: u32) -> Result<(), Refusal> {
+    let node = node_id(server);
+    let mut refusal = None;
+    let mut remove = false;
+    let found = model::update(server.store(), id, |m| {
+        refusal = match (m.status, m.node_id) {
+            (Status::Closed, _) => Some(Refusal::Closed),
+            (Status::Running, _) => Some(Refusal::Running),
+            (_, Some(other)) if other != node => Some(Refusal::OtherNode(other)),
+            _ => None,
+        };
+        if refusal.is_some() {
+            return false;
+        }
+        remove = m.node_id.is_some();
+        m.status = Status::Closed;
+        m.passwords_waiting = false;
+        m.disk_used = 0;
+        true
+    })
+    .await
+    .map_err(|err| Refusal::Failed(err.to_string()))?;
+    match (found, refusal) {
+        (None, _) => return Err(Refusal::NotFound),
+        (_, Some(refusal)) => return Err(refusal),
+        _ => {}
+    }
+    SECRETS.lock().await.remove(&id);
+    if remove {
+        let dir = work_dir(id);
+        if let Err(err) = tokio::fs::remove_dir_all(&dir).await
+            && err.kind() != std::io::ErrorKind::NotFound
+        {
+            trc::error!(
+                trc::StoreEvent::UnexpectedError
+                    .into_err()
+                    .details(format!(
+                        "Failed to remove a closed migration's work directory, {}",
+                        dir.display()
+                    ))
+                    .reason(err)
+            );
+        }
+    }
+    Ok(())
+}
+
+/// MG-13: the new accounts' passwords, once: the file is deleted as it's
+/// read. The CSV goes into a blob of whoever started the migration.
+pub async fn take_passwords(server: &Server, id: u32) -> Result<String, Refusal> {
+    let migration = model::get(server.store(), id)
+        .await
+        .map_err(|err| Refusal::Failed(err.to_string()))?
+        .ok_or(Refusal::NotFound)?;
+    match migration.node_id {
+        Some(node) if node != node_id(server) => return Err(Refusal::OtherNode(node)),
+        _ => {}
+    }
+    if !migration.passwords_waiting {
+        return Err(Refusal::NoPasswords);
+    }
+    let mut files = passwords_files(&work_dir(id)).await;
+    let mut csv = Vec::new();
+    for path in &files {
+        if let Ok(bytes) = tokio::fs::read(path).await {
+            // One header for every file, the first's
+            let skip = if csv.is_empty() {
+                0
+            } else {
+                bytes
+                    .iter()
+                    .position(|b| *b == b'\n')
+                    .map_or(bytes.len(), |n| n + 1)
+            };
+            csv.extend_from_slice(&bytes[skip..]);
+        }
+    }
+    if csv.is_empty() {
+        return Err(Refusal::NoPasswords);
+    }
+    let blob = server
+        .put_jmap_blob(migration.created_by_id, &csv)
+        .await
+        .map_err(|err| Refusal::Failed(err.to_string()))?;
+    for path in files.drain(..) {
+        let _ = tokio::fs::remove_file(path).await;
+    }
+    model::update(server.store(), id, |m| {
+        m.passwords_waiting = false;
+        true
+    })
+    .await
+    .map_err(|err| Refusal::Failed(err.to_string()))?;
+    Ok(blob.to_string())
+}
+
+async fn passwords_files(dir: &Path) -> Vec<PathBuf> {
+    let mut files = Vec::new();
+    if let Ok(mut entries) = tokio::fs::read_dir(dir).await {
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            let name = entry.file_name();
+            let name = name.to_string_lossy();
+            if name.starts_with("passwords-") && name.ends_with(".csv") {
+                files.push(entry.path());
+            }
+        }
+    }
+    files.sort();
+    files
+}
+
+/// How a run ended, before it's saved.
+struct Ended {
+    result: RunResult,
+    error: Option<String>,
+    progress: model::Progress,
+    failures: Vec<Failure>,
+}
+
+impl Ended {
+    fn failed(error: impl Into<String>) -> Self {
+        Ended {
+            result: RunResult::Failed,
+            error: Some(error.into()),
+            progress: Default::default(),
+            failures: Vec::new(),
+        }
+    }
+}
+
+/// Runs the engine to the end, or until it's stopped.
+async fn run(
+    server: &Server,
+    migration: &Migration,
+    kind: RunKind,
+    secret: &str,
+    stop: &Notify,
+) -> Ended {
+    let dir = work_dir(migration.id);
+    if let Err(err) = private_dir(&dir).await {
+        return Ended::failed(format!(
+            "The work directory {} can't be made: {err}",
+            dir.display()
+        ));
+    }
+    let (url, tls) = match local_url(server).await {
+        Ok(found) => found,
+        Err(err) => return Ended::failed(err),
+    };
+
+    // The plans, from the answers
+    let answers = migration.answers(&url, tls);
+    if let Err(err) = write_private(
+        &dir.join("answers.json"),
+        &serde_json::to_vec_pretty(&answers).unwrap_or_default(),
+    )
+    .await
+    {
+        return Ended::failed(format!("The answers can't be written: {err}"));
+    }
+    let planned = Command::new(engine())
+        .args(["plan", "--answers", "answers.json", "."])
+        .current_dir(&dir)
+        .env_clear()
+        .env("PATH", SAFE_PATH)
+        .stdin(Stdio::null())
+        .output()
+        .await;
+    let plans: Value = match planned {
+        Ok(out) if out.status.success() => {
+            serde_json::from_slice(&out.stdout).unwrap_or(Value::Null)
+        }
+        Ok(out) => return Ended::failed(tail(&String::from_utf8_lossy(&out.stderr))),
+        Err(err) => {
+            return Ended::failed(format!(
+                "inbuxa-migrate couldn't be started ({}): {err}",
+                engine().display()
+            ));
+        }
+    };
+    let plan = match kind {
+        RunKind::Preview => plans["preview"].as_str(),
+        _ => plans["plan"].as_str(),
+    }
+    .unwrap_or(if kind == RunKind::Preview {
+        "preview-plan.json"
+    } else {
+        "plan.json"
+    })
+    .to_string();
+
+    // The run, with a token that ends with it
+    let token = migration_token::issue(migration.created_by_id, migration.id);
+    let mut command = Command::new(engine());
+    command
+        .args([
+            "migrate",
+            "--plan",
+            &plan,
+            "--progress",
+            "json",
+            "--parallel",
+        ])
+        .arg(migration.parallel.max(1).to_string())
+        .args(match kind {
+            RunKind::Preview => &["--dry-run"][..],
+            RunKind::FinalPass => &["--again"][..],
+            RunKind::Move => &[][..],
+        })
+        .arg("work")
+        .current_dir(&dir)
+        .env_clear()
+        .env("PATH", SAFE_PATH)
+        .env(SOURCE_SECRET_VAR, secret)
+        .env(TARGET_TOKEN_VAR, &token)
+        .stdin(Stdio::null())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::piped())
+        .kill_on_drop(true);
+    let mut child = match command.spawn() {
+        Ok(child) => child,
+        Err(err) => {
+            migration_token::revoke(&token);
+            return Ended::failed(format!("inbuxa-migrate couldn't be started: {err}"));
+        }
+    };
+
+    let mut stderr = child.stderr.take();
+    let errors = tokio::spawn(async move {
+        let mut text = String::new();
+        if let Some(stderr) = stderr.as_mut() {
+            let _ = stderr.read_to_string(&mut text).await;
+        }
+        text
+    });
+    let mut lines = child.stdout.take().map(|out| BufReader::new(out).lines());
+
+    let mut ended = Ended {
+        result: RunResult::Done,
+        error: None,
+        progress: Default::default(),
+        failures: Vec::new(),
+    };
+    let mut stopped = false;
+    let mut saved = tokio::time::Instant::now();
+    loop {
+        let line = async {
+            match lines.as_mut() {
+                Some(lines) => lines.next_line().await.ok().flatten(),
+                None => None,
+            }
+        };
+        tokio::select! {
+            line = line => {
+                let Some(line) = line else { break };
+                let account_ended = progress(&mut ended, &line);
+                if account_ended || saved.elapsed() >= SAVE_EVERY {
+                    save_progress(server, migration.id, &ended).await;
+                    saved = tokio::time::Instant::now();
+                }
+            }
+            _ = tokio::time::sleep_until(saved + SAVE_EVERY) => {
+                save_progress(server, migration.id, &ended).await;
+                saved = tokio::time::Instant::now();
+            }
+            _ = stop.notified() => {
+                stopped = true;
+                let _ = child.start_kill();
+                break;
+            }
+        }
+    }
+    let status = child.wait().await;
+    migration_token::revoke(&token);
+    let errors = errors.await.unwrap_or_default();
+    if stopped {
+        ended.result = RunResult::Stopped;
+        return ended;
+    }
+    match status.ok().and_then(|s| s.code()) {
+        Some(0) => ended.result = RunResult::Done,
+        // migrate's own code for "some account didn't finish"
+        Some(5) => ended.result = RunResult::Unfinished,
+        _ => {
+            ended.result = RunResult::Failed;
+            ended.error = Some(tail(&errors));
+        }
+    }
+    ended
+}
+
+/// Takes one of the engine's events into the run's progress. True when an
+/// account ended, which is worth saving at once.
+fn progress(ended: &mut Ended, line: &str) -> bool {
+    let Ok(event) = serde_json::from_str::<Value>(line) else {
+        return false;
+    };
+    let number = |value: &Value| value.as_u64().unwrap_or(0);
+    let p = &mut ended.progress;
+    match event["event"].as_str() {
+        Some("start") => {
+            p.total = number(&event["accounts"]);
+            false
+        }
+        Some("step") if event.get("result").is_some() => {
+            let counts = &event["counts"];
+            if event["step"] == "export" {
+                p.written += number(&counts["created"]) + number(&counts["updated"]);
+            } else {
+                p.read += number(&counts["created"]) + number(&counts["updated"]);
+            }
+            p.failed += number(&counts["failed"]);
+            let result = event["result"].as_str().unwrap_or_default();
+            if result == "failed" || result == "partial" {
+                ended.failures.push(Failure {
+                    email: event["email"].as_str().unwrap_or_default().to_string(),
+                    step: event["step"].as_str().unwrap_or_default().to_string(),
+                    detail: event["detail"]
+                        .as_str()
+                        .unwrap_or_default()
+                        .chars()
+                        .take(500)
+                        .collect(),
+                });
+                let over = ended.failures.len().saturating_sub(model::MAX_FAILURES);
+                ended.failures.drain(..over);
+            }
+            false
+        }
+        Some("account") => {
+            p.finished = number(&event["n"]).max(p.finished);
+            if event["result"] != "done" {
+                p.unfinished += 1;
+            }
+            true
+        }
+        Some("end") => {
+            p.unfinished = number(&event["unfinished"]);
+            true
+        }
+        _ => false,
+    }
+}
+
+/// MG-8: the run's progress and a heartbeat, written over whatever the
+/// record says. A failed write is tried again at the next.
+async fn save_progress(server: &Server, id: u32, ended: &Ended) {
+    let result = model::update(server.store(), id, |m| {
+        if m.status != Status::Running {
+            return false;
+        }
+        m.progress = ended.progress.clone();
+        m.failures.clone_from(&ended.failures);
+        m.heartbeat = Some(now());
+        true
+    })
+    .await;
+    if let Err(err) = result {
+        trc::error!(err.details("Failed to save a migration's progress"));
+    }
+}
+
+/// MG-6: a run's end, with its report for whoever started it.
+async fn finish(
+    server: &Server,
+    migration: &Migration,
+    kind: RunKind,
+    ended: Ended,
+) -> trc::Result<()> {
+    let dir = work_dir(migration.id);
+    let report = dir.join("work").join(if kind == RunKind::Preview {
+        "dry-run-report.csv"
+    } else {
+        "report.csv"
+    });
+    let report_blob = match tokio::fs::read(&report).await {
+        Ok(bytes) if !bytes.is_empty() => Some(
+            server
+                .put_jmap_blob(migration.created_by_id, &bytes)
+                .await?
+                .to_string(),
+        ),
+        _ => None,
+    };
+    let passwords = !passwords_files(&dir).await.is_empty();
+    let disk_used = disk_used(&dir).await;
+    model::update(server.store(), migration.id, |m| {
+        m.status = Status::Idle;
+        m.result = Some(ended.result);
+        m.error.clone_from(&ended.error);
+        m.progress = ended.progress.clone();
+        m.failures.clone_from(&ended.failures);
+        m.heartbeat = None;
+        m.finished_at = Some(now());
+        m.disk_used = disk_used;
+        m.passwords_waiting = passwords;
+        if report_blob.is_some() {
+            m.report_blob_id.clone_from(&report_blob);
+        }
+        if kind != RunKind::Preview
+            && matches!(ended.result, RunResult::Done | RunResult::Unfinished)
+        {
+            m.moved = true;
+        }
+        true
+    })
+    .await
+    .map(|_| ())
+}
+
+/// MG-11, MG-12: at start-up, a run this node had is over: it stopped with
+/// the node. Then, now and then, a run on a node that went quiet is taken
+/// to have stopped with it.
+pub fn spawn_migrations(inner: Arc<Inner>) {
+    tokio::spawn(async move {
+        let server = inner.build_server();
+        let node = node_id(&server);
+        mark_interrupted(&server, |m| m.node_id == Some(node)).await;
+        loop {
+            tokio::time::sleep(Duration::from_secs(60)).await;
+            let server = inner.build_server();
+            let cutoff = now().saturating_sub(STALE_AFTER);
+            mark_interrupted(&server, |m| {
+                m.node_id != Some(node) && m.heartbeat.is_some_and(|at| at < cutoff)
+            })
+            .await;
+        }
+    });
+}
+
+async fn mark_interrupted(server: &Server, which: impl Fn(&Migration) -> bool) {
+    let migrations = match model::all(server.store()).await {
+        Ok(migrations) => migrations,
+        Err(err) => {
+            trc::error!(err.details("Failed to read the migrations"));
+            return;
+        }
+    };
+    for migration in migrations
+        .into_iter()
+        .filter(|m| m.status == Status::Running && which(m))
+    {
+        let result = model::update(server.store(), migration.id, |m| {
+            if m.status != Status::Running || !which(m) {
+                return false;
+            }
+            m.status = Status::Idle;
+            m.result = Some(RunResult::Interrupted);
+            m.heartbeat = None;
+            m.finished_at = Some(now());
+            true
+        })
+        .await;
+        if let Err(err) = result {
+            trc::error!(err.details("Failed to mark an interrupted migration"));
+        }
+    }
+}
+
+fn tail(text: &str) -> String {
+    let text = text.trim();
+    let start = text.len().saturating_sub(KEEP_STDERR);
+    let start = (start..text.len())
+        .find(|i| text.is_char_boundary(*i))
+        .unwrap_or(text.len());
+    if text.is_empty() {
+        "inbuxa-migrate stopped without saying why.".into()
+    } else {
+        text[start..].to_string()
+    }
+}
+
+async fn disk_used(dir: &Path) -> u64 {
+    let mut total = 0;
+    let mut pending = vec![dir.to_path_buf()];
+    while let Some(dir) = pending.pop() {
+        let Ok(mut entries) = tokio::fs::read_dir(&dir).await else {
+            continue;
+        };
+        while let Ok(Some(entry)) = entries.next_entry().await {
+            match entry.metadata().await {
+                Ok(meta) if meta.is_dir() => pending.push(entry.path()),
+                Ok(meta) => total += meta.len(),
+                Err(_) => {}
+            }
+        }
+    }
+    total
+}
+
+async fn private_dir(dir: &Path) -> std::io::Result<()> {
+    tokio::fs::create_dir_all(dir).await?;
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        tokio::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)).await?;
+    }
+    Ok(())
+}
+
+async fn write_private(path: &Path, bytes: &[u8]) -> std::io::Result<()> {
+    let mut options = tokio::fs::OpenOptions::new();
+    options.write(true).create(true).truncate(true);
+    #[cfg(unix)]
+    options.mode(0o600);
+    let mut file = options.open(path).await?;
+    tokio::io::AsyncWriteExt::write_all(&mut file, bytes).await
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn ended() -> Ended {
+        Ended {
+            result: RunResult::Done,
+            error: None,
+            progress: Default::default(),
+            failures: Vec::new(),
+        }
+    }
+
+    #[test]
+    fn the_engines_events_make_the_progress() {
+        let mut e = ended();
+        assert!(!progress(
+            &mut e,
+            r#"{"event":"start","accounts":2,"parallel":1,"dryRun":false}"#
+        ));
+        assert!(!progress(
+            &mut e,
+            r#"{"event":"step","email":"[email protected]","step":"import"}"#
+        ));
+        assert!(!progress(
+            &mut e,
+            r#"{"event":"step","email":"a@x.org","step":"import","result":"done",
+                "counts":{"fetched":9,"created":7,"updated":1,"deleted":0,"skipped":0,"failed":0},"detail":""}"#
+        ));
+        assert!(!progress(
+            &mut e,
+            r#"{"event":"step","email":"a@x.org","step":"export","result":"partial",
+                "counts":{"fetched":0,"created":6,"updated":0,"deleted":0,"skipped":0,"failed":2},
+                "detail":"2 messages over the size limit"}"#
+        ));
+        assert!(progress(
+            &mut e,
+            r#"{"event":"account","email":"[email protected]","n":1,"total":2,"result":"partial"}"#
+        ));
+        assert!(!progress(&mut e, "not json"));
+        assert_eq!(e.progress.total, 2);
+        assert_eq!(e.progress.read, 8);
+        assert_eq!(e.progress.written, 6);
+        assert_eq!(e.progress.failed, 2);
+        assert_eq!(e.progress.finished, 1);
+        assert_eq!(e.progress.unfinished, 1);
+        assert_eq!(e.failures.len(), 1);
+        assert_eq!(e.failures[0].step, "export");
+        assert!(progress(
+            &mut e,
+            r#"{"event":"end","done":1,"unfinished":1,"report":"r"}"#
+        ));
+        assert_eq!(e.progress.unfinished, 1);
+    }
+
+    #[test]
+    fn a_failure_says_the_end_of_what_the_engine_said() {
+        assert_eq!(tail("  \n"), "inbuxa-migrate stopped without saying why.");
+        let long = format!("{}é{}", "x".repeat(KEEP_STDERR), "end");
+        let kept = tail(&long);
+        assert!(kept.ends_with("end"));
+        assert!(kept.len() <= KEEP_STDERR + 1);
+    }
+}

crates/services/src/lib.rs

@@ -26,6 +26,7 @@ pub mod broadcast;
 pub mod inbuxa_deliverability; // inbuxa: the deliverability check
 pub mod inbuxa_lock_expiry;
 pub mod inbuxa_log_retention; // inbuxa: personal-data catalog, D1
+pub mod inbuxa_migrations; // inbuxa: migrations into this server
 pub mod inbuxa_relay_certificates; // inbuxa: inbuxa Relays' certificates
 pub mod inbuxa_relay_link; // inbuxa: links to inbuxa Relays
 pub mod inbuxa_scheduled_reports; // inbuxa: scheduled reports and the weekly digest
@@ -80,6 +81,9 @@ impl SpawnServices for IpcReceivers {
             // inbuxa: deliverability spec, DL-14: each node checks itself daily
             inbuxa_deliverability::spawn_deliverability(inner.clone());
 
+            // inbuxa: migrations spec, MG-11: a run this node had stopped with it
+            inbuxa_migrations::spawn_migrations(inner.clone());
+
             // inbuxa: relay spec, "The link service": every node links to every relay
             inbuxa_relay_link::spawn_relay_links(inner.clone());
 

docs/spec/SPEC.md

@@ -384,6 +384,8 @@ Not a rebuild: the **deliverability check** is INBUXA's own design (inbuxa-draft
 
 Not a rebuild: **scheduled reports and the weekly digest** are INBUXA's own design (inbuxa-drafts `specs/scheduled-reports.md`). An administrator picks sections, a schedule in a time zone and recipients on this server; every node looks for due reports once a minute, one claims each run with the task lock, and the report is built from data the server already keeps and mailed DKIM-signed: `inbuxa:ScheduledReport`, `inbuxa:ScheduledReportSettings` and `inbuxa:ReportExport` (a download: a ZIP of a summary and CSVs) (`crates/jmap/src/inbuxa/scheduled_reports.rs`, `crates/features/src/scheduled_reports/`, `crates/services/src/inbuxa_scheduled_reports.rs`, `crates/smtp/src/reporting/inbuxa_send.rs`), and the `sysScheduledReportGet` and `sysScheduledReportUpdate` permissions. The weekly digest is a built-in report, on by default.
 
+Not a rebuild: **migrations into this server** are INBUXA's own design (`features/migrations.md`, approved 2026-10-30, in progress). An administrator moves a domain in from inbuxa Admin; the server keeps the answers as `inbuxa:Migration` and runs the bundled inbuxa-migrate on them as a child of one node, writing in as that administrator with a run token held in memory only (`crates/jmap/src/inbuxa/migration.rs`, `crates/features/src/migration/`, `crates/services/src/inbuxa_migrations.rs`, `crates/common/src/migration_token.rs`), and the `sysMigrationRun` permission.
+
 Not a rebuild: **inbuxa-relay** is INBUXA's own design (`features/relay.md`, settled 2026-10-29, in progress). It's a small edge appliance that takes mail in on port 25 for one downstream server or cluster, and sends that downstream's outgoing mail on. The downstream dials out to it over JMAP (`urn:inbuxa:jmap:relay`), so the protected network exposes no mail port, and it is administered only from the downstream's inbuxa-admin.
 
 ## 5. The web front ends

docs/spec/features/migrations.md

@@ -0,0 +1,101 @@
+# Feature spec: migrations into this server
+
+Status: building, 2026-10-30. Not a rebuild (SPEC.md §4): inbuxa's own
+design, approved by John on 2026-10-30.
+
+## Provenance
+
+| Source | Used for |
+|---|---|
+| inbuxa-migrate's `guide`, `plan --answers` and `migrate --progress json` | The questions, the plans, the engine and its progress |
+| This server's `inbuxa:HoldExport` and `inbuxa:Relay` | How a fork object, a background job and their permissions are built |
+
+Upstream's open code has nothing like it, and nothing here comes from
+Enterprise code.
+
+## What it is
+
+An administrator moves a whole domain into this server from inbuxa Admin,
+without a machine of their own to run inbuxa-migrate on. The console asks
+the guide's questions. The server keeps the answers as an
+`inbuxa:Migration` and runs the bundled inbuxa-migrate on them, as a child
+process of the node the administrator reached.
+
+## Requirements
+
+- **MG-1.** A migration is an `inbuxa:Migration` under `urn:inbuxa:jmap`.
+  It's kept in the fork's subspace under `I`, never in the registry.
+- **MG-2.** Its source is IMAP (a login that names each account, such as
+  `{email}*master`), JMAP (an administrator or an API key), Microsoft 365
+  (an Entra app's tenant and client id) or Google Workspace (a service
+  account's key). Maildir isn't offered, because the folders would have to
+  be on the node.
+- **MG-3.** Its accounts are addresses, lowercased, without duplicates, at
+  most 10,000. It can create the accounts this server doesn't have yet. It
+  moves 1 to 16 accounts at once, 4 by default.
+- **MG-4.** The target is always this server. A run writes the answers into
+  the migration's work directory, `$INBUXA_MIGRATIONS_DIR/<id>`, or
+  `migrations/<id>` in the server's working directory (the image's data
+  volume). `inbuxa-migrate plan` turns them into plans, so there's one
+  engine and one plan builder. The binary is `$INBUXA_MIGRATE_BIN`, or the
+  one the image carries at `/usr/local/bin/inbuxa-migrate`.
+- **MG-5.** An update's `action` runs it:
+  - `preview`: a dry run;
+  - `move`: the move, which resumes if run again;
+  - `finalPass`: the move with `--again`, only after a move;
+  - `stop`;
+  - `passwords` (MG-13);
+  - `close` (MG-14).
+
+  One run at a time per migration. Its answers change only while nothing
+  runs, and its source not at all once accounts were moved.
+- **MG-6.** A run ends:
+  - `done`;
+  - `unfinished`: some account didn't finish (the engine's exit 5);
+  - `failed`, with the end of what the engine said;
+  - `stopped`;
+  - `interrupted` (MG-11).
+
+  Its report goes into a blob of whoever started the migration.
+- **MG-8.** A run's progress comes from `--progress json`: accounts finished
+  of the total, items read and written, and failures. It's saved as each
+  account ends, and at least every 30 seconds with a heartbeat.
+- **MG-9.** A migration keeps the latest 200 accounts that didn't finish,
+  each with its step and why.
+- **MG-10.** Only administrators run migrations: `sysMigrationRun`, in the
+  built-in Administrator role and granted once to existing administrator
+  roles. A tenant administrator is refused. The child writes into this
+  server as the administrator who started the migration:
+  - It uses a run token that this node holds in memory only. The token is
+    taken only from loopback, ends with the run, and is never stored, not
+    even hashed.
+  - It reaches this node through one of its HTTP listeners on loopback.
+    That's a listener bound to loopback or every address, and not expecting
+    the PROXY protocol there. It's given `--stay-on-url`, and
+    `--allow-invalid-certs` for a TLS listener, since its certificate
+    doesn't name 127.0.0.1. `$INBUXA_MIGRATIONS_URL` overrides the choice.
+  - What the token does is audited as `via: migration`. A run token can't
+    manage migrations.
+- **MG-11.** A migration's work directory is on the node its first run was
+  on, and its runs happen there. An action that reaches another node is
+  refused, naming the node; sealing it over to the right node is the next
+  step. At start-up, a node's runs from before are `interrupted`. So is a
+  run whose node has been quiet for five minutes.
+- **MG-12.** The source's secret is given with an action, held in the
+  node's memory between runs, and passed to the child in its environment:
+  `MIGRATE_SOURCE_SECRET`, beside `MIGRATE_INBUXA_TOKEN`. The child's
+  environment holds those two and `PATH` only. After a restart the next run
+  needs the secret again. The secret is never stored, and the audit log
+  redacts it.
+- **MG-13.** New accounts' passwords are in the work directory, readable by
+  the server only. The `passwords` action hands them over once, as a blob
+  of whoever started the migration, and deletes the file.
+- **MG-14.** Closing a migration deletes its work directory, archives
+  included, and forgets its secret. A closed migration stays listed;
+  destroy is refused.
+
+## Not yet
+
+- Actions sealed over to the node a migration runs on (MG-11).
+- Accounts set up from a JMAP source's own directory (`--source-directory`).
+  For now, a created account is a plain user.

resources/privacy/catalog.toml

@@ -20,7 +20,7 @@
 [vocabulary]
 categories = ["identifier", "contact", "network", "content", "metadata", "credential"]
 whose = ["holder", "correspondent", "administrator"]
-where = ["data-store", "blob-store", "search-store", "in-memory-store", "memory", "log-file", "external"]
+where = ["data-store", "blob-store", "search-store", "in-memory-store", "memory", "log-file", "data-directory", "external"]
 scope = ["tenant", "server"]
 retention = ["unbounded", "object-life", "receiver"]
 
@@ -270,6 +270,23 @@ reason = ["content"]
 createdBy = ["identifier"]
 blobId = ["content", "identifier", "contact", "metadata"]
 
+[object."inbuxa:Migration"]
+file = "inbuxa_migration.rs"
+default = "none"
+whose = ["holder", "administrator"]
+where = ["data-store", "blob-store", "memory"]
+scope = "server"
+retention = "object-life"
+[object."inbuxa:Migration".properties]
+description = ["content"]
+source = ["network", "identifier"]
+accounts = ["identifier"]
+failures = ["identifier", "content"]
+createdBy = ["identifier"]
+reportBlobId = ["identifier", "metadata"]
+passwordsBlobId = ["credential"]
+secret = ["credential"]
+
 [object."inbuxa:DeletedAccount"]
 file = "inbuxa_deleted_account.rs"
 default = "none"
@@ -383,6 +400,18 @@ retention = "receiver"
 leaves_host = false
 written_by = ["crates/services/src/inbuxa_scheduled_reports.rs"]
 
+# A migration's work directory on the node that runs it: an archive per
+# account, which is a full copy of the mailbox, logs and the report. Gone
+# when the migration is closed (migrations spec, MG-14).
+[source."migration-work-directory"]
+categories = ["content", "identifier", "contact", "metadata", "credential"]
+whose = ["holder", "correspondent", "administrator"]
+where = ["data-directory"]
+scope = "server"
+retention = "object-life"
+leaves_host = false
+written_by = ["crates/services/src/inbuxa_migrations.rs"]
+
 [source."full-text-index"]
 categories = ["content", "identifier", "contact", "metadata"]
 whose = ["holder", "correspondent"]

resources/schema/schema.json.sha256

@@ -1 +1 @@
-bxhi7hGulmWKjZiUh2eaxDH78ZzRFaijGqkp16KrLXc
+5cEvhf04ZcIdmZyse9xrcwg1YfsVnLZUo06dAhnFkio

tests/resources/migrations/dovecot-source/Dockerfile

@@ -0,0 +1,18 @@
+# SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+# SPDX-License-Identifier: AGPL-3.0-only
+#
+# The IMAP source for migration_real_engine_tests: two accounts and a
+# master login. `python3 seed.py PORT` puts their mail in.
+
+FROM debian:bookworm-20260518-slim
+RUN apt-get update && \
+    DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends dovecot-imapd && \
+    rm -rf /var/lib/apt/lists/* && \
+    addgroup --gid 5000 vmail && \
+    adduser --system --no-create-home --uid 5000 --gid 5000 vmail && \
+    mkdir -p /var/vmail && chown -R vmail:vmail /var/vmail
+COPY dovecot.conf /etc/dovecot/dovecot.conf
+COPY users /etc/dovecot/users
+COPY masters /etc/dovecot/masters
+EXPOSE 143
+CMD ["dovecot", "-F"]

tests/resources/migrations/dovecot-source/dovecot.conf

@@ -0,0 +1,31 @@
+# SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+# SPDX-License-Identifier: AGPL-3.0-only
+
+protocols = imap
+listen = *
+mail_location = maildir:/var/vmail/%u/Maildir
+mail_uid = 5000
+mail_gid = 5000
+namespace inbox {
+  inbox = yes
+  separator = /
+  prefix =
+}
+auth_mechanisms = plain login
+disable_plaintext_auth = no
+auth_master_user_separator = *
+passdb {
+  driver = passwd-file
+  args = /etc/dovecot/masters
+  master = yes
+}
+passdb {
+  driver = passwd-file
+  args = /etc/dovecot/users
+}
+userdb {
+  driver = static
+  args = uid=5000 gid=5000 home=/var/vmail/%u
+}
+ssl = no
+log_path = /dev/stderr

tests/resources/migrations/dovecot-source/masters

@@ -0,0 +1 @@
+master:{plain}m4ster-e2e

tests/resources/migrations/dovecot-source/seed.py

@@ -0,0 +1,23 @@
+# SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+# SPDX-License-Identifier: AGPL-3.0-only
+
+import imaplib, sys, time
+port = int(sys.argv[1])
+def msg(n, subject):
+    return (f"From: [email protected]\r\nTo: [email protected]\r\nSubject: {subject}\r\n"
+            f"Message-ID: <e2e-{n}@example.net>\r\nDate: Mon, 4 Mar 2024 09:00:00 +0000\r\n\r\nHi {n}\r\n").encode()
+for _ in range(50):
+    try:
+        imaplib.IMAP4("127.0.0.1", port).logout(); break
+    except OSError:
+        time.sleep(0.2)
+a = imaplib.IMAP4("127.0.0.1", port); a.login("[email protected]", "pass-a")
+for n in range(1, 4):
+    a.append("INBOX", r"(\Seen)" if n == 1 else None, None, msg(f"a{n}", f"A message {n}"))
+a.logout()
+b = imaplib.IMAP4("127.0.0.1", port); b.login("[email protected]", "pass-b")
+b.create("Projects")
+b.append("INBOX", None, None, msg("b1", "B inbox"))
+b.append("Projects", r"(\Flagged)", None, msg("b2", "B project"))
+b.logout()
+print("seeded")

tests/resources/migrations/dovecot-source/users

@@ -0,0 +1,2 @@
+[email protected]:{plain}pass-a:5000:5000::/var/vmail/[email protected]::
+[email protected]:{plain}pass-b:5000:5000::/var/vmail/[email protected]::

tests/resources/migrations/fake-inbuxa-migrate

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+# SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+# SPDX-License-Identifier: AGPL-3.0-only
+#
+# A stand-in for inbuxa-migrate in tests/src/system/migration.rs: it answers
+# `plan` and `migrate` the way the real one does, says what it was given in
+# seen-*.json, and signs in to the server with its run token. An account
+# named slow@ makes a run wait, so it can be stopped.
+
+import json
+import os
+import ssl
+import sys
+import time
+import urllib.error
+import urllib.request
+
+args = sys.argv[1:]
+
+if args[0] == "plan":
+    answers = json.load(open(args[2]))
+    json.dump({"args": args, "env": sorted(os.environ)}, open("seen-plan.json", "w"))
+    print(json.dumps({
+        "plan": "plan.json",
+        "preview": "preview-plan.json",
+        "passwords": "passwords-1.csv" if answers["create"] else None,
+        "accounts": len(answers["accounts"]),
+        "secrets": {"source": "MIGRATE_SOURCE_SECRET", "target": "MIGRATE_INBUXA_TOKEN"},
+    }))
+    sys.exit(0)
+
+assert args[0] == "migrate", args
+answers = json.load(open("answers.json"))
+token = os.environ.get("MIGRATE_INBUXA_TOKEN", "")
+url = answers["target"]["url"]
+insecure = ssl.create_default_context()
+insecure.check_hostname = False
+insecure.verify_mode = ssl.CERT_NONE
+signed_in = None
+try:
+    request = urllib.request.Request(
+        url + "/jmap/session", headers={"Authorization": "Bearer " + token}
+    )
+    session = json.load(urllib.request.urlopen(request, context=insecure, timeout=10))
+    signed_in = session.get("username")
+except (urllib.error.URLError, OSError, ValueError) as err:
+    signed_in = "failed: " + str(err)
+
+json.dump({
+    "args": args,
+    "url": url,
+    "secret": os.environ.get("MIGRATE_SOURCE_SECRET"),
+    "token": token,
+    "signedIn": signed_in,
+    "env": sorted(os.environ),
+}, open("seen-run.json", "w"))
+
+
+def emit(event, **fields):
+    fields["event"] = event
+    print(json.dumps(fields), flush=True)
+
+
+accounts = answers["accounts"]
+dry_run = "--dry-run" in args
+emit("start", accounts=len(accounts), parallel=1, dryRun=dry_run)
+if "[email protected]" in accounts:
+    time.sleep(60)
+rows = ["account,step,result,fetched,created,updated,deleted,skipped,failed,detail"]
+unfinished = 0
+for n, email in enumerate(accounts, 1):
+    emit("step", email=email, step="import")
+    counts = {"fetched": 3, "created": 3, "updated": 0, "deleted": 0, "skipped": 0, "failed": 0}
+    emit("step", email=email, step="import", result="done", counts=counts, detail="")
+    rows.append(f"{email},import,done,3,3,0,0,0,0,")
+    result = "done"
+    if not dry_run:
+        partial = email.startswith("b@")
+        counts = dict(counts, created=2 if partial else 3, failed=1 if partial else 0)
+        result = "partial" if partial else "done"
+        detail = "1 message over the size limit" if partial else ""
+        emit("step", email=email, step="export", result=result, counts=counts, detail=detail)
+        rows.append(f"{email},export,{result},3,{counts['created']},0,0,0,{counts['failed']},{detail}")
+    if result != "done":
+        unfinished += 1
+    emit("account", email=email, n=n, total=len(accounts), result=result)
+
+os.makedirs("work", exist_ok=True)
+report = "work/dry-run-report.csv" if dry_run else "work/report.csv"
+open(report, "w").write("\n".join(rows) + "\n")
+if answers["create"] and not dry_run and "--again" not in args:
+    open("passwords-1.csv", "w").write(
+        "email,password\n" + "".join(f"{a},generated-{n}\n" for n, a in enumerate(accounts))
+    )
+emit("end", done=len(accounts) - unfinished, unfinished=unfinished, report=report)
+sys.exit(5 if unfinished else 0)

tests/src/system/migration.rs

@@ -0,0 +1,661 @@
+/*
+ * SPDX-FileCopyrightText: 2026 Coffey Labs LLC
+ *
+ * SPDX-License-Identifier: AGPL-3.0-only
+ */
+
+//! `inbuxa:Migration` (docs/spec/features/migrations.md), with a stand-in
+//! for inbuxa-migrate (`tests/resources/migrations/fake-inbuxa-migrate`)
+//! that answers like the real one and says what it was given.
+//!
+//! A migration is created and checked; a run needs the secret once, and
+//! the child gets it and a run token through its environment and nothing
+//! else of the server's. The token signs in as the administrator, from
+//! loopback, and ends with the run. Preview, move, the passwords once, the
+//! final pass, stop, close; destroy is refused, a tenant is refused, and
+//! the audit log never holds the secret.
+//!
+//! `migration_real_engine_tests` runs the real inbuxa-migrate against a
+//! Dovecot (`tests/resources/migrations/dovecot-source`), on demand.
+
+use crate::utils::{
+    account::Account,
+    server::{TestServer, TestServerBuilder},
+};
+use registry::schema::{
+    prelude::{ObjectType, Property},
+    structs::{CertificateManagement, DkimManagement, DnsManagement, Domain, Tenant, UserRoles},
+};
+use serde_json::{Value, json};
+use std::{
+    path::{Path, PathBuf},
+    str::FromStr,
+    time::Duration,
+};
+
+const USING: &[&str] = &[
+    "urn:ietf:params:jmap:core",
+    "urn:inbuxa:jmap",
+    "urn:inbuxa:jmap:registry",
+];
+const SECRET: &str = "s3cret-source-7731";
+
+async fn call(account: &Account, method: &str, mut arguments: Value) -> (String, Value) {
+    if arguments.get("accountId").is_none() {
+        arguments["accountId"] = account.id_string().into();
+    }
+    let response = account
+        .jmap_request(USING, json!([[method, arguments, "0"]]))
+        .await;
+    let call = response
+        .0
+        .pointer("/methodResponses/0")
+        .cloned()
+        .unwrap_or_else(|| panic!("{method}: {}", response.0));
+    (
+        call[0].as_str().unwrap_or_default().to_string(),
+        call[1].clone(),
+    )
+}
+
+async fn update(account: &Account, id: &str, patch: Value) -> Value {
+    call(
+        account,
+        "inbuxa:Migration/set",
+        json!({"update": {id: patch}}),
+    )
+    .await
+    .1
+}
+
+async fn get(account: &Account, id: &str) -> Value {
+    let (_, response) = call(account, "inbuxa:Migration/get", json!({"ids": [id]})).await;
+    response["list"][0].clone()
+}
+
+/// The migration once its run is over.
+async fn settled(account: &Account, id: &str) -> Value {
+    for _ in 0..300 {
+        let migration = get(account, id).await;
+        if migration["status"] != "running" {
+            return migration;
+        }
+        tokio::time::sleep(Duration::from_millis(100)).await;
+    }
+    panic!("the run didn't end: {}", get(account, id).await);
+}
+
+fn seen(dir: &Path, id: &str, what: &str) -> Value {
+    let text = std::fs::read_to_string(dir.join(id).join(format!("seen-{what}.json")))
+        .unwrap_or_else(|err| panic!("seen-{what}.json: {err}"));
+    serde_json::from_str(&text).unwrap()
+}
+
+/// Migration ids are the store's numbers; the work directory is named by
+/// the number, the JMAP id is its encoding.
+fn number(id: &str) -> String {
+    types::id::Id::from_str(id).unwrap().id().to_string()
+}
+
+pub async fn test(test: &mut TestServer, dir: PathBuf) {
+    println!("Running migration tests...");
+    let admin = test.account("[email protected]");
+    let source = json!({"url": "imaps://old.example.org", "login": "{email}*master"});
+
+    // Created, and checked
+    let (_, response) = call(
+        admin,
+        "inbuxa:Migration/set",
+        json!({"create": {
+            "m": {"description": "From the old Dovecot", "sourceKind": "imap",
+                  "source": source, "accounts": ["[email protected]", "[email protected]"],
+                  "createAccounts": true},
+            "maildir": {"sourceKind": "maildir", "accounts": ["[email protected]"]},
+            "noLogin": {"sourceKind": "imap", "source": {"url": "imaps://old.example.org",
+                        "login": "master"}, "accounts": ["[email protected]"]},
+            "withSecret": {"sourceKind": "imap", "source": source,
+                           "accounts": ["[email protected]"], "secret": SECRET},
+            "serverSet": {"sourceKind": "imap", "source": source,
+                          "accounts": ["[email protected]"], "status": "running"},
+        }}),
+    )
+    .await;
+    let id = response["created"]["m"]["id"]
+        .as_str()
+        .unwrap_or_else(|| panic!("{response}"))
+        .to_string();
+    for (client, property) in [
+        ("maildir", "sourceKind"),
+        ("noLogin", "source"),
+        ("withSecret", "action"),
+        ("serverSet", "status"),
+    ] {
+        assert_eq!(
+            response["notCreated"][client]["properties"][0], property,
+            "{client}: {response}"
+        );
+    }
+    let m = get(admin, &id).await;
+    assert_eq!(m["accounts"], json!(["[email protected]", "[email protected]"]));
+    assert_eq!(m["status"], "idle");
+    assert_eq!(m["createdBy"], "[email protected]");
+    assert_eq!(m["parallel"], 4);
+
+    // A run needs the secret, once
+    let response = update(admin, &id, json!({"action": "preview"})).await;
+    assert_eq!(
+        response["notUpdated"][&id]["properties"][0], "secret",
+        "{response}"
+    );
+    let response = update(admin, &id, json!({"action": "finalPass", "secret": SECRET})).await;
+    assert!(
+        response["notUpdated"][&id]["description"]
+            .as_str()
+            .is_some_and(|d| d.contains("Move the accounts first")),
+        "{response}"
+    );
+
+    // Preview: a dry run, on this node's loopback, as the administrator
+    let response = update(admin, &id, json!({"action": "preview"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    let m = settled(admin, &id).await;
+    assert_eq!(m["run"], "preview", "{m}");
+    assert_eq!(m["result"], "done", "{m}");
+    assert_eq!(m["progress"]["total"], 2, "{m}");
+    assert_eq!(m["progress"]["read"], 6, "{m}");
+    assert_eq!(m["moved"], false);
+    assert!(m["reportBlobId"].is_string(), "{m}");
+    assert!(m["nodeId"].is_string(), "{m}");
+    let work = number(&id);
+    let plan = seen(&dir, &work, "plan");
+    assert_eq!(
+        plan["args"],
+        json!(["plan", "--answers", "answers.json", "."])
+    );
+    let run = seen(&dir, &work, "run");
+    assert!(
+        run["args"]
+            .as_array()
+            .unwrap()
+            .contains(&json!("--dry-run")),
+        "{run}"
+    );
+    assert_eq!(run["secret"], SECRET);
+    assert_eq!(run["signedIn"], "[email protected]", "{run}");
+    assert!(
+        run["url"]
+            .as_str()
+            .unwrap()
+            .starts_with("https://127.0.0.1:"),
+        "{run}"
+    );
+    // Python sets LC_CTYPE itself when the locale is C
+    let env: Vec<&Value> = run["env"]
+        .as_array()
+        .unwrap()
+        .iter()
+        .filter(|var| *var != "LC_CTYPE")
+        .collect();
+    assert_eq!(
+        env,
+        ["MIGRATE_INBUXA_TOKEN", "MIGRATE_SOURCE_SECRET", "PATH"],
+        "only its secrets reach the child"
+    );
+    let answers: Value = serde_json::from_str(
+        &std::fs::read_to_string(dir.join(&work).join("answers.json")).unwrap(),
+    )
+    .unwrap();
+    assert_eq!(answers["target"]["stayOnUrl"], true);
+    assert_eq!(answers["target"]["allowInvalidCerts"], true);
+    assert!(!answers.to_string().contains(SECRET), "{answers}");
+
+    // The run's token ended with it
+    let token = run["token"].as_str().unwrap();
+    assert!(token.starts_with("MIG_"), "{token}");
+    let client = reqwest::Client::builder()
+        .danger_accept_invalid_certs(true)
+        .build()
+        .unwrap();
+    let status = client
+        .get(format!("{}/jmap/session", run["url"].as_str().unwrap()))
+        .bearer_auth(token)
+        .send()
+        .await
+        .unwrap()
+        .status();
+    assert_eq!(status.as_u16(), 401, "a run token outlived its run");
+
+    // Move: the secret is held; one account doesn't finish
+    let response = update(admin, &id, json!({"action": "move"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    let m = settled(admin, &id).await;
+    assert_eq!(m["run"], "move", "{m}");
+    assert_eq!(m["result"], "unfinished", "{m}");
+    assert_eq!(m["progress"]["written"], 5, "{m}");
+    assert_eq!(m["progress"]["unfinished"], 1, "{m}");
+    assert_eq!(m["failures"][0]["email"], "[email protected]", "{m}");
+    assert_eq!(m["failures"][0]["step"], "export", "{m}");
+    assert_eq!(m["moved"], true);
+    assert_eq!(m["passwordsWaiting"], true, "{m}");
+    assert!(m["diskUsed"].as_u64().unwrap() > 0, "{m}");
+    let run = seen(&dir, &work, "run");
+    assert!(
+        !run["args"]
+            .as_array()
+            .unwrap()
+            .contains(&json!("--dry-run"))
+    );
+
+    // The answers can't change source once moved
+    let response = update(
+        admin,
+        &id,
+        json!({"source": {"url": "imaps://elsewhere.example.org",
+                                                       "login": "{email}"}}),
+    )
+    .await;
+    assert_eq!(
+        response["notUpdated"][&id]["properties"][0], "source",
+        "{response}"
+    );
+    let response = update(admin, &id, json!({"description": "Dovecot, done"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+
+    // The new passwords, once
+    let response = update(admin, &id, json!({"action": "passwords"})).await;
+    let blob = response["updated"][&id]["passwordsBlobId"]
+        .as_str()
+        .unwrap_or_else(|| panic!("{response}"))
+        .to_string();
+    let csv = client
+        .get(format!(
+            "{}/jmap/download/{}/{blob}/passwords.csv",
+            run["url"].as_str().unwrap(),
+            admin.id_string()
+        ))
+        .basic_auth(admin.name(), Some(admin.secret()))
+        .send()
+        .await
+        .unwrap()
+        .bytes()
+        .await
+        .unwrap();
+    assert!(
+        String::from_utf8_lossy(&csv).contains("[email protected],generated-0"),
+        "{}",
+        String::from_utf8_lossy(&csv)
+    );
+    assert!(!dir.join(&work).join("passwords-1.csv").exists());
+    let response = update(admin, &id, json!({"action": "passwords"})).await;
+    assert!(response["notUpdated"].get(&id).is_some(), "{response}");
+    assert_eq!(get(admin, &id).await["passwordsWaiting"], false);
+
+    // The final pass
+    update(admin, &id, json!({"action": "finalPass"})).await;
+    let m = settled(admin, &id).await;
+    assert_eq!(m["run"], "finalPass", "{m}");
+    let run = seen(&dir, &work, "run");
+    assert!(
+        run["args"].as_array().unwrap().contains(&json!("--again")),
+        "{run}"
+    );
+
+    // Stopped while it runs
+    let (_, response) = call(
+        admin,
+        "inbuxa:Migration/set",
+        json!({"create": {"slow": {"sourceKind": "jmap",
+            "source": {"url": "https://old.example.org", "login": ""},
+            "accounts": ["[email protected]"]}}}),
+    )
+    .await;
+    let slow = response["created"]["slow"]["id"]
+        .as_str()
+        .unwrap()
+        .to_string();
+    update(admin, &slow, json!({"action": "move", "secret": SECRET})).await;
+    assert_eq!(get(admin, &slow).await["status"], "running");
+    let response = update(admin, &slow, json!({"action": "close"})).await;
+    assert!(response["notUpdated"].get(&slow).is_some(), "{response}");
+    let response = update(admin, &slow, json!({"action": "stop"})).await;
+    assert!(response["updated"].get(&slow).is_some(), "{response}");
+    let m = settled(admin, &slow).await;
+    assert_eq!(m["result"], "stopped", "{m}");
+
+    // Closed: the work directory goes, the migration stays listed
+    let response = update(admin, &id, json!({"action": "close"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    assert!(!dir.join(&work).exists(), "the archives outlived the close");
+    let m = get(admin, &id).await;
+    assert_eq!(m["status"], "closed");
+    let response = update(admin, &id, json!({"action": "move", "secret": SECRET})).await;
+    assert!(
+        response["notUpdated"][&id]["description"]
+            .as_str()
+            .is_some_and(|d| d.contains("closed")),
+        "{response}"
+    );
+    let (_, response) = call(admin, "inbuxa:Migration/set", json!({"destroy": [&id]})).await;
+    assert!(response["notDestroyed"].get(&id).is_some(), "{response}");
+    update(admin, &slow, json!({"action": "close"})).await;
+
+    // Nobody in a tenant reaches migrations
+    let tenant = admin
+        .registry_create_object(Tenant {
+            name: "Migration tenant".to_string(),
+            ..Default::default()
+        })
+        .await;
+    admin
+        .registry_create_object(Domain {
+            name: "migrations.example.org".to_string(),
+            is_enabled: true,
+            member_tenant_id: Some(tenant),
+            certificate_management: CertificateManagement::Manual,
+            dns_management: DnsManagement::Manual,
+            dkim_management: DkimManagement::Manual,
+            ..Default::default()
+        })
+        .await;
+    let t_admin = admin
+        .create_user_account(
+            "[email protected]",
+            "tenant-admin-secret-6610",
+            "Tenant admin",
+            &[],
+            vec![],
+        )
+        .await;
+    admin
+        .registry_update_object(
+            ObjectType::Account,
+            t_admin.id(),
+            json!({Property::Roles: UserRoles::Admin}),
+        )
+        .await;
+    for (method, arguments) in [
+        ("inbuxa:Migration/get", json!({"ids": null})),
+        (
+            "inbuxa:Migration/set",
+            json!({"create": {"x": {"sourceKind": "imap", "source": source,
+                                    "accounts": ["[email protected]"]}}}),
+        ),
+    ] {
+        let (name, response) = call(&t_admin, method, arguments).await;
+        assert_eq!(name, "error", "{method} by a tenant: {response}");
+    }
+
+    // Every change audited, never the secret
+    let (_, response) = call(
+        admin,
+        "inbuxa:AuditEvent/query",
+        json!({"filter": {"targetKind": "inbuxa:Migration"}}),
+    )
+    .await;
+    let ids = response["ids"].clone();
+    assert!(
+        ids.as_array().is_some_and(|ids| ids.len() >= 8),
+        "creates and actions audited: {response}"
+    );
+    let (_, response) = call(admin, "inbuxa:AuditEvent/get", json!({"ids": ids})).await;
+    assert!(
+        !response.to_string().contains(SECRET),
+        "the secret was audited"
+    );
+}
+
+#[ignore]
+#[tokio::test(flavor = "multi_thread")]
+pub async fn migration_tests() {
+    let mut test = TestServerBuilder::new("migration_tests")
+        .await
+        .with_default_listeners()
+        .await
+        .build()
+        .await;
+    let dir = test.temp_dir.path.join("migrations");
+    let engine =
+        PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("resources/migrations/fake-inbuxa-migrate");
+    // SAFETY: set before any run reads them, and only this test reads them
+    unsafe {
+        std::env::set_var("INBUXA_MIGRATIONS_DIR", &dir);
+        std::env::set_var("INBUXA_MIGRATE_BIN", &engine);
+    }
+    let admin = test.create_admin_account("[email protected]").await;
+    test.insert_account(admin);
+    self::test(&mut test, dir).await;
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}
+
+/// The real inbuxa-migrate against a real IMAP source, on demand: a
+/// Dovecot with a master login (`{email}*master`) holding [email protected]
+/// (three messages in INBOX) and [email protected] (one in INBOX, one in
+/// Projects), and the engine's binary.
+///
+/// ```text
+/// cd tests/resources/migrations/dovecot-source
+/// docker build -t inbuxa-e2e-dovecot . && docker run -d --rm \
+///   --name inbuxa-e2e-dovecot -p 127.0.0.1:11143:143 inbuxa-e2e-dovecot
+/// python3 seed.py 11143
+/// INBUXA_MIGRATE_E2E_BIN=…/inbuxa-migrate \
+/// INBUXA_MIGRATE_E2E_IMAP=imap://127.0.0.1:11143 \
+/// INBUXA_MIGRATE_E2E_SECRET=m4ster-e2e \
+///   cargo test -p tests system::migration::migration_real_engine_tests -- --ignored
+/// ```
+pub async fn real_engine(test: &mut TestServer, imap: &str, secret: &str) {
+    println!("Running migrations with the real engine...");
+    let admin = test.account("[email protected]");
+    let (_, response) = call(
+        admin,
+        "inbuxa:Migration/set",
+        json!({"create": {"m": {
+            "description": "From Dovecot", "sourceKind": "imap",
+            "source": {"url": imap, "login": "{email}*master", "cleartext": true},
+            "accounts": ["[email protected]", "[email protected]"], "createAccounts": true,
+        }}}),
+    )
+    .await;
+    let id = response["created"]["m"]["id"]
+        .as_str()
+        .unwrap_or_else(|| panic!("{response}"))
+        .to_string();
+
+    // Preview, then the move, then the final pass
+    let response = update(admin, &id, json!({"action": "preview", "secret": secret})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    let m = settled(admin, &id).await;
+    assert_eq!(m["result"], "done", "{m}");
+    assert_eq!(m["moved"], false, "{m}");
+
+    let response = update(admin, &id, json!({"action": "move"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    let m = settled(admin, &id).await;
+    assert_eq!(m["result"], "done", "{m}");
+    assert_eq!(m["progress"]["finished"], 2, "{m}");
+    assert_eq!(m["moved"], true, "{m}");
+
+    // The new accounts' passwords sign them in, and their mail is there
+    let response = update(admin, &id, json!({"action": "passwords"})).await;
+    let blob = response["updated"][&id]["passwordsBlobId"]
+        .as_str()
+        .unwrap_or_else(|| panic!("{response}"))
+        .to_string();
+    let client = reqwest::Client::builder()
+        .danger_accept_invalid_certs(true)
+        .build()
+        .unwrap();
+    let base = format!("https://127.0.0.1:{}", admin.http_listener_port);
+    let csv = client
+        .get(format!(
+            "{base}/jmap/download/{}/{blob}/passwords.csv",
+            admin.id_string()
+        ))
+        .basic_auth(admin.name(), Some(admin.secret()))
+        .send()
+        .await
+        .unwrap()
+        .text()
+        .await
+        .unwrap();
+    let password = |email: &str| {
+        csv.lines()
+            .find_map(|line| line.strip_prefix(&format!("{email},")))
+            .unwrap_or_else(|| panic!("no password for {email}: {csv}"))
+            .to_string()
+    };
+    let mail = async |email: &str| -> Vec<(String, String)> {
+        let password = password(email);
+        let session: Value = client
+            .get(format!("{base}/jmap/session"))
+            .basic_auth(email, Some(&password))
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        let account = session["primaryAccounts"]["urn:ietf:params:jmap:mail"]
+            .as_str()
+            .unwrap_or_else(|| panic!("{session}"));
+        let response: Value = client
+            .post(format!("{base}/jmap"))
+            .basic_auth(email, Some(&password))
+            .json(&json!({
+                "using": ["urn:ietf:params:jmap:core", "urn:ietf:params:jmap:mail"],
+                "methodCalls": [
+                    ["Mailbox/get", {"accountId": account, "properties": ["name"]}, "b"],
+                    ["Email/query", {"accountId": account}, "q"],
+                    ["Email/get", {"accountId": account,
+                        "#ids": {"resultOf": "q", "name": "Email/query", "path": "/ids"},
+                        "properties": ["subject", "mailboxIds"]}, "g"],
+                ],
+            }))
+            .send()
+            .await
+            .unwrap()
+            .json()
+            .await
+            .unwrap();
+        let mailboxes = &response["methodResponses"][0][1]["list"];
+        let name_of = |id: &str| {
+            mailboxes
+                .as_array()
+                .unwrap()
+                .iter()
+                .find(|m| m["id"] == id)
+                .map(|m| m["name"].as_str().unwrap().to_string())
+                .unwrap_or_default()
+        };
+        let mut found: Vec<(String, String)> = response["methodResponses"][2][1]["list"]
+            .as_array()
+            .unwrap_or_else(|| panic!("{response}"))
+            .iter()
+            .flat_map(|e| {
+                let subject = e["subject"].as_str().unwrap_or_default().to_string();
+                e["mailboxIds"]
+                    .as_object()
+                    .unwrap()
+                    .keys()
+                    .map(|m| (name_of(m), subject.clone()))
+                    .collect::<Vec<_>>()
+            })
+            .collect();
+        found.sort();
+        found
+    };
+    let inbox = |subjects: &[&str]| -> Vec<(String, String)> {
+        subjects
+            .iter()
+            .map(|s| ("Inbox".to_string(), s.to_string()))
+            .collect()
+    };
+    assert_eq!(
+        mail("[email protected]").await,
+        inbox(&["A message 1", "A message 2", "A message 3"])
+    );
+    let mut b = vec![
+        ("Inbox".to_string(), "B inbox".to_string()),
+        ("Projects".to_string(), "B project".to_string()),
+    ];
+    b.sort();
+    assert_eq!(mail("[email protected]").await, b);
+
+    // Mail that arrives at the source before the switch comes over with
+    // the final pass, and nothing comes over twice
+    append(imap, "[email protected]*master", secret, "A message 4").await;
+    let response = update(admin, &id, json!({"action": "finalPass"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+    let m = settled(admin, &id).await;
+    assert_eq!(m["run"], "finalPass", "{m}");
+    assert_eq!(m["result"], "done", "{m}");
+    assert_eq!(
+        mail("[email protected]").await,
+        inbox(&["A message 1", "A message 2", "A message 3", "A message 4"])
+    );
+
+    let response = update(admin, &id, json!({"action": "close"})).await;
+    assert!(response["updated"].get(&id).is_some(), "{response}");
+}
+
+/// One message into a source account's INBOX, over plain IMAP.
+async fn append(url: &str, login: &str, password: &str, subject: &str) {
+    use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
+    let address = url.trim_start_matches("imap://");
+    let stream = tokio::net::TcpStream::connect(address).await.unwrap();
+    let (read, mut write) = stream.into_split();
+    let mut lines = BufReader::new(read).lines();
+    let message = format!(
+        "From: [email protected]\r\nTo: [email protected]\r\nSubject: {subject}\r\n\
+         Message-ID: <e2e-late@example.net>\r\nDate: Tue, 5 Mar 2024 09:00:00 +0000\r\n\r\nLate\r\n"
+    );
+    let commands = [
+        format!("1 LOGIN \"{login}\" \"{password}\"\r\n"),
+        format!("2 APPEND INBOX {{{}+}}\r\n{message}\r\n", message.len()),
+        "3 LOGOUT\r\n".to_string(),
+    ];
+    lines.next_line().await.unwrap();
+    for (n, command) in commands.iter().enumerate() {
+        write.write_all(command.as_bytes()).await.unwrap();
+        let tag = format!("{} ", n + 1);
+        loop {
+            let line = lines.next_line().await.unwrap().unwrap_or_default();
+            if let Some(status) = line.strip_prefix(&tag) {
+                assert!(status.starts_with("OK"), "{command}: {line}");
+                break;
+            }
+        }
+    }
+}
+
+#[tokio::test(flavor = "multi_thread")]
+#[ignore]
+pub async fn migration_real_engine_tests() {
+    let var = |name: &str| {
+        std::env::var(name).unwrap_or_else(|_| panic!("{name} isn't set; see real_engine()"))
+    };
+    let (engine, imap, secret) = (
+        var("INBUXA_MIGRATE_E2E_BIN"),
+        var("INBUXA_MIGRATE_E2E_IMAP"),
+        var("INBUXA_MIGRATE_E2E_SECRET"),
+    );
+    let mut test = TestServerBuilder::new("migration_real_engine_tests")
+        .await
+        .with_default_listeners()
+        .await
+        .build()
+        .await;
+    let dir = test.temp_dir.path.join("migrations");
+    // SAFETY: set before any run reads them, and only this test reads them
+    unsafe {
+        std::env::set_var("INBUXA_MIGRATIONS_DIR", &dir);
+        std::env::set_var("INBUXA_MIGRATE_BIN", &engine);
+    }
+    let admin = test.create_admin_account("[email protected]").await;
+    test.insert_account(admin);
+    real_engine(&mut test, &imap, &secret).await;
+    if test.is_reset() {
+        test.temp_dir.delete();
+    }
+}

tests/src/system/mod.rs

@@ -27,6 +27,7 @@ pub mod legal_hold; // inbuxa: legal hold
 pub mod mail_rules; // inbuxa: DLP and mail flow rules
 pub mod masked_email;
 pub mod masked_secrets; // inbuxa: "****" is never saved as a secret
+pub mod migration; // inbuxa: migrations into this server
 pub mod monitoring;
 pub mod oidc;
 pub mod purge;

tools/fork/tests/test_privacy_check.py

@@ -18,7 +18,7 @@ VOCAB = """
 [vocabulary]
 categories = ["identifier", "contact", "network", "content", "metadata", "credential"]
 whose = ["holder", "correspondent", "administrator"]
-where = ["data-store", "blob-store", "search-store", "in-memory-store", "memory", "log-file", "external"]
+where = ["data-store", "blob-store", "search-store", "in-memory-store", "memory", "log-file", "data-directory", "external"]
 scope = ["tenant", "server"]
 retention = ["unbounded", "object-life", "receiver"]
 """