agentmux_srv/migrations/
m0012_dedup_identity_accounts.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Deduplicate identity accounts that share the same (provider, credential).
5//!
6//! Root cause: the old per-channel `backfill_shared_store_once` ran at every
7//! channel startup and copied the channel's local account UUID into the shared
8//! store. Each channel had a different UUID for the same underlying credential
9//! (e.g. the shared claude OAuth dir), so the shared store accumulated one row
10//! per channel launch rather than one row per credential.
11//!
12//! Strategy: group accounts by (provider, secret_ref JSON). For each group of
13//! duplicates, keep whichever account is currently referenced by a direct
14//! agent identity link (or the oldest by created_at if none is linked).
15//! Reroute any agent identity links that reference a dupe to the canonical,
16//! then delete the dupes — the CASCADE FK handles any remaining joins.
17//! (Identity bundle bindings were rerouted here too, prior to Phase 4c of
18//! SPEC_PRESET_TO_BUNDLE_REFACTOR_2026_07_02.md dropping
19//! db_identity_bundles/db_identity_bindings.)
20
21use std::collections::{HashMap, HashSet};
22
23use crate::backend::storage::store::Store;
24use super::{Migration, MigrationContext, MigrationError, MigrationScope};
25
26pub struct M0012DedupIdentityAccounts;
27
28impl Migration for M0012DedupIdentityAccounts {
29    fn id(&self) -> &'static str { "0012_dedup_identity_accounts" }
30    fn scope(&self) -> MigrationScope { MigrationScope::Global }
31    fn description(&self) -> &'static str {
32        "Deduplicate identity accounts sharing the same provider and credential"
33    }
34
35    fn up(&self, ctx: &MigrationContext) -> Result<(), MigrationError> {
36        let shared = Store::open_shared(&ctx.shared_store_path)
37            .map_err(|e| MigrationError(format!("dedup_identity: open shared store: {}", e)))?;
38
39        let accounts = shared.identity_list(None)
40            .map_err(|e| MigrationError(format!("dedup_identity: list accounts: {}", e)))?;
41
42        // Group by (provider, serialized secret_ref). Using JSON serialization
43        // gives a stable string key; serde_json output is deterministic for
44        // structs with fixed field order.
45        let mut groups: HashMap<(String, String), Vec<usize>> = HashMap::new();
46        for (i, acct) in accounts.iter().enumerate() {
47            let ref_key = serde_json::to_string(&acct.secret_ref)
48                .unwrap_or_else(|_| acct.id.clone());
49            groups.entry((acct.provider.clone(), ref_key)).or_default().push(i);
50        }
51
52        let has_dupes = groups.values().any(|v| v.len() > 1);
53        if !has_dupes {
54            tracing::info!("dedup_identity_accounts: no duplicates found");
55            return Ok(());
56        }
57
58        // Gather agent links before mutating. (Identity bundle bindings
59        // were gathered and rebound here too, prior to Phase 4c of
60        // SPEC_PRESET_TO_BUNDLE_REFACTOR_2026_07_02.md dropping
61        // db_identity_bundles/db_identity_bindings — there's no longer
62        // anything to rebind on that axis.)
63        let agent_links = shared.agent_identity_list_all()
64            .map_err(|e| MigrationError(format!("dedup_identity: list agent links: {}", e)))?;
65
66        let bound_ids: HashSet<&str> = agent_links.iter().map(|l| l.account_id.as_str()).collect();
67
68        let mut deleted = 0usize;
69
70        for ((provider, _), indices) in &groups {
71            if indices.len() <= 1 {
72                continue;
73            }
74
75            // Sort: bound accounts first, then oldest by created_at.
76            let mut sorted: Vec<&_> = indices.iter().map(|&i| &accounts[i]).collect();
77            sorted.sort_by(|a, b| {
78                let a_bound = bound_ids.contains(a.id.as_str());
79                let b_bound = bound_ids.contains(b.id.as_str());
80                b_bound.cmp(&a_bound).then_with(|| a.created_at.cmp(&b.created_at))
81            });
82
83            let canonical_id = &sorted[0].id;
84            let dupe_ids: HashSet<&str> = sorted[1..].iter().map(|a| a.id.as_str()).collect();
85
86            // Relink any agent identity links that reference a dupe.
87            for link in &agent_links {
88                if &link.provider == provider && dupe_ids.contains(link.account_id.as_str()) {
89                    shared.agent_identity_link(&link.agent_id, canonical_id, provider)
90                        .map_err(|e| MigrationError(format!(
91                            "dedup_identity: relink agent {}: {}", link.agent_id, e
92                        )))?;
93                }
94            }
95
96            // Delete dupes — CASCADE handles any remaining bindings/links.
97            for dupe_id in &dupe_ids {
98                shared.identity_delete(dupe_id)
99                    .map_err(|e| MigrationError(format!("dedup_identity: delete {}: {}", dupe_id, e)))?;
100                deleted += 1;
101            }
102        }
103
104        tracing::info!(deleted, "dedup_identity_accounts: complete");
105        Ok(())
106    }
107}
agentmux_srv/migrations/m0012_dedup_identity_accounts.rs

agentmux_srv/migrations/
m0012_dedup_identity_accounts.rs
