Follow the selected account and safely consolidate confirmed relinks

crates/chat-store/migrations/2026-10-26-000001_account_retirement/down.sql

@@ -0,0 +1,5 @@
+-- Losing an intent or redirect could strand cleanup or select a removed id.
+CREATE TABLE retirement_downgrade_guard (empty INTEGER NOT NULL CHECK(empty = 0));
+INSERT INTO retirement_downgrade_guard SELECT COUNT(*) FROM account_retirement;
+DROP TABLE retirement_downgrade_guard;
+DROP TABLE account_retirement;

crates/chat-store/migrations/2026-10-26-000001_account_retirement/up.sql

@@ -0,0 +1,8 @@
+-- Survives both device-row deletion and interrupted external cleanup.
+CREATE TABLE account_retirement (
+    old_id INTEGER PRIMARY KEY NOT NULL CHECK(old_id > 0),
+    successor INTEGER NOT NULL CHECK(successor > old_id),
+    pn TEXT NOT NULL,
+    lid TEXT NOT NULL,
+    complete INTEGER NOT NULL DEFAULT 0 CHECK(complete IN (0, 1))
+);

crates/chat-store/src/lib.rs

@@ -53,6 +53,7 @@ pub use materialize::{
 };
 pub use queries::MessageReadState;
 pub use store::ChatStore;
+pub use store::retirement::AccountRetirement;
 pub use store::shared_history::PendingHistoryBundle;
 pub use types::{
     ArrivalCursor, AvatarDescriptor, ChatArchiveFilter, ChatCursor, ChatEntry, ChatNameExpected,

crates/chat-store/src/store/mod.rs

@@ -23,6 +23,7 @@ mod reaction;
 pub(crate) mod read_state;
 mod receipt;
 mod recovery;
+pub(crate) mod retirement;
 mod revoke;
 pub(crate) mod shared_history;
 mod writer;
@@ -1198,6 +1199,40 @@ mod migration_tests {
         result
     }
 
+    /// Isolate this migration's downgrade edge from later independent
+    /// migrations, rather than assuming a fixed latest version or count.
+    fn rewind_to_retirement_migration(
+        conn: &mut diesel::SqliteConnection,
+    ) -> std::result::Result<(), StoreError> {
+        use diesel::migration::{Migration, MigrationSource};
+        use diesel::sqlite::Sqlite;
+        let migrations: Vec<Box<dyn Migration<Sqlite>>> =
+            MIGRATIONS.migrations().map_err(StoreError::Migration)?;
+        // Diesel strips the directory date's hyphens in the shared ledger.
+        let target = migrations
+            .iter()
+            .find(|migration| migration.name().version().to_string() == "20261026000001")
+            .expect("the account retirement migration is embedded")
+            .name()
+            .version()
+            .as_owned();
+        loop {
+            let applied = conn.applied_migrations().map_err(StoreError::Migration)?;
+            assert!(
+                applied.contains(&target),
+                "retirement migration must remain applied"
+            );
+            let newer = migrations.iter().any(|migration| {
+                let version = migration.name().version();
+                version > target && applied.contains(&version)
+            });
+            if !newer {
+                return Ok(());
+            }
+            revert_last_chat_migration(conn)?;
+        }
+    }
+
     #[tokio::test]
     async fn stable_id_downgrade_round_trips_then_sender_identity_refuses() {
         let store = SqliteStore::open(&format!(
@@ -1207,7 +1242,19 @@ mod migration_tests {
         .await
         .expect("create store");
         ChatStore::new(&store).await.expect("run migrations");
+        store
+            .shared()
+            .run(rewind_to_retirement_migration)
+            .await
+            .expect("isolate the retirement downgrade edge");
 
+        assert!(has_table(&store, "account_retirement").await);
+        store
+            .shared()
+            .run(revert_last_chat_migration)
+            .await
+            .expect("an empty retirement journal downgrade is reversible");
+        assert!(!has_table(&store, "account_retirement").await);
         assert!(has_table(&store, "session_recovery").await);
         store
             .shared()
@@ -1341,6 +1388,44 @@ mod migration_tests {
         assert_eq!(table_count(&store).await, 1);
     }
 
+    #[tokio::test]
+    async fn retirement_downgrade_preserves_pending_cleanup_and_completed_redirects() {
+        let store = SqliteStore::open(&format!(
+            "file:retirement-downgrade-{}?mode=memory&cache=shared",
+            std::process::id()
+        ))
+        .await
+        .unwrap();
+        ChatStore::prepare(&store).await.unwrap();
+        store
+            .shared()
+            .run(rewind_to_retirement_migration)
+            .await
+            .expect("isolate the retirement downgrade edge");
+        store.shared().run(|conn| {
+            diesel::sql_query("INSERT INTO account_retirement (old_id, successor, pn, lid) VALUES (1, 2, '100001@s.whatsapp.net', '200001@lid')").execute(conn).map(|_| ()).map_err(crate::error::db_err)
+        }).await.unwrap();
+        for complete in [false, true] {
+            if complete {
+                ChatStore::finish_account_retirement(&store, 1)
+                    .await
+                    .unwrap();
+            }
+            store
+                .shared()
+                .run(revert_last_chat_migration)
+                .await
+                .expect_err("journal records cannot be silently discarded by downgrade");
+            let records = ChatStore::account_retirements(&store).await.unwrap();
+            assert_eq!(records.len(), 1);
+            assert_eq!(records[0].complete, complete);
+            assert!(
+                !has_table(&store, "retirement_downgrade_guard").await,
+                "failed migration must roll back its guard table"
+            );
+        }
+    }
+
     async fn table_count(store: &SqliteStore) -> i64 {
         #[derive(QueryableByName)]
         struct Count {

crates/chat-store/src/store/retirement.rs

@@ -0,0 +1,84 @@
+use diesel::prelude::*;
+use diesel::sql_types::{Bool, Integer, Text};
+use whatsapp_rust_sqlite_storage::SqliteStore;
+
+use super::ChatStore;
+use crate::error::{Result, db_err};
+
+/// A durable redirect and retry record, independent of the retired device row.
+#[derive(Clone, Debug, PartialEq, Eq, QueryableByName)]
+pub struct AccountRetirement {
+    #[diesel(sql_type = Integer)]
+    pub old_id: i32,
+    #[diesel(sql_type = Integer)]
+    pub successor: i32,
+    #[diesel(sql_type = Text)]
+    pub pn: String,
+    #[diesel(sql_type = Text)]
+    pub lid: String,
+    #[diesel(sql_type = Bool)]
+    pub complete: bool,
+}
+
+impl ChatStore {
+    pub async fn account_retirements(store: &SqliteStore) -> Result<Vec<AccountRetirement>> {
+        Ok(store.shared().run(|conn| {
+            diesel::sql_query("SELECT old_id, successor, pn, lid, complete FROM account_retirement ORDER BY old_id")
+                .load(conn).map_err(db_err)
+        }).await?)
+    }
+
+    /// Freeze an existing recovery edge and its exact persisted identity evidence.
+    /// The caller must pin the successor lifecycle and verify full device loading.
+    pub async fn begin_account_retirement(
+        store: &SqliteStore,
+        old_id: i32,
+        successor: i32,
+        old_pn: String,
+        old_lid: String,
+        pn: String,
+        lid: String,
+    ) -> Result<bool> {
+        if old_id <= 0 || successor <= old_id {
+            return Ok(false);
+        }
+        Ok(store.shared().run(move |conn| {
+            conn.immediate_transaction::<_, diesel::result::Error, _>(|conn| {
+                let count = diesel::sql_query(
+                    "INSERT INTO account_retirement (old_id, successor, pn, lid) \
+                     SELECT ?, ?, ?, ? WHERE \
+                     EXISTS (SELECT 1 FROM session_recovery WHERE device_id = ? AND successor = ?) AND \
+                     EXISTS (SELECT 1 FROM device WHERE id = ? AND pn = ? AND lid = ?) AND \
+                     EXISTS (SELECT 1 FROM device WHERE id = ? AND pn = ? AND lid = ?) \
+                     ON CONFLICT(old_id) DO NOTHING")
+                    .bind::<Integer, _>(old_id).bind::<Integer, _>(successor)
+                    .bind::<Text, _>(&pn).bind::<Text, _>(&lid)
+                    .bind::<Integer, _>(old_id).bind::<Integer, _>(successor)
+                    .bind::<Integer, _>(old_id).bind::<Text, _>(&old_pn).bind::<Text, _>(&old_lid)
+                    .bind::<Integer, _>(successor).bind::<Text, _>(&pn).bind::<Text, _>(&lid)
+                    .execute(conn)?;
+                if count == 0 { return Ok(false); }
+                // Do this before removing the intermediate row. Its FK otherwise
+                // sets predecessors to NULL and loses the rest of a relink chain.
+                diesel::sql_query("UPDATE session_recovery SET successor = ? WHERE successor = ? AND device_id < ?")
+                    .bind::<Integer, _>(successor).bind::<Integer, _>(old_id).bind::<Integer, _>(old_id)
+                    .execute(conn)?;
+                Ok(true)
+            }).map_err(db_err)
+        }).await?)
+    }
+
+    pub async fn finish_account_retirement(store: &SqliteStore, old_id: i32) -> Result<()> {
+        store
+            .shared()
+            .run(move |conn| {
+                diesel::sql_query("UPDATE account_retirement SET complete = 1 WHERE old_id = ?")
+                    .bind::<Integer, _>(old_id)
+                    .execute(conn)
+                    .map(|_| ())
+                    .map_err(db_err)
+            })
+            .await?;
+        Ok(())
+    }
+}

crates/cli/src/args.rs

@@ -22,6 +22,10 @@ pub struct OxidezapCli {
     #[usage(long, global, env = "OXIDEZAP_ACCOUNT")]
     pub account: Option<String>,
 
+    /// Select a local account inside the daemon, independently of --account
+    #[usage(long, global, env = "OXIDEZAP_LOCAL_ACCOUNT")]
+    pub local_account: Option<String>,
+
     /// Override daemon socket path
     #[usage(long, global, env = "OXIDEZAP_SOCKET")]
     pub socket: Option<String>,

crates/cli/src/main.rs

@@ -191,6 +191,19 @@ fn main() -> ExitCode {
         }
     }
 
+    let local_account = match cli.local_account.as_deref().map(str::parse::<i32>) {
+        Some(Ok(id)) if id > 0 => Some(id),
+        None => None,
+        _ => {
+            print_error(
+                output_mode,
+                "invalid_local_account",
+                "--local-account must be a positive SQLite account id",
+            );
+            return ExitCode::from(2);
+        }
+    };
+
     // Connect, starting the daemon beside this binary when nothing is
     // listening. An explicit socket never spawns: the caller named a daemon it
     // expects to exist.
@@ -232,13 +245,28 @@ fn main() -> ExitCode {
     let handshake_req = ClientRequest::Hello {
         protocol: CURRENT_PROTOCOL_VERSION,
         client_name: "oxidezap-cli".into(),
+        local_account,
+        account_selection: true,
         read_only: cli.read_only,
         session_events,
     };
 
-    if let Err(err) = client.request(handshake_req) {
-        print_error(output_mode, &err.code, &err.message);
-        return ExitCode::from(3);
+    match client.request(handshake_req) {
+        Ok(DaemonResponse::AccountSelected {
+            local_account: selected,
+        }) if local_account.is_none_or(|requested| requested == selected) => {}
+        Ok(_) => {
+            print_error(
+                output_mode,
+                "account_selection_unsupported",
+                "daemon did not confirm local account selection; update the daemon before retrying",
+            );
+            return ExitCode::from(3);
+        }
+        Err(err) => {
+            print_error(output_mode, &err.code, &err.message);
+            return ExitCode::from(3);
+        }
     }
 
     // Execute command

crates/cli/tests/cli_surface.rs

@@ -9,7 +9,8 @@ fn cli() -> Command {
     command
         .env_remove("OXIDEZAP_ACCOUNT")
         .env_remove("OXIDEZAP_SOCKET")
-        .env_remove("OXIDEZAP_READONLY");
+        .env_remove("OXIDEZAP_READONLY")
+        .env_remove("OXIDEZAP_LOCAL_ACCOUNT");
     command
 }
 
@@ -370,10 +371,10 @@ fn export_continues_from_the_daemon_cursor() {
         let mut writer = stream;
         let mut line = String::new();
 
-        // Handshake: any request is answered with an Ack.
+        // The daemon confirms the account before receiving business requests.
         reader.read_line(&mut line).expect("hello");
         writer
-            .write_all(b"{\"id\":1,\"status\":\"ok\",\"type\":\"ack\"}\n")
+            .write_all(b"{\"id\":1,\"status\":\"ok\",\"type\":\"account_selected\",\"data\":{\"local_account\":2}}\n")
             .expect("answer hello");
 
         let mut page = 0;
@@ -452,3 +453,52 @@ fn export_continues_from_the_daemon_cursor() {
         "the second page must continue from the daemon cursor, not a message id"
     );
 }
+/// A daemon predating account selection ignores unknown hello fields. Its Ack
+/// must never authorize a send against the historical account by accident.
+#[cfg(unix)]
+#[test]
+fn an_old_daemon_cannot_silently_ignore_local_account_selection() {
+    use std::io::{BufRead as _, Write as _};
+    let dir = std::env::temp_dir().join(format!("oxi-old-hello-{}", std::process::id()));
+    std::fs::create_dir_all(&dir).unwrap();
+    let socket = dir.join("daemon.sock");
+    let listener = std::os::unix::net::UnixListener::bind(&socket).unwrap();
+    let server = std::thread::spawn(move || {
+        let (stream, _) = listener.accept().unwrap();
+        stream
+            .set_read_timeout(Some(std::time::Duration::from_secs(5)))
+            .unwrap();
+        let mut reader = std::io::BufReader::new(stream);
+        let mut line = String::new();
+        reader.read_line(&mut line).unwrap();
+        let hello: serde_json::Value = serde_json::from_str(&line).unwrap();
+        assert_eq!(hello["params"]["local_account"], 2);
+        assert_eq!(hello["params"]["account_selection"], true);
+        reader
+            .get_mut()
+            .write_all(b"{\"id\":1,\"status\":\"ok\",\"type\":\"ack\"}\n")
+            .unwrap();
+        line.clear();
+        assert_eq!(
+            reader.read_line(&mut line).unwrap(),
+            0,
+            "no business request may follow an unconfirmed selection"
+        );
+    });
+    let output = cli()
+        .args([
+            "--socket",
+            socket.to_str().unwrap(),
+            "--local-account",
+            "2",
+            "--json",
+            "status",
+        ])
+        .output()
+        .unwrap();
+    assert_eq!(output.status.code(), Some(3));
+    let error: serde_json::Value = serde_json::from_slice(&output.stderr).unwrap();
+    assert_eq!(error["error"]["code"], "account_selection_unsupported");
+    server.join().unwrap();
+    std::fs::remove_dir_all(dir).unwrap();
+}