feat(import-fifa17): --apply — recoverable two-store real-profile import

FIFA17-specific orchestration that installs the real profile across BOTH durable
stores (openfut-identity + Core SQLite) recoverably and idempotently, handing
Core only a GENERIC request (all FIFA17 semantics stay in this adapter layer).

apply module:
- owned_item_id(persona, wire) = deterministic UUIDv5 from a private namespace;
  identical in BOTH stores (identity core_id AND Core owned_cards.id), so the
  running host's wire->owned reverse lookup resolves exactly what Core stored.
- plan_apply(report, raw_profile, fp): pure translation to a GenericImportRequest
  (card_id = fifa17_<resourceId>) + the preserved (owned_item_id <-> source wire)
  mappings + watermark. Canonical squad + Fifa17SquadExtensionV1 are built by the
  SAME adapter code (parse_squad_put + build_squad_write) the retail-validated
  live squad path uses. Refuses if the report has blockers.
- Two-store protocol (apply): staging gate -> local Core-preflight mirror ->
  identity dry-preflight -> idempotent seed (insert_existing_mapping per instance
  + set_watermark) -> ONE generic Core import transaction (spawned binary) ->
  cross-store post-validation -> completion record. A crash after identity
  seeding re-converges on re-run (idempotent mappings + Core already_imported):
  no cleanup, no reminting.
- gate_staging: deferred players are ABSENT from an import; allowed only for a
  staged run behind --allow-deferred-players-for-staging (never a silent default;
  prints an INCOMPLETE banner). Production requires zero deferred instances.

CLI: --apply (with --emit-content, --core-bin, --core-db, --core-data,
--identity-store, --allow-deferred-players-for-staging). Depends on
openfut-adapter-fifa17 + openfut-identity + uuid(v5).

Proven end-to-end on the real 33068179/CAGE profile (staged): first apply
imports 1949 supported instances (293 base + versioned), 11/11 f433 squad +
opaque extension, coins 28,112,944, fingerprint 8dc5582d2414af28; re-run is an
idempotent no-op (already_imported, DB unchanged); staging-not-default refuses
before any write; every OwnedItemId is an opaque UUID; identity wire-id set ==
source supported set exactly (0 minted, 0 dropped); 13 deferred instances leak 0.

10 new apply tests (determinism, request/mapping/squad translation, blocker
refusal, staging gate both ways, local preflight, identity seed/dry/postvalidate/
idempotency, conflict detection, graceful spawn failure). clippy -D warnings
clean; crate suite 23 tests green.

Cargo.lock

@@ -3218,9 +3218,12 @@ name = "openfut-import-fifa17"
 version = "0.1.0"
 dependencies = [
  "anyhow",
+ "openfut-adapter-fifa17",
+ "openfut-identity",
  "serde",
  "serde_json",
  "tempfile",
+ "uuid",
 ]
 
 [[package]]
@@ -4205,6 +4208,12 @@ dependencies = [
  "digest",
 ]
 
+[[package]]
+name = "sha1_smol"
+version = "1.0.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bbfa15b3dddfee50a0fff136974b3e1bde555604ba463834a7eb7deb6417705d"
+
 [[package]]
 name = "sha2"
 version = "0.10.9"
@@ -5345,6 +5354,7 @@ dependencies = [
  "getrandom 0.4.3",
  "js-sys",
  "serde_core",
+ "sha1_smol",
  "wasm-bindgen",
 ]
 

openfut-import-fifa17/Cargo.toml

@@ -16,6 +16,9 @@ path = "src/lib.rs"
 anyhow = "1"
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
+uuid = { version = "1", features = ["v5"] }
+openfut-adapter-fifa17 = { path = "../openfut-adapter-fifa17" }
+openfut-identity = { path = "../openfut-identity" }
 
 [dev-dependencies]
 tempfile = "3"

openfut-import-fifa17/src/apply.rs

@@ -0,0 +1,494 @@
+//! `--apply`: the FIFA17-specific orchestration that installs a real profile
+//! across the TWO durable stores — the `openfut-identity` external-id store and
+//! OpenFUT Core's SQLite DB — recoverably and idempotently.
+//!
+//! Boundary: everything FIFA17-specific stays here. This module chooses every
+//! `CardDefinitionId` (`fifa17_<resourceId>`, from the report) and every
+//! deterministic opaque Core `OwnedItemId`, preserves each owned instance's
+//! existing Python wire id, builds the canonical squad + `Fifa17SquadExtensionV1`
+//! with the SAME adapter code the retail-validated live squad path uses, then
+//! hands Core a GENERIC [`GenericImportRequest`]. Core never sees a resourceId,
+//! a wire id, or the extension's meaning.
+//!
+//! Recoverable two-store protocol (see [`apply`]):
+//!   validate → local Core-preflight mirror → identity dry-preflight →
+//!   idempotently install existing wire-id mappings + watermark →
+//!   ONE generic Core import transaction → cross-store post-validation →
+//!   completion record.
+//! If Core fails after identity seeding, rerunning the same manifest re-installs
+//! identical identity mappings (idempotent) and Core converges (same
+//! `source_fingerprint` → `already_imported`). No cleanup, no reminting.
+
+use std::collections::{BTreeSet, HashMap};
+use std::path::{Path, PathBuf};
+use std::process::Command;
+
+use anyhow::{bail, Context, Result};
+use serde::Serialize;
+use uuid::Uuid;
+
+use openfut_adapter_fifa17::fut::catalog::Fifa17WireItemIdPolicy;
+use openfut_adapter_fifa17::fut::squad::{parse_squad_put, SquadWireResolver};
+use openfut_adapter_fifa17::fut::squad_ext::{
+    build_squad_write, EXT_NAMESPACE, EXT_SCHEMA_VERSION,
+};
+use openfut_identity::{ExternalIdentityStore, JsonIdentityStore};
+
+use crate::Report;
+
+/// Private, project-specific UUIDv5 namespace for deriving Core OwnedItemIds from
+/// a source account + source owned-item identity. Fixed forever so a re-import of
+/// the same account yields the SAME opaque ids (idempotent across both stores).
+/// Core never parses why these ids are stable — they are opaque primary keys.
+const OWNED_ITEM_NAMESPACE: Uuid = Uuid::from_bytes([
+    0x0f, 0x17, 0x0f, 0x17, 0x9a, 0x11, 0x5e, 0xd0, 0xb0, 0x0f, 0x0e, 0x11, 0x0c, 0x0a, 0x11, 0x17,
+]);
+/// Deterministic opaque Core OwnedItemId for one source owned instance. Stable
+/// across runs and identical in BOTH stores (identity mapping core_id AND Core
+/// `owned_cards.id`), so the running host's wire→owned reverse lookup resolves
+/// exactly what Core stored.
+pub fn owned_item_id(persona_id: i64, source_wire_id: i64) -> String {
+    let name = format!("fifa17:{persona_id}:{source_wire_id}");
+    Uuid::new_v5(&OWNED_ITEM_NAMESPACE, name.as_bytes()).to_string()
+}
+
+// ----------------------------------------------------- generic Core request
+
+#[derive(Debug, Serialize)]
+pub struct GenericProfile {
+    pub username: String,
+    pub game_id: String,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GenericClub {
+    pub name: String,
+    pub coins: i64,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GenericOwned {
+    pub owned_item_id: String,
+    pub card_id: String,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GenericSlot {
+    pub owned_item_id: String,
+    pub position_index: i64,
+    pub is_captain: bool,
+    pub is_on_bench: bool,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GenericExtension {
+    pub namespace: String,
+    pub schema_version: i64,
+    pub payload: String,
+}
+
+#[derive(Debug, Serialize)]
+pub struct GenericSquad {
+    pub formation: String,
+    pub name: String,
+    pub slots: Vec<GenericSlot>,
+    pub extension: GenericExtension,
+}
+
+/// The game-agnostic request Core consumes. Field names match
+/// `openfut_core::services::import::ProfileImportRequest` exactly.
+#[derive(Debug, Serialize)]
+pub struct GenericImportRequest {
+    pub source_fingerprint: String,
+    pub profile: GenericProfile,
+    pub club: GenericClub,
+    pub owned: Vec<GenericOwned>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub squad: Option<GenericSquad>,
+}
+
+/// One preserved (opaque OwnedItemId ↔ source wire id) mapping to install into
+/// the identity store under `(fifa17, owned-item)`.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct IdentityMapping {
+    pub core_id: String,
+    pub wire_id: i64,
+}
+
+/// The complete, side-effect-free plan for an apply: the generic Core request,
+/// the identity mappings to install, and the watermark to set.
+#[derive(Debug)]
+pub struct ApplyPlan {
+    pub request: GenericImportRequest,
+    pub mappings: Vec<IdentityMapping>,
+    pub watermark: i64,
+    pub supported_instances: usize,
+    pub deferred_instances: usize,
+    pub source_fingerprint: String,
+}
+
+struct MapResolver<'a>(&'a HashMap<i64, String>);
+impl SquadWireResolver for MapResolver<'_> {
+    fn owned_id_for_wire(&self, wire: i64) -> Option<String> {
+        self.0.get(&wire).cloned()
+    }
+}
+
+/// Build the full apply plan from an already-analysed [`Report`] + the raw
+/// profile JSON (needed verbatim for the squad wire: `custom`, `kicktakers`,
+/// `kitNumber`, manager, squadType). Pure: no store or Core writes.
+///
+/// Refuses if the report has blockers — never plans an unsafe or lossy import.
+pub fn plan_apply(
+    report: &Report,
+    raw_profile: &serde_json::Value,
+    snapshot_fingerprint: &str,
+) -> Result<ApplyPlan> {
+    if report.has_blockers() {
+        bail!(
+            "refusing to plan apply while blockers are present: {}",
+            report.blockers().join("; ")
+        );
+    }
+    let persona = report.persona_id;
+
+    // Owned cards + identity mappings: one per SUPPORTED owned instance. Each
+    // instance keeps its existing Python wire id; its Core OwnedItemId is derived
+    // deterministically from (persona, wire).
+    let mut owned = Vec::with_capacity(report.identity.import_wire_ids.len());
+    let mut mappings = Vec::with_capacity(report.identity.import_wire_ids.len());
+    let mut wire_to_owned: HashMap<i64, String> = HashMap::new();
+    for def in &report.definitions.supported {
+        for &wire in &def.wire_ids {
+            let core_id = owned_item_id(persona, wire);
+            wire_to_owned.insert(wire, core_id.clone());
+            owned.push(GenericOwned {
+                owned_item_id: core_id.clone(),
+                card_id: def.card_id.clone(),
+            });
+            mappings.push(IdentityMapping {
+                core_id,
+                wire_id: wire,
+            });
+        }
+    }
+
+    // Canonical squad + opaque extension, built by the SAME adapter code the live
+    // squad-write path uses, over the raw source squad. The resolver maps every
+    // occupied wire id to its deterministic OwnedItemId.
+    let squad = if report.squad.present {
+        let raw_squad = raw_profile
+            .get("squads")
+            .and_then(|s| s.get(0))
+            .context("report says a squad is present but profile has no squads[0]")?;
+        let body = serde_json::to_vec(raw_squad).context("re-serialize source squad")?;
+        let put = parse_squad_put(&body).map_err(|e| anyhow::anyhow!("parse source squad: {e}"))?;
+        let build = build_squad_write(&put, &MapResolver(&wire_to_owned))
+            .map_err(|e| anyhow::anyhow!("build squad write: {e}"))?;
+        let formation = build
+            .canonical
+            .formation
+            .clone()
+            .context("source squad has no formation token")?;
+        let slots = build
+            .canonical
+            .slots
+            .iter()
+            .map(|s| GenericSlot {
+                owned_item_id: s.owned_card_id.clone(),
+                position_index: s.index,
+                is_captain: s.is_captain,
+                is_on_bench: s.is_on_bench,
+            })
+            .collect();
+        Some(GenericSquad {
+            formation,
+            name: build
+                .canonical
+                .name
+                .clone()
+                .unwrap_or_else(|| report.club_name.clone()),
+            slots,
+            extension: GenericExtension {
+                namespace: EXT_NAMESPACE.to_string(),
+                schema_version: EXT_SCHEMA_VERSION,
+                payload: build.extension.to_payload(),
+            },
+        })
+    } else {
+        None
+    };
+
+    let request = GenericImportRequest {
+        source_fingerprint: snapshot_fingerprint.to_string(),
+        profile: GenericProfile {
+            username: report.persona_name.clone(),
+            game_id: report.game.clone(),
+        },
+        club: GenericClub {
+            name: report.club_name.clone(),
+            coins: report.coins,
+        },
+        owned,
+        squad,
+    };
+
+    Ok(ApplyPlan {
+        request,
+        mappings,
+        watermark: report.identity.source_watermark,
+        supported_instances: report.identity.import_wire_ids.len(),
+        deferred_instances: report.deferred_instances(),
+        source_fingerprint: snapshot_fingerprint.to_string(),
+    })
+}
+
+// -------------------------------------------------------------- side effects
+
+/// Where the apply reads/writes the two stores and Core.
+pub struct ApplyPaths {
+    /// The `openfut-core` binary to invoke for the generic import transaction.
+    pub core_bin: PathBuf,
+    /// `DATABASE_URL` passed to that Core invocation (the target profile DB).
+    pub core_db_url: String,
+    /// The emitted production content pack (`OPENFUT_CONTENT_PACKS`).
+    pub content_pack: PathBuf,
+    /// `DATA_DIR` for the Core invocation (base catalog directory).
+    pub data_dir: String,
+    /// The identity store JSON file.
+    pub identity_store: PathBuf,
+    /// Where to write the generic request JSON handed to Core.
+    pub request_out: PathBuf,
+    /// Where to write the completion record.
+    pub completion_out: PathBuf,
+}
+
+#[derive(Debug, Serialize)]
+pub struct ApplyOutcome {
+    pub core_outcome: String,
+    pub mappings_installed: usize,
+    pub watermark: i64,
+    pub supported_instances: usize,
+    pub deferred_instances: usize,
+    pub staging: bool,
+    pub source_fingerprint: String,
+}
+
+/// The staging/production gate. Deferred players are absent from an import; that
+/// is acceptable ONLY for a staged run, and ONLY with the explicit opt-in — never
+/// a silent default. Production requires zero deferred instances.
+pub fn gate_staging(plan: &ApplyPlan, allow_deferred_staging: bool) -> Result<bool> {
+    if plan.deferred_instances == 0 {
+        return Ok(false); // production-complete
+    }
+    if !allow_deferred_staging {
+        bail!(
+            "{} player instance(s) are DEFERRED (unsupported/conflicted) and would be ABSENT from \
+             this import. Production import requires zero deferred instances. To import anyway for \
+             a STAGED test, pass --allow-deferred-players-for-staging (this is never the default).",
+            plan.deferred_instances
+        );
+    }
+    eprintln!(
+        "\n================= STAGING IMPORT (INCOMPLETE) =================\n\
+         {} player instance(s) are DEFERRED and WILL BE ABSENT from this import.\n\
+         This profile is NOT production-complete. Resolve the deferred definitions\n\
+         before a production cutover.\n\
+         ==============================================================\n",
+        plan.deferred_instances
+    );
+    Ok(true)
+}
+
+/// Local mirror of Core's pre-transaction preflight, run BEFORE touching either
+/// store so a doomed import never seeds identity. `content_card_ids` is the set
+/// of CardDefinitionIds in the emitted production pack.
+pub fn local_core_preflight(plan: &ApplyPlan, content_card_ids: &BTreeSet<String>) -> Result<()> {
+    let mut owned_ids: BTreeSet<&str> = BTreeSet::new();
+    for o in &plan.request.owned {
+        if !content_card_ids.contains(&o.card_id) {
+            bail!(
+                "local preflight: owned card references CardDefinitionId {} absent from the \
+                 production content pack",
+                o.card_id
+            );
+        }
+        if !owned_ids.insert(o.owned_item_id.as_str()) {
+            bail!("local preflight: duplicate OwnedItemId {}", o.owned_item_id);
+        }
+    }
+    if let Some(sq) = &plan.request.squad {
+        for slot in &sq.slots {
+            if !owned_ids.contains(slot.owned_item_id.as_str()) {
+                bail!(
+                    "local preflight: squad slot OwnedItemId {} not in imported ownership set",
+                    slot.owned_item_id
+                );
+            }
+        }
+    }
+    Ok(())
+}
+
+/// Dry-check that installing the identity mappings would not conflict with any
+/// mapping already present (a re-run's identical mappings are fine). No writes.
+pub fn identity_dry_preflight(store: &JsonIdentityStore, plan: &ApplyPlan) -> Result<()> {
+    let (g, k) = (
+        Fifa17WireItemIdPolicy::GAME,
+        Fifa17WireItemIdPolicy::OWNED_ITEM_KIND,
+    );
+    let mut conflicts = Vec::new();
+    for m in &plan.mappings {
+        if let Some(existing) = store.external_for(g, k, &m.core_id)? {
+            if existing != m.wire_id {
+                conflicts.push(format!(
+                    "core {} already maps to wire {existing} (want {})",
+                    m.core_id, m.wire_id
+                ));
+            }
+        }
+        if let Some(existing) = store.core_for(g, k, m.wire_id)? {
+            if existing != m.core_id {
+                conflicts.push(format!(
+                    "wire {} already maps to core {existing} (want {})",
+                    m.wire_id, m.core_id
+                ));
+            }
+        }
+    }
+    if !conflicts.is_empty() {
+        bail!(
+            "identity dry-preflight found {} conflict(s): {}",
+            conflicts.len(),
+            conflicts.join("; ")
+        );
+    }
+    Ok(())
+}
+
+/// Idempotently install every preserved wire-id mapping + the allocation
+/// watermark. Re-running re-inserts identical mappings (no-op) and re-sets the
+/// same watermark — safe after a mid-apply crash.
+pub fn seed_identity(store: &JsonIdentityStore, plan: &ApplyPlan) -> Result<()> {
+    let (g, k) = (
+        Fifa17WireItemIdPolicy::GAME,
+        Fifa17WireItemIdPolicy::OWNED_ITEM_KIND,
+    );
+    for m in &plan.mappings {
+        store
+            .insert_existing_mapping(g, k, &m.core_id, m.wire_id)
+            .with_context(|| format!("install mapping {} -> {}", m.core_id, m.wire_id))?;
+    }
+    store
+        .set_watermark(g, k, plan.watermark)
+        .context("set allocation watermark")?;
+    Ok(())
+}
+
+/// After a successful Core import, re-open the identity store and assert every
+/// preserved wire id resolves exactly, and the watermark is set. Core's own
+/// return (`imported`/`already_imported`) is the Core-side proof.
+pub fn post_validate_identity(store: &JsonIdentityStore, plan: &ApplyPlan) -> Result<()> {
+    let (g, k) = (
+        Fifa17WireItemIdPolicy::GAME,
+        Fifa17WireItemIdPolicy::OWNED_ITEM_KIND,
+    );
+    for m in &plan.mappings {
+        let got = store.external_for(g, k, &m.core_id)?;
+        if got != Some(m.wire_id) {
+            bail!(
+                "post-validate: core {} resolves to {:?}, expected wire {}",
+                m.core_id,
+                got,
+                m.wire_id
+            );
+        }
+    }
+    match store.watermark_for(g, k) {
+        Some(w) if w == plan.watermark => Ok(()),
+        other => bail!(
+            "post-validate: watermark is {:?}, expected {}",
+            other,
+            plan.watermark
+        ),
+    }
+}
+
+/// Load the CardDefinitionIds from an emitted production content pack.
+pub fn content_card_ids(pack: &Path) -> Result<BTreeSet<String>> {
+    let raw =
+        std::fs::read(pack).with_context(|| format!("read content pack {}", pack.display()))?;
+    let defs: Vec<serde_json::Value> =
+        serde_json::from_slice(&raw).context("parse content pack as CardDefinition[]")?;
+    Ok(defs
+        .iter()
+        .filter_map(|d| d.get("id").and_then(|v| v.as_str()).map(String::from))
+        .collect())
+}
+
+/// Full recoverable apply across both stores. See module docs for the protocol.
+pub fn apply(
+    plan: &ApplyPlan,
+    paths: &ApplyPaths,
+    allow_deferred_staging: bool,
+) -> Result<ApplyOutcome> {
+    // 1. staging/production gate (explicit, never silent).
+    let staging = gate_staging(plan, allow_deferred_staging)?;
+
+    // 2. local Core-preflight mirror BEFORE any store write.
+    let card_ids = content_card_ids(&paths.content_pack)?;
+    local_core_preflight(plan, &card_ids)?;
+
+    // 3. identity dry-preflight, then idempotent seed + watermark.
+    let store = JsonIdentityStore::open(&paths.identity_store)
+        .with_context(|| format!("open identity store {}", paths.identity_store.display()))?;
+    identity_dry_preflight(&store, plan)?;
+    seed_identity(&store, plan)?;
+
+    // 4. one generic Core import transaction (Core owns atomicity).
+    let req_json = serde_json::to_vec_pretty(&plan.request)?;
+    std::fs::write(&paths.request_out, &req_json)
+        .with_context(|| format!("write request {}", paths.request_out.display()))?;
+    let out = Command::new(&paths.core_bin)
+        .arg("import")
+        .arg(&paths.request_out)
+        .env("DATABASE_URL", &paths.core_db_url)
+        .env("DATA_DIR", &paths.data_dir)
+        .env("OPENFUT_CONTENT_PACKS", paths.content_pack.as_os_str())
+        .output()
+        .with_context(|| format!("spawn core import: {}", paths.core_bin.display()))?;
+    if !out.status.success() {
+        bail!(
+            "core import failed ({}): {}\n(identity already seeded idempotently; re-run the same \
+             manifest to converge — no cleanup needed)",
+            out.status,
+            String::from_utf8_lossy(&out.stderr)
+        );
+    }
+    let stdout = String::from_utf8_lossy(&out.stdout);
+    let core_result: serde_json::Value =
+        serde_json::from_str(stdout.trim()).context("parse core import outcome JSON")?;
+    let core_outcome = core_result
+        .get("outcome")
+        .and_then(|v| v.as_str())
+        .unwrap_or("unknown")
+        .to_string();
+
+    // 5. cross-store post-validation.
+    post_validate_identity(&store, plan)?;
+
+    // 6. completion record.
+    let outcome = ApplyOutcome {
+        core_outcome,
+        mappings_installed: plan.mappings.len(),
+        watermark: plan.watermark,
+        supported_instances: plan.supported_instances,
+        deferred_instances: plan.deferred_instances,
+        staging,
+        source_fingerprint: plan.source_fingerprint.clone(),
+    };
+    let rec = serde_json::to_vec_pretty(&outcome)?;
+    std::fs::write(&paths.completion_out, &rec)
+        .with_context(|| format!("write completion {}", paths.completion_out.display()))?;
+    Ok(outcome)
+}

openfut-import-fifa17/src/lib.rs

@@ -30,6 +30,7 @@ use anyhow::{Context, Result};
 /// FIFA 17 owned-item wire-id floor (adapter `Fifa17WireItemIdPolicy`).
 pub const OWNED_ITEM_BASE_FLOOR: i64 = 100_000_001;
 
+pub mod apply;
 pub mod model;
 
 use model::{Item, Profile};

openfut-import-fifa17/src/main.rs

@@ -10,6 +10,7 @@ use std::collections::BTreeSet;
 use std::path::Path;
 use std::process::ExitCode;
 
+use openfut_import_fifa17::apply::{apply, plan_apply, ApplyPaths};
 use openfut_import_fifa17::{
     analyze, emit_content, fingerprint, load_roster, model::Profile, Entities,
 };
@@ -36,7 +37,12 @@ fn print_help() {
          --tables <dir>            tables dir (default fifa17-recon/data/tables)\n\
          --defer-conflict <rid>    explicitly approve deferring a reviewed resourceId conflict (repeatable)\n\
          --emit-content <dir>      write the production content pack + host catalog + private manifest\n\
-         --apply                   REFUSED in this phase (no Core/identity writes)\n"
+         --apply                   apply across the identity store + Core (requires --emit-content)\n\
+         --allow-deferred-players-for-staging   import despite deferred players (STAGED only; never default)\n\
+         --core-bin <path>         openfut-core binary for the generic import transaction\n\
+         --core-db <url>           DATABASE_URL for the target Core DB\n\
+         --core-data <dir>         DATA_DIR base catalog for Core (default data)\n\
+         --identity-store <path>   openfut-identity JSON store to seed\n"
     );
 }
 
@@ -46,6 +52,12 @@ fn run() -> Result<ExitCode> {
     let mut tables_dir = "fifa17-recon/data/tables".to_string();
     let mut emit_dir: Option<String> = None;
     let mut approved: BTreeSet<i64> = BTreeSet::new();
+    let mut do_apply = false;
+    let mut allow_staging = false;
+    let mut core_bin: Option<String> = None;
+    let mut core_db: Option<String> = None;
+    let mut core_data = "data".to_string();
+    let mut identity_store: Option<String> = None;
 
     let mut args = std::env::args().skip(1);
     while let Some(a) = args.next() {
@@ -62,9 +74,14 @@ fn run() -> Result<ExitCode> {
                 );
             }
             "--dry-run" => {}
-            "--apply" => bail!(
-                "--apply is not implemented in this phase; refusing to write the Core DB or identity store"
-            ),
+            "--apply" => do_apply = true,
+            "--allow-deferred-players-for-staging" => allow_staging = true,
+            "--core-bin" => core_bin = Some(args.next().context("--core-bin needs a path")?),
+            "--core-db" => core_db = Some(args.next().context("--core-db needs a DATABASE_URL")?),
+            "--core-data" => core_data = args.next().context("--core-data needs a dir")?,
+            "--identity-store" => {
+                identity_store = Some(args.next().context("--identity-store needs a path")?)
+            }
             "-h" | "--help" => {
                 print_help();
                 return Ok(ExitCode::SUCCESS);
@@ -108,6 +125,48 @@ fn run() -> Result<ExitCode> {
         );
     }
 
+    if do_apply {
+        let dir = emit_dir
+            .as_deref()
+            .context("--apply requires --emit-content <dir> (it reads the emitted content pack)")?;
+        if report.has_blockers() {
+            bail!("not applying: unsafe blockers present (resolve them first)");
+        }
+        let raw_value: serde_json::Value =
+            serde_json::from_slice(&raw).context("parse profile JSON for squad wire")?;
+        let fp = fingerprint(&raw);
+        let plan = plan_apply(&report, &raw_value, &fp)?;
+        let base = Path::new(dir);
+        let paths = ApplyPaths {
+            core_bin: core_bin
+                .context("--apply requires --core-bin <openfut-core binary>")?
+                .into(),
+            core_db_url: core_db.context("--apply requires --core-db <DATABASE_URL>")?,
+            content_pack: base.join("content/fifa17-production-cards.json"),
+            data_dir: core_data,
+            identity_store: identity_store
+                .context("--apply requires --identity-store <path>")?
+                .into(),
+            request_out: base.join("manifest/fifa17-core-import-request.json"),
+            completion_out: base.join("manifest/fifa17-import-completion.json"),
+        };
+        let outcome = apply(&plan, &paths, allow_staging)?;
+        println!(
+            "\nAPPLIED (core_outcome={} staging={})",
+            outcome.core_outcome, outcome.staging
+        );
+        println!(
+            "  identity mappings installed: {} ; watermark: {}",
+            outcome.mappings_installed, outcome.watermark
+        );
+        println!(
+            "  supported instances: {} ; deferred instances: {}",
+            outcome.supported_instances, outcome.deferred_instances
+        );
+        println!("  request    : {}", paths.request_out.display());
+        println!("  completion : {}", paths.completion_out.display());
+    }
+
     if report.has_blockers() {
         eprintln!("import-fifa17: FAILED — unsafe blockers present (see above)");
         Ok(ExitCode::FAILURE)

openfut-import-fifa17/src/tests.rs

@@ -318,3 +318,243 @@ fn emit_refuses_when_blocked() {
     let dir = tempfile::tempdir().unwrap();
     assert!(emit_content(&rep, dir.path(), "x").is_err());
 }
+
+// ------------------------------------------------------------------ apply
+
+use crate::apply::{
+    apply as apply_import, content_card_ids, gate_staging, identity_dry_preflight,
+    local_core_preflight, owned_item_id, plan_apply, post_validate_identity, seed_identity,
+};
+use openfut_identity::{ExternalIdentityStore, JsonIdentityStore};
+
+const SQUAD_F433: &str = r#"[{"formation":"f433","squadName":"OpenFUT","captain":100000001,
+    "squadType":"REGULAR_SQUAD","custom":"[0,0,0]",
+    "players":[{"index":0,"itemData":{"id":100000001},"kitNumber":7},
+               {"index":1,"itemData":{"id":100000002},"kitNumber":9}],
+    "kicktakers":[{"index":0,"id":100000001}],"manager":[{"id":100000427}]}]"#;
+
+/// Build a report + raw profile Value from the same JSON (typed profile for
+/// analysis, raw Value for the squad wire).
+fn report_and_raw(
+    items: &[String],
+    squads: &str,
+    next_item_id: i64,
+) -> (Report, serde_json::Value) {
+    let json = format!(
+        r#"{{"personaId":33068179,"personaName":"CAGE","clubName":"OpenFUT","clubAbbr":"OFC",
+             "coins":1000,"nextItemId":{next_item_id},"items":[{}],"squads":{}}}"#,
+        items.join(","),
+        squads
+    );
+    let prof = Profile::from_json_str(&json).unwrap();
+    let report = analyze(&prof, &roster(), &entities(), &none());
+    let raw: serde_json::Value = serde_json::from_str(&json).unwrap();
+    (report, raw)
+}
+
+#[test]
+fn owned_item_id_is_deterministic_and_distinct() {
+    let a = owned_item_id(33068179, 100000001);
+    assert_eq!(a, owned_item_id(33068179, 100000001), "stable across calls");
+    assert_ne!(
+        a,
+        owned_item_id(33068179, 100000002),
+        "distinct per wire id"
+    );
+    assert_ne!(
+        a,
+        owned_item_id(90909090, 100000001),
+        "distinct per account"
+    );
+}
+
+#[test]
+fn plan_apply_builds_generic_request_mappings_and_squad() {
+    let items = vec![
+        player(100000001, 20801, 20801, 94),
+        player(100000002, VER5_176580, 176580, 92),
+    ];
+    let (report, raw) = report_and_raw(&items, SQUAD_F433, 100000500);
+    let plan = plan_apply(&report, &raw, "fp-test").unwrap();
+
+    assert_eq!(plan.request.owned.len(), 2);
+    assert_eq!(plan.mappings.len(), 2);
+    assert_eq!(plan.watermark, 100000500);
+    assert_eq!(plan.supported_instances, 2);
+    assert_eq!(plan.deferred_instances, 0);
+    assert_eq!(plan.source_fingerprint, "fp-test");
+    // card ids are fifa17_<resourceId>, base and versioned distinct.
+    let mut cards: Vec<&str> = plan
+        .request
+        .owned
+        .iter()
+        .map(|o| o.card_id.as_str())
+        .collect();
+    cards.sort_unstable();
+    assert_eq!(
+        cards,
+        vec!["fifa17_20801", &format!("fifa17_{VER5_176580}")[..]]
+    );
+    // mapping core_id == the owned_item_id for that wire (both stores agree).
+    for m in &plan.mappings {
+        assert_eq!(m.core_id, owned_item_id(33068179, m.wire_id));
+    }
+    // squad + opaque extension.
+    let sq = plan.request.squad.as_ref().expect("squad present");
+    assert_eq!(sq.formation, "f433");
+    assert_eq!(sq.slots.len(), 2);
+    assert_eq!(sq.extension.namespace, "fifa17.squad");
+    assert_eq!(sq.extension.schema_version, 1);
+    assert!(
+        sq.extension.payload.contains("[0,0,0]"),
+        "custom carried verbatim"
+    );
+    // captain flag follows wire 100000001, keyed by its OwnedItemId.
+    let cap_owned = owned_item_id(33068179, 100000001);
+    assert!(sq
+        .slots
+        .iter()
+        .any(|s| s.owned_item_id == cap_owned && s.is_captain));
+}
+
+#[test]
+fn plan_apply_refuses_when_blockers_present() {
+    // unapproved same-resourceId conflict -> blocker.
+    let items = vec![
+        player(100000003, 20801, 20801, 94),
+        player(100000004, 20801, 20801, 90),
+    ];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    assert!(plan_apply(&report, &raw, "fp").is_err());
+}
+
+#[test]
+fn gate_staging_requires_explicit_optin_for_deferred() {
+    // one supported + one NoName (unresolvable asset) deferred instance.
+    let items = vec![
+        player(100000001, 20801, 20801, 94),
+        format!(
+            r#"{{"id":100000009,"resourceId":999999,"assetId":999999,"itemType":"player","rating":80,
+            "preferredPosition":"ST","nation":38,"teamid":243,"leagueId":53,"attributeList":{}}}"#,
+            attrs()
+        ),
+    ];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+    assert_eq!(plan.deferred_instances, 1);
+    assert!(gate_staging(&plan, false).is_err(), "deferred needs opt-in");
+    assert!(gate_staging(&plan, true).unwrap(), "staging with opt-in");
+}
+
+#[test]
+fn gate_staging_production_complete_needs_no_flag() {
+    let items = vec![player(100000001, 20801, 20801, 94)];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+    assert_eq!(plan.deferred_instances, 0);
+    assert!(!gate_staging(&plan, false).unwrap());
+}
+
+#[test]
+fn local_preflight_rejects_owned_card_absent_from_content() {
+    let items = vec![player(100000001, 20801, 20801, 94)];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+    let mut present = BTreeSet::new();
+    present.insert("fifa17_20801".to_string());
+    assert!(local_core_preflight(&plan, &present).is_ok());
+    assert!(local_core_preflight(&plan, &BTreeSet::new()).is_err());
+}
+
+#[test]
+fn identity_seed_dry_postvalidate_and_idempotent_rerun() {
+    let items = vec![
+        player(100000001, 20801, 20801, 94),
+        player(100000002, VER5_176580, 176580, 92),
+    ];
+    let (report, raw) = report_and_raw(&items, SQUAD_F433, 100004617);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+
+    let dir = tempfile::tempdir().unwrap();
+    let store = JsonIdentityStore::open(dir.path().join("ids.json")).unwrap();
+    identity_dry_preflight(&store, &plan).unwrap();
+    seed_identity(&store, &plan).unwrap();
+
+    // every preserved wire id resolves exactly; watermark set.
+    for m in &plan.mappings {
+        assert_eq!(
+            store
+                .external_for("fifa17", "owned-item", &m.core_id)
+                .unwrap(),
+            Some(m.wire_id)
+        );
+    }
+    assert_eq!(store.watermark_for("fifa17", "owned-item"), Some(100004617));
+    post_validate_identity(&store, &plan).unwrap();
+
+    // re-seeding the SAME plan is an idempotent no-op (crash recovery).
+    seed_identity(&store, &plan).unwrap();
+    post_validate_identity(&store, &plan).unwrap();
+}
+
+#[test]
+fn identity_dry_preflight_detects_conflicting_existing_mapping() {
+    let items = vec![player(100000001, 20801, 20801, 94)];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+
+    let dir = tempfile::tempdir().unwrap();
+    let store = JsonIdentityStore::open(dir.path().join("ids.json")).unwrap();
+    // wire 100000001 already owned by a DIFFERENT core id.
+    store
+        .insert_existing_mapping("fifa17", "owned-item", "someone-else", 100000001)
+        .unwrap();
+    assert!(identity_dry_preflight(&store, &plan).is_err());
+}
+
+#[test]
+fn content_card_ids_reads_emitted_pack() {
+    let items = vec![player(100000001, 20801, 20801, 94)];
+    let rep = analyze(
+        &profile(&items, "[]", 100000500),
+        &roster(),
+        &entities(),
+        &none(),
+    );
+    let dir = tempfile::tempdir().unwrap();
+    let sum = emit_content(&rep, dir.path(), "fp").unwrap();
+    let ids = content_card_ids(&sum.content_pack).unwrap();
+    assert!(ids.contains("fifa17_20801"));
+}
+
+/// The full apply spawns the Core binary; that end-to-end path is exercised by
+/// the staged runtime gate, not here. This asserts the orchestration refuses
+/// cleanly (no panic, no partial identity seed) when the Core binary is absent
+/// AFTER the two local gates pass — proving ordering: gates first, spawn last.
+#[test]
+fn apply_fails_gracefully_when_core_binary_missing() {
+    use crate::apply::ApplyPaths;
+    let items = vec![player(100000001, 20801, 20801, 94)];
+    let (report, raw) = report_and_raw(&items, "[]", 100000500);
+    let plan = plan_apply(&report, &raw, "fp").unwrap();
+    let dir = tempfile::tempdir().unwrap();
+    // emit a real content pack so local preflight passes.
+    let rep2 = analyze(
+        &profile(&items, "[]", 100000500),
+        &roster(),
+        &entities(),
+        &none(),
+    );
+    let sum = emit_content(&rep2, dir.path(), "fp").unwrap();
+    let paths = ApplyPaths {
+        core_bin: dir.path().join("no-such-core-binary"),
+        core_db_url: "sqlite::memory:".to_string(),
+        content_pack: sum.content_pack,
+        data_dir: "data".to_string(),
+        identity_store: dir.path().join("ids.json"),
+        request_out: dir.path().join("req.json"),
+        completion_out: dir.path().join("done.json"),
+    };
+    let err = apply_import(&plan, &paths, false).unwrap_err();
+    assert!(format!("{err:#}").contains("spawn core import"), "{err:#}");
+}