feat(import): identity import API + FIFA17 real-profile dry-run importer

openfut-identity:
- insert_existing_mapping(game,kind,core_id,external_id): preserve an existing
  external wire id instead of minting; idempotent for an identical mapping,
  rejects conflicting forward/reverse with IdError::Conflict, persists atomically.
- persisted per-scope allocator watermark (set_watermark/watermark_for) so a
  future mint continues past the source high-water even across burned-id gaps;
  next id = max(base_floor, live_max+1, watermark). Backward-compatible on-disk
  format (legacy bare [Row] still loads). +4 tests (10 total).

openfut-import-fifa17 (new): read-only dry-run analysis of a real FIFA17 Python
profile for a faithful Core import. Enforces disjoint item-class balance;
proposes profile-derived CardDefinitions keyed fifa17_<resourceId> (base vs
versioned never collapse) with a resourceId-group consistency gate (hard-fail on
disagreement, never pick a winner) and honest buildability (roster name +
version formula + metadata, never fabricated); plans owned-instance identity
(preserve Python wire ids, preserve nextItemId watermark); checks active-squad
coverage. --apply/--emit-content refuse to write in this phase. 11 tests.

Real profile (33068179/CAGE) dry-run: 1982 items balance (1962 players + 17
consumables + 3 staff); 1681 supported defs (155 base + 1535 versioned), 9
NoName unsupported, 1 hard conflict (resourceId 169193: one of 4 copies has a
divergent nation/team/league); 1949 importable player instances, watermark
100004617 -> first new alloc 100004617; active squad f433 fully supported.
fmt + clippy -D warnings clean.

openfut-identity/src/lib.rs

@@ -37,6 +37,8 @@ use serde::{Deserialize, Serialize};
 pub enum IdError {
     Io(std::io::Error),
     Corrupt(String),
+    /// A caller-supplied import mapping conflicts with an existing one.
+    Conflict(String),
 }
 
 impl std::fmt::Display for IdError {
@@ -44,6 +46,7 @@ impl std::fmt::Display for IdError {
         match self {
             IdError::Io(e) => write!(f, "identity store io: {e}"),
             IdError::Corrupt(e) => write!(f, "identity store corrupt: {e}"),
+            IdError::Conflict(e) => write!(f, "identity store conflict: {e}"),
         }
     }
 }
@@ -80,12 +83,36 @@ struct Row {
     external_id: i64,
 }
 
+/// A persisted per-scope allocator watermark: the next external id to issue,
+/// imported from a source system so historical allocation (including ids burned
+/// in gaps) is preserved, not just currently-live mappings.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+struct Watermark {
+    game_id: String,
+    entity_kind: String,
+    next_id: i64,
+}
+
+/// On-disk store file. Backward-compatible: a legacy bare `[Row, ...]` array is
+/// still accepted on load; new writes always use this object form so watermarks
+/// persist alongside the rows.
+#[derive(Debug, Default, Serialize, Deserialize)]
+struct StoreFile {
+    #[serde(default)]
+    rows: Vec<Row>,
+    #[serde(default)]
+    watermarks: Vec<Watermark>,
+}
+
 #[derive(Default)]
 struct Index {
     rows: Vec<Row>,
     fwd: HashMap<(String, String, String), i64>,
     rev: HashMap<(String, String, i64), String>,
     max: HashMap<(String, String), i64>,
+    /// Per-scope imported allocation watermark (next id to issue): a floor that
+    /// preserves source allocation history beyond the live `max`.
+    watermarks: HashMap<(String, String), i64>,
 }
 
 impl Index {
@@ -115,6 +142,21 @@ impl Index {
     }
 }
 
+/// The next external id to issue in a scope: the maximum of the namespace floor,
+/// one past the highest live mapping, and any imported watermark. Taking the max
+/// of all three preserves source allocation history (burned-id gaps) while never
+/// reissuing a live id nor dropping below the adapter's policy floor.
+fn next_external(ix: &Index, game: &str, kind: &str, base_floor: i64) -> i64 {
+    let scope = (game.to_string(), kind.to_string());
+    let from_max = ix.max.get(&scope).map(|m| m + 1);
+    let from_wm = ix.watermarks.get(&scope).copied();
+    [Some(base_floor), from_max, from_wm]
+        .into_iter()
+        .flatten()
+        .max()
+        .unwrap_or(base_floor)
+}
+
 /// JSON-file backed [`ExternalIdentityStore`]. Single-writer (one host process).
 pub struct JsonIdentityStore {
     path: PathBuf,
@@ -129,9 +171,20 @@ impl JsonIdentityStore {
         if path.exists() {
             let raw = std::fs::read_to_string(&path).map_err(IdError::Io)?;
             if !raw.trim().is_empty() {
-                let rows: Vec<Row> =
-                    serde_json::from_str(&raw).map_err(|e| IdError::Corrupt(e.to_string()))?;
-                for row in rows {
+                // Accept the modern object form; fall back to the legacy bare
+                // `[Row, ...]` array so pre-watermark stores keep loading.
+                let file: StoreFile = match serde_json::from_str::<StoreFile>(&raw) {
+                    Ok(f) => f,
+                    Err(_) => {
+                        let rows: Vec<Row> = serde_json::from_str(&raw)
+                            .map_err(|e| IdError::Corrupt(e.to_string()))?;
+                        StoreFile {
+                            rows,
+                            watermarks: Vec::new(),
+                        }
+                    }
+                };
+                for row in file.rows {
                     // Reject a torn file that violates reverse-uniqueness rather
                     // than silently serving an ambiguous reverse lookup.
                     let rkey = (
@@ -147,6 +200,11 @@ impl JsonIdentityStore {
                     }
                     index.insert(row);
                 }
+                for wm in file.watermarks {
+                    index
+                        .watermarks
+                        .insert((wm.game_id, wm.entity_kind), wm.next_id);
+                }
             }
         }
         Ok(JsonIdentityStore {
@@ -157,13 +215,123 @@ impl JsonIdentityStore {
 
     /// Atomically persist the full row set (temp + rename).
     fn persist(&self, index: &Index) -> Result<(), IdError> {
-        let json =
-            serde_json::to_vec_pretty(&index.rows).map_err(|e| IdError::Corrupt(e.to_string()))?;
+        let file = StoreFile {
+            rows: index.rows.clone(),
+            watermarks: index
+                .watermarks
+                .iter()
+                .map(|((g, k), &n)| Watermark {
+                    game_id: g.clone(),
+                    entity_kind: k.clone(),
+                    next_id: n,
+                })
+                .collect(),
+        };
+        let json = serde_json::to_vec_pretty(&file).map_err(|e| IdError::Corrupt(e.to_string()))?;
         let tmp = self.path.with_extension("tmp");
         std::fs::write(&tmp, &json).map_err(IdError::Io)?;
         std::fs::rename(&tmp, &self.path).map_err(IdError::Io)?;
         Ok(())
     }
+
+    /// Register an EXISTING external id for `(game, kind, core_id)` instead of
+    /// allocating one — the import path that preserves a source system's wire
+    /// ids. Idempotent for an identical mapping; rejects a conflicting forward
+    /// (core already mapped elsewhere) or reverse (external already owned)
+    /// mapping with [`IdError::Conflict`]; persists both directions atomically.
+    pub fn insert_existing_mapping(
+        &self,
+        game: &str,
+        kind: &str,
+        core_id: &str,
+        external_id: i64,
+    ) -> Result<(), IdError> {
+        let mut ix = self.inner.lock();
+        let fkey = (game.to_string(), kind.to_string(), core_id.to_string());
+        let rkey = (game.to_string(), kind.to_string(), external_id);
+        let fwd_existing = ix.fwd.get(&fkey).copied();
+        let rev_existing = ix.rev.get(&rkey).cloned();
+        if let Some(e) = fwd_existing {
+            if e == external_id && rev_existing.as_deref() == Some(core_id) {
+                return Ok(()); // identical mapping — idempotent
+            }
+            return Err(IdError::Conflict(format!(
+                "core '{core_id}' already maps to {e} in ({game}, {kind}), not {external_id}"
+            )));
+        }
+        if let Some(c) = rev_existing {
+            return Err(IdError::Conflict(format!(
+                "external {external_id} already owned by core '{c}' in ({game}, {kind}), not '{core_id}'"
+            )));
+        }
+        ix.insert(Row {
+            game_id: game.to_string(),
+            entity_kind: kind.to_string(),
+            core_id: core_id.to_string(),
+            external_id,
+        });
+        if let Err(e) = self.persist(&ix) {
+            // roll the in-memory insert back so memory and disk never diverge
+            ix.rows.pop();
+            ix.fwd.remove(&fkey);
+            ix.rev.remove(&rkey);
+            let scope = (game.to_string(), kind.to_string());
+            let new_max = ix
+                .rows
+                .iter()
+                .filter(|r| r.game_id == game && r.entity_kind == kind)
+                .map(|r| r.external_id)
+                .max();
+            match new_max {
+                Some(m) => {
+                    ix.max.insert(scope, m);
+                }
+                None => {
+                    ix.max.remove(&scope);
+                }
+            }
+            return Err(e);
+        }
+        Ok(())
+    }
+
+    /// Set the per-scope allocator watermark (the next external id to issue),
+    /// imported from the source system so future allocations continue past its
+    /// historical high-water even where those ids left no live mapping (gaps).
+    /// A floor: it never lowers the effective next id below the live max.
+    pub fn set_watermark(&self, game: &str, kind: &str, next_id: i64) -> Result<(), IdError> {
+        let mut ix = self.inner.lock();
+        let scope = (game.to_string(), kind.to_string());
+        let prev = ix.watermarks.insert(scope.clone(), next_id);
+        if let Err(e) = self.persist(&ix) {
+            match prev {
+                Some(p) => {
+                    ix.watermarks.insert(scope, p);
+                }
+                None => {
+                    ix.watermarks.remove(&scope);
+                }
+            }
+            return Err(e);
+        }
+        Ok(())
+    }
+
+    /// The imported watermark for a scope, if any.
+    pub fn watermark_for(&self, game: &str, kind: &str) -> Option<i64> {
+        self.inner
+            .lock()
+            .watermarks
+            .get(&(game.to_string(), kind.to_string()))
+            .copied()
+    }
+
+    /// The external id the next allocation in this scope WOULD receive, without
+    /// allocating. For dry-run reporting.
+    pub fn peek_next_external(&self, game: &str, kind: &str, base_floor: i64) -> i64 {
+        let ix = self.inner.lock();
+        next_external(&ix, game, kind, base_floor)
+    }
 }
 
 impl ExternalIdentityStore for JsonIdentityStore {
@@ -180,10 +348,7 @@ impl ExternalIdentityStore for JsonIdentityStore {
             return Ok(ext);
         }
         let scope = (game.to_string(), kind.to_string());
-        let next = match ix.max.get(&scope) {
-            Some(&m) => (m + 1).max(base_floor),
-            None => base_floor,
-        };
+        let next = next_external(&ix, game, kind, base_floor);
         let row = Row {
             game_id: game.to_string(),
             entity_kind: kind.to_string(),
@@ -334,4 +499,83 @@ mod tests {
             "ambiguous reverse must not load"
         );
     }
+
+    #[test]
+    fn insert_existing_mapping_preserves_and_is_idempotent() {
+        let (s, _d) = store();
+        s.insert_existing_mapping(G, K, "oc-A", 100_004_600)
+            .unwrap();
+        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(100_004_600));
+        assert_eq!(
+            s.core_for(G, K, 100_004_600).unwrap().as_deref(),
+            Some("oc-A")
+        );
+        // identical re-insert is idempotent
+        s.insert_existing_mapping(G, K, "oc-A", 100_004_600)
+            .unwrap();
+    }
+
+    #[test]
+    fn insert_existing_mapping_rejects_conflicts() {
+        let (s, _d) = store();
+        s.insert_existing_mapping(G, K, "oc-A", 100_000_010)
+            .unwrap();
+        // same core -> different external: reject
+        assert!(matches!(
+            s.insert_existing_mapping(G, K, "oc-A", 100_000_011),
+            Err(IdError::Conflict(_))
+        ));
+        // same external -> different core: reject
+        assert!(matches!(
+            s.insert_existing_mapping(G, K, "oc-B", 100_000_010),
+            Err(IdError::Conflict(_))
+        ));
+    }
+
+    #[test]
+    fn watermark_preserves_allocation_history_past_gaps() {
+        let (s, _d) = store();
+        // a live mapping (highest surviving id) plus a source watermark that sits
+        // ABOVE it because ids in between were issued then removed (burned gap).
+        s.insert_existing_mapping(G, K, "oc-live", 100_004_605)
+            .unwrap();
+        s.set_watermark(G, K, 100_004_617).unwrap();
+        assert_eq!(s.peek_next_external(G, K, BASE), 100_004_617);
+        // first NEW allocation is the source next-to-issue, NOT live_max+1
+        assert_eq!(
+            s.resolve_or_allocate(G, K, "oc-new", BASE).unwrap(),
+            100_004_617
+        );
+        // then continues monotonically
+        assert_eq!(
+            s.resolve_or_allocate(G, K, "oc-new2", BASE).unwrap(),
+            100_004_618
+        );
+    }
+
+    #[test]
+    fn watermark_and_imported_ids_survive_reopen() {
+        let dir = tempfile::tempdir().unwrap();
+        let path = dir.path().join("ids.json");
+        {
+            let s = JsonIdentityStore::open(&path).unwrap();
+            s.insert_existing_mapping(G, K, "oc-A", 100_000_001)
+                .unwrap();
+            s.insert_existing_mapping(G, K, "oc-B", 100_004_605)
+                .unwrap();
+            s.set_watermark(G, K, 100_004_617).unwrap();
+        }
+        let s = JsonIdentityStore::open(&path).unwrap();
+        assert_eq!(s.external_for(G, K, "oc-A").unwrap(), Some(100_000_001));
+        assert_eq!(
+            s.core_for(G, K, 100_004_605).unwrap().as_deref(),
+            Some("oc-B")
+        );
+        assert_eq!(s.watermark_for(G, K), Some(100_004_617));
+        // sparse import: next allocation respects the persisted watermark
+        assert_eq!(
+            s.resolve_or_allocate(G, K, "oc-C", BASE).unwrap(),
+            100_004_617
+        );
+    }
 }