hypercall/rsm/portfolio_margin/

settlement_state.rs

use hypercall_engine::command::{
    AccruePmSettlementInterestCommand, ApplyPmSettlementRepaymentCommand,
    JournalPmRecoveryPlanCommand, MarkPmRecoveryActionSubmittedCommand, PmRecoveryActionKind,
    PmRecoveryActionResult, PmRecoveryReason, PmSettlementEventKey, RecordPmVaultDepositCommand,
    RequestPmVaultWithdrawalCommand, ResolvePmRecoveryActionCommand,
    SetPmSettlementPoolConfigCommand, TickExpiryPmSettlement,
};
use hypercall_margin::portfolio::{
    classify_liquidity_gap, pool_capacity_usdc, pool_utilization, utilization_apr,
    validate_pool_config, PmLiquidityClassification, PmSettlementPoolConfig,
    PmSettlementPoolSnapshot,
};
use hypercall_types::WalletAddress;
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::collections::BTreeMap;
use uuid::Uuid;

const INTEREST_YEAR_MS: u64 = 365 * 24 * 60 * 60 * 1000;

#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementState {
    pub pools: BTreeMap<String, PmSettlementPoolState>,
    pub accounts: BTreeMap<PmSettlementAccountKey, PmSettlementAccountState>,
    /// Durable classification history retained for replay idempotency.
    ///
    /// TODO: move this out of serialized engine snapshot state. Cardinality
    /// grows with the historical settlement cross-product of PM wallets,
    /// expiring markets, expiry timestamps, margin modes, and settlement event
    /// sequences. The long-term shape should keep only active settlement state
    /// in `PmSettlementState` and put historical event/request dedupe behind a
    /// bounded cursor, a gap-buffered stream index, or an engine-owned durable
    /// KV store such as RocksDB.
    pub events: BTreeMap<PmSettlementEventKey, PmSettlementEventState>,
    /// Request-id dedupe for PM settlement commands.
    ///
    /// TODO: remove this from snapshot state with `events`. It grows with
    /// historical command count rather than active accounts, so it should become
    /// a bounded replay-protection index instead of an in-memory audit log.
    pub idempotency: BTreeMap<String, PmSettlementRequestRecord>,
    #[serde(default)]
    pub vault_deposits: BTreeMap<Uuid, PmVaultDepositState>,
    #[serde(default)]
    pub vault_withdrawals: BTreeMap<Uuid, PmVaultWithdrawalState>,
    // CALL-1900: AWS staging replay bridge for snapshots written before PM
    // recovery plans were appended. Remove after staging boots once, writes a
    // current snapshot, and restarts without unsafe snapshot replay.
    #[serde(default)]
    pub recovery_plans: BTreeMap<String, PmRecoveryPlanState>,
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub struct PmSettlementAccountKey {
    pub wallet: WalletAddress,
    pub underlying: String,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementPoolState {
    pub underlying: String,
    pub pool_available_usdc: Decimal,
    pub pool_target_usdc: Decimal,
    pub active_timing_bridge_usdc: Decimal,
    pub active_settlement_debt_usdc: Decimal,
    pub config_version: u32,
    pub policy_version: u32,
    pub config: Option<PmSettlementPoolConfig>,
    pub utilization: Option<Decimal>,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PmSettlementAccountStatus {
    Clean,
    Bridged,
    Debt,
    Resolved,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementAccountState {
    pub wallet: WalletAddress,
    pub underlying: String,
    pub bridge_principal_usdc: Decimal,
    pub debt_principal_usdc: Decimal,
    pub accrued_interest_usdc: Decimal,
    pub last_interest_accrual_ms: i64,
    pub policy_version: u32,
    pub bridge_deadline_ms: Option<i64>,
    pub status: PmSettlementAccountStatus,
    pub active_recovery_plan_id: Option<String>,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementEventState {
    pub event_key: PmSettlementEventKey,
    pub input_digest: String,
    pub output_digest: String,
    pub status: String,
    pub request_id: Uuid,
    pub underlying: String,
    pub event_type: String,
    pub amount_usdc: Decimal,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementRequestRecord {
    pub request_id: Uuid,
    pub input_digest: String,
    pub output_digest: String,
    pub command_type: String,
    pub applied_at_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmVaultDepositState {
    pub deposit_id: Uuid,
    pub depositor: WalletAddress,
    pub underlying: String,
    pub principal_usdc: Decimal,
    pub remaining_usdc: Decimal,
    pub withdrawn_usdc: Decimal,
    pub reserved_withdrawal_usdc: Decimal,
    pub chain_id: u64,
    pub source_contract_address: WalletAddress,
    pub tx_hash: String,
    pub log_index: u32,
    pub max_listed_expiry_ts_ms: i64,
    pub settlement_grace_ms: i64,
    pub lock_until_ms: i64,
    pub status: PmVaultDepositStatus,
    pub created_at_ms: u64,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PmVaultDepositStatus {
    Active,
    PartiallyReserved,
    Reserved,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmVaultWithdrawalState {
    pub withdrawal_id: Uuid,
    pub deposit_id: Uuid,
    pub depositor: WalletAddress,
    pub underlying: String,
    pub amount_usdc: Decimal,
    pub lock_until_ms: i64,
    pub status: PmVaultWithdrawalStatus,
    pub requested_at_ms: u64,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum PmVaultWithdrawalStatus {
    Reserved,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmRecoveryPlanState {
    pub plan_id: String,
    pub wallet: WalletAddress,
    pub underlying: String,
    pub trigger: String,
    pub reason: String,
    pub policy_version: u32,
    pub recovery_priority_version: u32,
    pub status: String,
    pub input_digest: String,
    pub target_reduction_usdc: Decimal,
    pub expected_usdc_recovered: Decimal,
    pub expected_obligation_reduced: Decimal,
    pub expected_impact_usdc: Decimal,
    pub post_plan_utilization: Option<Decimal>,
    pub actions: Vec<PmRecoveryActionState>,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmRecoveryActionState {
    pub action_index: u32,
    pub action_type: String,
    pub target: String,
    pub action_payload: String,
    pub status: String,
    pub expected_usdc_recovered: Decimal,
    pub expected_obligation_reduced: Decimal,
    pub expected_impact_usdc: Decimal,
    pub attempt: u32,
    pub submitted_external_id: Option<String>,
    pub external_kind: Option<String>,
    pub result: Option<String>,
    pub result_external_id: Option<String>,
    pub recovered_usdc: Decimal,
    pub liability_reduction_usdc: Decimal,
    pub updated_at_ms: u64,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub enum PmSettlementProjectionEffect {
    PoolUpsert(PmSettlementPoolState),
    AccountUpsert(PmSettlementAccountState),
    InterestEvent(PmSettlementInterestEvent),
    RepaymentEvent(PmSettlementRepaymentEvent),
    VaultDepositUpsert(PmVaultDepositState),
    VaultWithdrawalUpsert(PmVaultWithdrawalState),
    EventUpsert(PmSettlementEventState),
    RecoveryPlanUpsert(PmRecoveryPlanState),
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementInterestEvent {
    pub request_id: Uuid,
    pub wallet: WalletAddress,
    pub underlying: String,
    pub accrual_start_ms: i64,
    pub accrual_end_ms: i64,
    pub utilization: Decimal,
    pub apr: Decimal,
    pub interest_amount_usdc: Decimal,
    pub policy_version: u32,
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct PmSettlementRepaymentEvent {
    pub request_id: Uuid,
    pub wallet: WalletAddress,
    pub underlying: String,
    pub source_event_id: String,
    pub amount_usdc: Decimal,
    pub interest_paid_usdc: Decimal,
    pub principal_paid_usdc: Decimal,
    pub pool_replenished_usdc: Decimal,
    pub reason: String,
    pub event_ts_ms: u64,
}

impl PmSettlementState {
    pub fn apply_classify_settlement(
        &mut self,
        command: TickExpiryPmSettlement,
        timestamp_ms: u64,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.facts_digest,
            "ClassifyPmSettlement",
        )?;
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.facts_digest,
            "ClassifyPmSettlement",
        )? {
            return Ok(existing);
        }
        if let Some(existing) = self.events.get(&command.event_key) {
            if existing.input_digest != command.facts_digest {
                return Err(format!(
                    "PM settlement event digest mismatch for {}",
                    format_event_key(&command.event_key)
                ));
            }
            self.record_request(
                command.request_id,
                command.facts_digest,
                "ClassifyPmSettlement",
                timestamp_ms,
            );
            return Ok(Vec::new());
        }

        let classification = match command.unavailable_reason.as_ref() {
            Some(_) => PmLiquidityClassification::Unavailable,
            None => {
                let facts = command.pm_facts.as_ref().ok_or_else(|| {
                    "ClassifyPmSettlement missing PM facts without unavailable reason".to_string()
                })?;
                if facts.wallet != command.wallet {
                    return Err(
                        "ClassifyPmSettlement PM facts wallet mismatch for settlement command"
                            .to_string(),
                    );
                }
                if facts.underlying != command.underlying {
                    return Err(
                        "ClassifyPmSettlement PM facts underlying mismatch for settlement command"
                            .to_string(),
                    );
                }
                if facts.liquid_usdc != command.liquid_usdc {
                    return Err(
                        "ClassifyPmSettlement liquid_usdc mismatch between command and PM facts"
                            .to_string(),
                    );
                }
                let pool_snapshot = command.pool_snapshot.as_ref().ok_or_else(|| {
                    "ClassifyPmSettlement missing pool snapshot without unavailable reason"
                        .to_string()
                })?;
                let pool = self
                    .pools
                    .get(&command.underlying)
                    .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;
                if pool.snapshot() != *pool_snapshot {
                    return Err(format!(
                        "PM settlement pool snapshot mismatch for {}",
                        command.underlying
                    ));
                }
                let config = pool.config.as_ref().ok_or_else(|| {
                    format!("missing PM settlement config {}", command.underlying)
                })?;
                classify_liquidity_gap(&command.obligation, facts, pool_snapshot, config)
                    .map_err(|error| error.to_string())?
            }
        };

        let shortfall =
            (command.settlement_obligation_usdc - command.liquid_usdc).max(Decimal::ZERO);
        let mut effects = Vec::new();
        let event = PmSettlementEventState {
            event_key: command.event_key.clone(),
            input_digest: command.facts_digest.clone(),
            output_digest: format!("{classification:?}"),
            status: format!("{classification:?}"),
            request_id: command.request_id,
            underlying: command.underlying.clone(),
            event_type: "ClassifyPmSettlement".to_string(),
            amount_usdc: shortfall,
        };

        match classification {
            PmLiquidityClassification::Paid | PmLiquidityClassification::Unavailable => {}
            PmLiquidityClassification::TimingBridge | PmLiquidityClassification::SettlementDebt => {
                if shortfall <= Decimal::ZERO {
                    return Err(
                        "PM settlement liability classification requires positive shortfall"
                            .to_string(),
                    );
                }

                let pool = self
                    .pools
                    .get_mut(&command.underlying)
                    .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;
                if pool.pool_available_usdc < shortfall {
                    return Err("PM settlement pool cannot front full shortfall".to_string());
                }
                let bridge_window_ms = pool
                    .config
                    .as_ref()
                    .ok_or_else(|| format!("missing PM settlement config {}", command.underlying))?
                    .bridge_window_ms;
                pool.pool_available_usdc -= shortfall;
                match classification {
                    PmLiquidityClassification::TimingBridge => {
                        pool.active_timing_bridge_usdc += shortfall;
                    }
                    PmLiquidityClassification::SettlementDebt => {
                        pool.active_settlement_debt_usdc += shortfall;
                    }
                    PmLiquidityClassification::Paid | PmLiquidityClassification::Unavailable => {}
                }
                pool.updated_at_ms = timestamp_ms;
                pool.recompute_utilization()?;

                let key = PmSettlementAccountKey {
                    wallet: command.wallet,
                    underlying: command.underlying.clone(),
                };
                let account = self.accounts.entry(key).or_insert_with(|| {
                    PmSettlementAccountState::clean(
                        command.wallet,
                        command.underlying.clone(),
                        command.policy_version,
                        timestamp_ms,
                    )
                });
                if account.principal_liability_usdc() == Decimal::ZERO
                    && account.accrued_interest_usdc == Decimal::ZERO
                {
                    account.last_interest_accrual_ms = timestamp_ms as i64;
                }
                match classification {
                    PmLiquidityClassification::TimingBridge => {
                        account.bridge_principal_usdc += shortfall;
                        account.bridge_deadline_ms = Some(timestamp_ms as i64 + bridge_window_ms);
                    }
                    PmLiquidityClassification::SettlementDebt => {
                        account.debt_principal_usdc += shortfall;
                    }
                    PmLiquidityClassification::Paid | PmLiquidityClassification::Unavailable => {}
                }
                account.policy_version = command.policy_version;
                account.status = account.status_after_repayment();
                account.updated_at_ms = timestamp_ms;

                effects.push(PmSettlementProjectionEffect::PoolUpsert(pool.clone()));
                effects.push(PmSettlementProjectionEffect::AccountUpsert(account.clone()));

                let should_create_debt_plan =
                    if classification == PmLiquidityClassification::SettlementDebt {
                        account
                            .active_recovery_plan_id
                            .as_deref()
                            .map(|plan_id| {
                                self.recovery_plans
                                    .get(plan_id)
                                    .map(|plan| !recovery_plan_blocks_new_plan(plan))
                                    .unwrap_or(true)
                            })
                            .unwrap_or(true)
                    } else {
                        false
                    };
                if should_create_debt_plan {
                    let plan = PmRecoveryPlanState {
                        plan_id: format!("pm-recovery-{}", command.request_id),
                        wallet: command.wallet,
                        underlying: command.underlying.clone(),
                        trigger: "settlement_debt_opened".to_string(),
                        reason: "PM settlement classified as debt".to_string(),
                        policy_version: command.policy_version,
                        recovery_priority_version: 1,
                        status: "EscalatedManual".to_string(),
                        input_digest: command.facts_digest.clone(),
                        target_reduction_usdc: shortfall,
                        expected_usdc_recovered: Decimal::ZERO,
                        expected_obligation_reduced: Decimal::ZERO,
                        expected_impact_usdc: Decimal::ZERO,
                        post_plan_utilization: pool.utilization,
                        actions: vec![PmRecoveryActionState {
                            action_index: 0,
                            action_type: "EscalateManual".to_string(),
                            target: "operator".to_string(),
                            action_payload: recovery_action_payload(
                                &PmRecoveryActionKind::EscalateManual {
                                    reason: PmRecoveryReason::SettlementDebt,
                                },
                            ),
                            status: "EscalatedManual".to_string(),
                            expected_usdc_recovered: Decimal::ZERO,
                            expected_obligation_reduced: Decimal::ZERO,
                            expected_impact_usdc: Decimal::ZERO,
                            attempt: 0,
                            submitted_external_id: None,
                            external_kind: None,
                            result: None,
                            result_external_id: None,
                            recovered_usdc: Decimal::ZERO,
                            liability_reduction_usdc: Decimal::ZERO,
                            updated_at_ms: timestamp_ms,
                        }],
                        updated_at_ms: timestamp_ms,
                    };
                    account.active_recovery_plan_id = Some(plan.plan_id.clone());
                    account.updated_at_ms = timestamp_ms;
                    self.recovery_plans
                        .insert(plan.plan_id.clone(), plan.clone());
                    effects.push(PmSettlementProjectionEffect::AccountUpsert(account.clone()));
                    effects.push(PmSettlementProjectionEffect::RecoveryPlanUpsert(plan));
                }
            }
        }

        if classification != PmLiquidityClassification::Unavailable {
            self.events.insert(command.event_key.clone(), event.clone());
        }
        effects.push(PmSettlementProjectionEffect::EventUpsert(event));
        if classification != PmLiquidityClassification::Unavailable {
            self.record_request(
                command.request_id,
                command.facts_digest,
                "ClassifyPmSettlement",
                timestamp_ms,
            );
        }
        Ok(effects)
    }

    pub fn apply_set_config(
        &mut self,
        command: SetPmSettlementPoolConfigCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "SetPmSettlementPoolConfig",
        )?;
        reject_projection_version("config_version", command.config_version)?;
        reject_projection_version("policy_version", command.config.policy_version)?;
        validate_pool_config(&command.config).map_err(|error| error.to_string())?;
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "SetPmSettlementPoolConfig",
        )? {
            return Ok(existing);
        }

        let pool = self
            .pools
            .entry(command.underlying.clone())
            .or_insert_with(|| PmSettlementPoolState::empty(command.underlying.clone()));
        if command.config_version <= pool.config_version {
            return Err(format!(
                "config_version must increase for {}: current {}, requested {}",
                command.underlying, pool.config_version, command.config_version
            ));
        }

        pool.config_version = command.config_version;
        pool.policy_version = command.config.policy_version;
        pool.config = Some(command.config);
        pool.updated_at_ms = command.timestamp_ms;
        pool.recompute_utilization()?;

        let effects = vec![PmSettlementProjectionEffect::PoolUpsert(pool.clone())];
        self.record_request(
            command.request_id,
            command.input_digest,
            "SetPmSettlementPoolConfig",
            command.timestamp_ms,
        );
        Ok(effects)
    }

    pub fn apply_record_vault_deposit(
        &mut self,
        command: RecordPmVaultDepositCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "RecordPmVaultDeposit",
        )?;
        reject_nonpositive("amount_usdc", command.amount_usdc)?;
        if command.underlying.trim().is_empty() {
            return Err("RecordPmVaultDeposit underlying must be nonempty".to_string());
        }
        let tx_hash = command.tx_hash.trim().to_ascii_lowercase();
        if tx_hash.is_empty() {
            return Err("RecordPmVaultDeposit tx_hash must be nonempty".to_string());
        }
        if command.max_listed_expiry_ts_ms <= 0 {
            return Err(
                "RecordPmVaultDeposit requires a positive max listed expiry timestamp".to_string(),
            );
        }
        if command.settlement_grace_ms < 0 {
            return Err("RecordPmVaultDeposit settlement_grace_ms must be nonnegative".to_string());
        }
        let lock_until_ms = command
            .max_listed_expiry_ts_ms
            .checked_add(command.settlement_grace_ms)
            .ok_or_else(|| "RecordPmVaultDeposit lock_until_ms overflow".to_string())?;
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "RecordPmVaultDeposit",
        )? {
            return Ok(existing);
        }
        if self.vault_deposits.contains_key(&command.request_id) {
            return Err(format!(
                "PM vault deposit {} already exists with different digest",
                command.request_id
            ));
        }
        if self.vault_deposits.values().any(|deposit| {
            deposit.chain_id == command.chain_id
                && deposit.source_contract_address == command.source_contract_address
                && deposit.tx_hash == tx_hash
                && deposit.log_index == command.log_index
        }) {
            return Err(format!(
                "PM vault deposit source event already recorded: chain {} vault {} tx {} log {}",
                command.chain_id, command.source_contract_address, tx_hash, command.log_index
            ));
        }

        let pool = self
            .pools
            .get_mut(&command.underlying)
            .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;
        pool.pool_available_usdc += command.amount_usdc;
        pool.updated_at_ms = command.timestamp_ms;
        pool.recompute_utilization()?;

        let deposit = PmVaultDepositState {
            deposit_id: command.request_id,
            depositor: command.depositor,
            underlying: command.underlying,
            principal_usdc: command.amount_usdc,
            remaining_usdc: command.amount_usdc,
            withdrawn_usdc: Decimal::ZERO,
            reserved_withdrawal_usdc: Decimal::ZERO,
            chain_id: command.chain_id,
            source_contract_address: command.source_contract_address,
            tx_hash,
            log_index: command.log_index,
            max_listed_expiry_ts_ms: command.max_listed_expiry_ts_ms,
            settlement_grace_ms: command.settlement_grace_ms,
            lock_until_ms,
            status: PmVaultDepositStatus::Active,
            created_at_ms: command.timestamp_ms,
            updated_at_ms: command.timestamp_ms,
        };
        self.vault_deposits
            .insert(deposit.deposit_id, deposit.clone());
        let effects = vec![
            PmSettlementProjectionEffect::PoolUpsert(pool.clone()),
            PmSettlementProjectionEffect::VaultDepositUpsert(deposit),
        ];
        self.record_request(
            command.request_id,
            command.input_digest,
            "RecordPmVaultDeposit",
            command.timestamp_ms,
        );
        Ok(effects)
    }

    pub fn apply_request_vault_withdrawal(
        &mut self,
        command: RequestPmVaultWithdrawalCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "RequestPmVaultWithdrawal",
        )?;
        reject_nonpositive("amount_usdc", command.amount_usdc)?;
        if command.underlying.trim().is_empty() {
            return Err("RequestPmVaultWithdrawal underlying must be nonempty".to_string());
        }
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "RequestPmVaultWithdrawal",
        )? {
            return Ok(existing);
        }
        if self.vault_withdrawals.contains_key(&command.request_id) {
            return Err(format!(
                "PM vault withdrawal {} already exists with different digest",
                command.request_id
            ));
        }

        let deposit = self
            .vault_deposits
            .get(&command.deposit_id)
            .ok_or_else(|| format!("unknown PM vault deposit {}", command.deposit_id))?;
        if deposit.depositor != command.depositor {
            return Err("PM vault withdrawal depositor does not own deposit".to_string());
        }
        if deposit.underlying != command.underlying {
            return Err("PM vault withdrawal underlying does not match deposit".to_string());
        }
        let request_ts_ms = i64::try_from(command.timestamp_ms)
            .map_err(|_| "PM vault withdrawal timestamp exceeds i64".to_string())?;
        if request_ts_ms < deposit.lock_until_ms {
            return Err(format!(
                "PM vault deposit {} is locked until {}",
                command.deposit_id, deposit.lock_until_ms
            ));
        }
        if deposit.remaining_usdc < command.amount_usdc {
            return Err("PM vault withdrawal exceeds deposit remaining amount".to_string());
        }
        let pool = self
            .pools
            .get(&command.underlying)
            .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;
        if pool.pool_available_usdc < command.amount_usdc {
            return Err("PM vault withdrawal would exceed usable pool availability".to_string());
        }

        let deposit = self
            .vault_deposits
            .get_mut(&command.deposit_id)
            .expect("deposit was validated before withdrawal mutation");
        deposit.remaining_usdc -= command.amount_usdc;
        deposit.reserved_withdrawal_usdc += command.amount_usdc;
        deposit.status = if deposit.remaining_usdc == Decimal::ZERO {
            PmVaultDepositStatus::Reserved
        } else {
            PmVaultDepositStatus::PartiallyReserved
        };
        deposit.updated_at_ms = command.timestamp_ms;

        let pool = self
            .pools
            .get_mut(&command.underlying)
            .expect("pool was validated before withdrawal mutation");
        pool.pool_available_usdc -= command.amount_usdc;
        pool.updated_at_ms = command.timestamp_ms;
        pool.recompute_utilization()?;

        let withdrawal = PmVaultWithdrawalState {
            withdrawal_id: command.request_id,
            deposit_id: command.deposit_id,
            depositor: command.depositor,
            underlying: command.underlying,
            amount_usdc: command.amount_usdc,
            lock_until_ms: deposit.lock_until_ms,
            status: PmVaultWithdrawalStatus::Reserved,
            requested_at_ms: command.timestamp_ms,
            updated_at_ms: command.timestamp_ms,
        };
        self.vault_withdrawals
            .insert(withdrawal.withdrawal_id, withdrawal.clone());
        let effects = vec![
            PmSettlementProjectionEffect::PoolUpsert(pool.clone()),
            PmSettlementProjectionEffect::VaultDepositUpsert(deposit.clone()),
            PmSettlementProjectionEffect::VaultWithdrawalUpsert(withdrawal),
        ];
        self.record_request(
            command.request_id,
            command.input_digest,
            "RequestPmVaultWithdrawal",
            command.timestamp_ms,
        );
        Ok(effects)
    }

    #[cfg(test)]
    pub fn set_test_pool_available_usdc(
        &mut self,
        underlying: String,
        amount_usdc: Decimal,
        timestamp_ms: u64,
    ) -> Result<(), String> {
        reject_negative_amount("amount_usdc", amount_usdc)?;
        let pool = self
            .pools
            .get_mut(&underlying)
            .ok_or_else(|| format!("unknown PM settlement pool {underlying}"))?;
        pool.pool_available_usdc = amount_usdc;
        pool.updated_at_ms = timestamp_ms;
        pool.recompute_utilization()
    }

    pub fn apply_accrue_interest(
        &mut self,
        command: AccruePmSettlementInterestCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "AccruePmSettlementInterest",
        )?;
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "AccruePmSettlementInterest",
        )? {
            return Ok(existing);
        }

        let key = PmSettlementAccountKey {
            wallet: command.wallet,
            underlying: command.underlying.clone(),
        };
        let account = self.accounts.get_mut(&key).ok_or_else(|| {
            format!(
                "unknown PM settlement account {}/{}",
                key.wallet, key.underlying
            )
        })?;
        let from_ms = account.last_interest_accrual_ms;
        if command.to_ms < from_ms {
            return Err(format!(
                "accrual end {} is before account cursor {}",
                command.to_ms, account.last_interest_accrual_ms
            ));
        }
        let timestamp_ms = i64::try_from(command.timestamp_ms)
            .map_err(|_| "command timestamp must fit in signed 64-bit milliseconds".to_string())?;
        if command.to_ms > timestamp_ms {
            return Err(format!(
                "accrual end {} is after command timestamp {}",
                command.to_ms, timestamp_ms
            ));
        }

        let pool = self
            .pools
            .get(&command.underlying)
            .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;
        let config = pool
            .config
            .as_ref()
            .ok_or_else(|| format!("missing PM settlement config {}", command.underlying))?;
        let utilization = pool.utilization.ok_or_else(|| {
            format!(
                "missing PM settlement pool utilization for {} - cannot accrue interest",
                command.underlying
            )
        })?;
        let apr = utilization_apr(utilization, config).map_err(|error| error.to_string())?;
        let policy_version = config.policy_version;
        reject_projection_version("policy_version", policy_version)?;

        let principal = account.principal_liability_usdc();
        let elapsed_ms = u64::try_from(command.to_ms - from_ms)
            .map_err(|_| "accrual window must be nonnegative".to_string())?;
        let interest_usdc = if principal == Decimal::ZERO || elapsed_ms == 0 {
            Decimal::ZERO
        } else {
            principal * apr * Decimal::from(elapsed_ms) / Decimal::from(INTEREST_YEAR_MS)
        };

        account.accrued_interest_usdc += interest_usdc;
        account.last_interest_accrual_ms = command.to_ms;
        account.policy_version = policy_version;
        account.updated_at_ms = command.timestamp_ms;

        let interest = PmSettlementInterestEvent {
            request_id: command.request_id,
            wallet: command.wallet,
            underlying: command.underlying,
            accrual_start_ms: from_ms,
            accrual_end_ms: command.to_ms,
            utilization,
            apr,
            interest_amount_usdc: interest_usdc,
            policy_version,
        };
        let effects = vec![
            PmSettlementProjectionEffect::InterestEvent(interest),
            PmSettlementProjectionEffect::AccountUpsert(account.clone()),
        ];
        self.record_request(
            command.request_id,
            command.input_digest,
            "AccruePmSettlementInterest",
            command.timestamp_ms,
        );
        Ok(effects)
    }

    pub fn apply_repayment(
        &mut self,
        command: ApplyPmSettlementRepaymentCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "ApplyPmSettlementRepayment",
        )?;
        reject_nonpositive("amount_usdc", command.amount_usdc)?;
        if command.reason.trim().is_empty() {
            return Err("repayment reason must be nonempty".to_string());
        }
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "ApplyPmSettlementRepayment",
        )? {
            return Ok(existing);
        }

        let key = PmSettlementAccountKey {
            wallet: command.wallet,
            underlying: command.underlying.clone(),
        };
        let account = self.accounts.get(&key).ok_or_else(|| {
            format!(
                "unknown PM settlement account {}/{}",
                key.wallet, key.underlying
            )
        })?;
        let pool = self
            .pools
            .get(&command.underlying)
            .ok_or_else(|| format!("unknown PM settlement pool {}", command.underlying))?;

        let mut remaining = command.amount_usdc;
        let interest_paid = remaining.min(account.accrued_interest_usdc);
        remaining -= interest_paid;

        let debt_paid = remaining.min(account.debt_principal_usdc);
        remaining -= debt_paid;

        let bridge_paid = remaining.min(account.bridge_principal_usdc);
        remaining -= bridge_paid;
        if remaining > Decimal::ZERO {
            return Err("repayment exceeds outstanding PM settlement liability".to_string());
        }
        if pool.active_timing_bridge_usdc < bridge_paid {
            return Err("pool active timing bridge would become negative".to_string());
        }
        if pool.active_settlement_debt_usdc < debt_paid {
            return Err("pool active settlement debt would become negative".to_string());
        }
        let active_plan_to_repay = match account.active_recovery_plan_id.as_deref() {
            Some(plan_id) => {
                let plan = self.recovery_plans.get(plan_id).ok_or_else(|| {
                    format!("active PM recovery plan {plan_id} missing during repayment resolution")
                })?;
                recovery_plan_repaid_by_repayment(
                    plan,
                    account,
                    interest_paid,
                    debt_paid,
                    bridge_paid,
                )
                .then(|| plan_id.to_string())
            }
            None => None,
        };

        let principal_paid = bridge_paid + debt_paid;
        let account = self
            .accounts
            .get_mut(&key)
            .expect("account was validated before repayment mutation");
        account.accrued_interest_usdc -= interest_paid;
        account.bridge_principal_usdc -= bridge_paid;
        account.debt_principal_usdc -= debt_paid;
        if active_plan_to_repay.is_some() {
            account.active_recovery_plan_id = None;
        }
        let pool = self
            .pools
            .get_mut(&command.underlying)
            .expect("pool was validated before repayment mutation");
        pool.pool_available_usdc += principal_paid;
        pool.active_timing_bridge_usdc -= bridge_paid;
        pool.active_settlement_debt_usdc -= debt_paid;
        pool.updated_at_ms = command.timestamp_ms;
        pool.recompute_utilization()?;

        account.status = account.status_after_repayment();
        account.updated_at_ms = command.timestamp_ms;
        let account_projection = account.clone();
        let pool_projection = pool.clone();
        let recovery_plan_effect = active_plan_to_repay.map(|plan_id| {
            let plan = self
                .recovery_plans
                .get_mut(&plan_id)
                .expect("active recovery plan was validated before repayment mutation");
            plan.status = "Repaid".to_string();
            plan.updated_at_ms = command.timestamp_ms;
            PmSettlementProjectionEffect::RecoveryPlanUpsert(plan.clone())
        });

        let repayment = PmSettlementRepaymentEvent {
            request_id: command.request_id,
            wallet: command.wallet,
            underlying: command.underlying,
            source_event_id: command.source_event_id,
            amount_usdc: command.amount_usdc,
            interest_paid_usdc: interest_paid,
            principal_paid_usdc: principal_paid,
            pool_replenished_usdc: principal_paid,
            reason: command.reason,
            event_ts_ms: command.timestamp_ms,
        };
        let mut effects = vec![
            PmSettlementProjectionEffect::RepaymentEvent(repayment),
            PmSettlementProjectionEffect::AccountUpsert(account_projection),
            PmSettlementProjectionEffect::PoolUpsert(pool_projection),
        ];
        if let Some(effect) = recovery_plan_effect {
            effects.push(effect);
        }
        self.record_request(
            command.request_id,
            command.input_digest,
            "ApplyPmSettlementRepayment",
            command.timestamp_ms,
        );
        Ok(effects)
    }

    pub fn apply_journal_recovery_plan(
        &mut self,
        command: JournalPmRecoveryPlanCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "JournalPmRecoveryPlan",
        )?;
        reject_projection_version("policy_version", command.plan.policy_version)?;
        reject_projection_version(
            "recovery_priority_version",
            command.plan.recovery_priority_version,
        )?;
        if command.plan.plan_id.trim().is_empty() {
            return Err("JournalPmRecoveryPlan plan_id must be nonempty".to_string());
        }
        if command.plan.underlying.trim().is_empty() {
            return Err("JournalPmRecoveryPlan underlying must be nonempty".to_string());
        }
        if command.plan.actions.is_empty() {
            return Err("JournalPmRecoveryPlan actions must be nonempty".to_string());
        }
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "JournalPmRecoveryPlan",
        )? {
            return Ok(existing);
        }
        if let Some(existing) = self.recovery_plans.get(&command.plan.plan_id) {
            if existing.input_digest != command.input_digest {
                return Err(format!(
                    "PM recovery plan {} digest mismatch",
                    command.plan.plan_id
                ));
            }
            if !recovery_plan_payload_matches(existing, &command.plan, &command.input_digest) {
                return Err(format!(
                    "PM recovery plan {} payload mismatch",
                    command.plan.plan_id
                ));
            }
            self.record_request(
                command.request_id,
                command.input_digest,
                "JournalPmRecoveryPlan",
                command.timestamp_ms,
            );
            return Ok(Vec::new());
        }
        if let Some(active_plan_id) = self
            .accounts
            .get(&PmSettlementAccountKey {
                wallet: command.plan.wallet,
                underlying: command.plan.underlying.clone(),
            })
            .and_then(|account| account.active_recovery_plan_id.as_deref())
        {
            if active_plan_id != command.plan.plan_id {
                let active_plan = self.recovery_plans.get(active_plan_id).ok_or_else(|| {
                    format!("active PM recovery plan {active_plan_id} is missing")
                })?;
                if recovery_plan_blocks_new_plan(active_plan) {
                    return Err(format!(
                        "PM recovery account already has active plan {active_plan_id}"
                    ));
                }
            }
        }
        if let Some(blocking_plan) = self.recovery_plans.values().find(|plan| {
            plan.wallet == command.plan.wallet
                && plan.underlying == command.plan.underlying
                && plan.plan_id != command.plan.plan_id
                && recovery_plan_blocks_new_plan(plan)
        }) {
            return Err(format!(
                "PM recovery account already has active plan {}",
                blocking_plan.plan_id
            ));
        }

        let plan_expected_usdc_recovered = command.plan.expected_usdc_recovered;
        let plan_expected_obligation_reduced = command.plan.expected_obligation_reduced;
        let plan_expected_impact_usdc = command.plan.expected_impact_usdc;
        let mut expected_index = 0u32;
        let mut actions = Vec::with_capacity(command.plan.actions.len());
        for action in command.plan.actions {
            if action.action_index != expected_index {
                return Err(format!(
                    "PM recovery action index must be contiguous: expected {}, got {}",
                    expected_index, action.action_index
                ));
            }
            validate_recovery_action_payload(&action.action)?;
            reject_negative_amount("expected_usdc_recovered", action.expected_usdc_recovered)?;
            reject_negative_amount(
                "expected_obligation_reduced",
                action.expected_obligation_reduced,
            )?;
            reject_negative_amount("expected_impact_usdc", action.expected_impact_usdc)?;
            actions.push(PmRecoveryActionState {
                action_index: action.action_index,
                action_type: recovery_action_type(&action.action).to_string(),
                target: recovery_action_target(&action.action),
                action_payload: recovery_action_payload(&action.action),
                status: recovery_initial_action_status(&action.action).to_string(),
                expected_usdc_recovered: action.expected_usdc_recovered,
                expected_obligation_reduced: action.expected_obligation_reduced,
                expected_impact_usdc: action.expected_impact_usdc,
                attempt: 0,
                submitted_external_id: None,
                external_kind: None,
                result: None,
                result_external_id: None,
                recovered_usdc: Decimal::ZERO,
                liability_reduction_usdc: Decimal::ZERO,
                updated_at_ms: command.timestamp_ms,
            });
            expected_index += 1;
        }
        reject_negative_amount("target_reduction_usdc", command.plan.target_reduction_usdc)?;
        reject_negative_amount(
            "expected_usdc_recovered",
            command.plan.expected_usdc_recovered,
        )?;
        reject_negative_amount(
            "expected_obligation_reduced",
            plan_expected_obligation_reduced,
        )?;
        reject_negative_amount("expected_impact_usdc", plan_expected_impact_usdc)?;
        validate_recovery_plan_totals(
            plan_expected_usdc_recovered,
            plan_expected_obligation_reduced,
            plan_expected_impact_usdc,
            &actions,
        )?;

        let plan_status = if actions
            .iter()
            .all(|action| action.status == "EscalatedManual")
        {
            "EscalatedManual"
        } else {
            "Planned"
        };
        let plan = PmRecoveryPlanState {
            plan_id: command.plan.plan_id,
            wallet: command.plan.wallet,
            underlying: command.plan.underlying,
            trigger: format!("{:?}", command.plan.trigger),
            reason: format!("{:?}", command.plan.reason),
            policy_version: command.plan.policy_version,
            recovery_priority_version: command.plan.recovery_priority_version,
            status: plan_status.to_string(),
            input_digest: command.input_digest.clone(),
            target_reduction_usdc: command.plan.target_reduction_usdc,
            expected_usdc_recovered: command.plan.expected_usdc_recovered,
            expected_obligation_reduced: command.plan.expected_obligation_reduced,
            expected_impact_usdc: command.plan.expected_impact_usdc,
            post_plan_utilization: command.plan.post_plan_utilization,
            actions,
            updated_at_ms: command.timestamp_ms,
        };
        let account_effect = if let Some(account) = self.accounts.get_mut(&PmSettlementAccountKey {
            wallet: plan.wallet,
            underlying: plan.underlying.clone(),
        }) {
            account.active_recovery_plan_id = Some(plan.plan_id.clone());
            account.updated_at_ms = command.timestamp_ms;
            Some(PmSettlementProjectionEffect::AccountUpsert(account.clone()))
        } else {
            None
        };
        self.recovery_plans
            .insert(plan.plan_id.clone(), plan.clone());
        self.record_request(
            command.request_id,
            command.input_digest,
            "JournalPmRecoveryPlan",
            command.timestamp_ms,
        );
        let mut effects = vec![PmSettlementProjectionEffect::RecoveryPlanUpsert(plan)];
        if let Some(effect) = account_effect {
            effects.push(effect);
        }
        Ok(effects)
    }

    pub fn apply_mark_recovery_action_submitted(
        &mut self,
        command: MarkPmRecoveryActionSubmittedCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "MarkPmRecoveryActionSubmitted",
        )?;
        if command.external_id.trim().is_empty() {
            return Err("submitted recovery external_id must be nonempty".to_string());
        }
        reject_projection_version("attempt", command.attempt)?;
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "MarkPmRecoveryActionSubmitted",
        )? {
            return Ok(existing);
        }
        let external_kind = format!("{:?}", command.external_kind);
        let plan = self
            .recovery_plans
            .get_mut(&command.plan_id)
            .ok_or_else(|| format!("unknown PM recovery plan {}", command.plan_id))?;
        if plan.wallet != command.wallet {
            return Err(format!(
                "PM recovery action wallet mismatch for plan {}",
                command.plan_id
            ));
        }
        if recovery_plan_status_is_terminal(&plan.status) {
            return Err(format!(
                "cannot submit PM recovery action for terminal plan {} status {}",
                command.plan_id, plan.status
            ));
        }
        let action = plan
            .actions
            .iter_mut()
            .find(|action| action.action_index == command.action_index)
            .ok_or_else(|| {
                format!(
                    "unknown PM recovery action {}:{}",
                    command.plan_id, command.action_index
                )
            })?;
        if action.status == "Submitted"
            && action.attempt == command.attempt
            && action.submitted_external_id.as_deref() == Some(command.external_id.as_str())
        {
            if action.external_kind.as_deref() != Some(external_kind.as_str()) {
                return Err(format!(
                    "conflicting PM recovery action submission for {}:{} attempt {}",
                    command.plan_id, command.action_index, command.attempt
                ));
            }
            self.record_request(
                command.request_id,
                command.input_digest,
                "MarkPmRecoveryActionSubmitted",
                command.timestamp_ms,
            );
            return Ok(Vec::new());
        }
        if !matches!(action.status.as_str(), "Planned" | "Failed" | "TimedOut") {
            return Err(format!(
                "cannot submit PM recovery action {}:{} from status {}",
                command.plan_id, command.action_index, action.status
            ));
        }
        if command.attempt <= action.attempt {
            return Err(format!(
                "PM recovery action attempt must increase: current {}, requested {}",
                action.attempt, command.attempt
            ));
        }
        action.status = "Submitted".to_string();
        action.attempt = command.attempt;
        action.submitted_external_id = Some(command.external_id);
        action.external_kind = Some(external_kind);
        action.result = None;
        action.result_external_id = None;
        action.recovered_usdc = Decimal::ZERO;
        action.liability_reduction_usdc = Decimal::ZERO;
        action.updated_at_ms = command.timestamp_ms;
        plan.status = "Submitted".to_string();
        plan.updated_at_ms = command.timestamp_ms;
        let effect = PmSettlementProjectionEffect::RecoveryPlanUpsert(plan.clone());
        self.record_request(
            command.request_id,
            command.input_digest,
            "MarkPmRecoveryActionSubmitted",
            command.timestamp_ms,
        );
        Ok(vec![effect])
    }

    pub fn apply_resolve_recovery_action(
        &mut self,
        command: ResolvePmRecoveryActionCommand,
    ) -> Result<Vec<PmSettlementProjectionEffect>, String> {
        validate_request_identity(
            command.request_id,
            &command.input_digest,
            "ResolvePmRecoveryAction",
        )?;
        reject_negative_amount("recovered_usdc", command.recovered_usdc)?;
        reject_negative_amount("liability_reduction_usdc", command.liability_reduction_usdc)?;
        if command.result != PmRecoveryActionResult::Confirmed
            && (command.recovered_usdc > Decimal::ZERO
                || command.liability_reduction_usdc > Decimal::ZERO)
        {
            return Err(format!(
                "{:?} PM recovery action resolution cannot carry recovered or liability reduction amounts",
                command.result
            ));
        }
        if let Some(existing) = self.idempotent_duplicate(
            command.request_id,
            &command.input_digest,
            "ResolvePmRecoveryAction",
        )? {
            return Ok(existing);
        }
        let plan = self
            .recovery_plans
            .get_mut(&command.plan_id)
            .ok_or_else(|| format!("unknown PM recovery plan {}", command.plan_id))?;
        if plan.wallet != command.wallet {
            return Err(format!(
                "PM recovery action wallet mismatch for plan {}",
                command.plan_id
            ));
        }
        let action = plan
            .actions
            .iter_mut()
            .find(|action| action.action_index == command.action_index)
            .ok_or_else(|| {
                format!(
                    "unknown PM recovery action {}:{}",
                    command.plan_id, command.action_index
                )
            })?;
        if action.result.is_some() && action.attempt == command.attempt {
            let stored_status = recovery_result_status(command.result);
            if action.result.as_deref() == Some(stored_status)
                && action.recovered_usdc == command.recovered_usdc
                && action.liability_reduction_usdc == command.liability_reduction_usdc
                && action.result_external_id == command.result_external_id
            {
                self.record_request(
                    command.request_id,
                    command.input_digest,
                    "ResolvePmRecoveryAction",
                    command.timestamp_ms,
                );
                return Ok(Vec::new());
            }
            return Err(format!(
                "conflicting PM recovery action resolution for {}:{} attempt {}",
                command.plan_id, command.action_index, command.attempt
            ));
        }
        if recovery_plan_status_is_terminal(&plan.status) {
            return Err(format!(
                "cannot resolve PM recovery action for terminal plan {} status {}",
                command.plan_id, plan.status
            ));
        }
        if action.status != "Submitted" {
            return Err(format!(
                "cannot resolve PM recovery action {}:{} from status {}",
                command.plan_id, command.action_index, action.status
            ));
        }
        if action.attempt != command.attempt {
            return Err(format!(
                "PM recovery action attempt mismatch: current {}, requested {}",
                action.attempt, command.attempt
            ));
        }
        let status = recovery_result_status(command.result).to_string();
        action.status = status.clone();
        action.result = Some(status);
        action.recovered_usdc = command.recovered_usdc;
        action.liability_reduction_usdc = command.liability_reduction_usdc;
        action.result_external_id = command.result_external_id;
        action.updated_at_ms = command.timestamp_ms;
        plan.status = recovery_plan_status_after_action_resolution(&plan.actions);
        plan.updated_at_ms = command.timestamp_ms;
        let effect = PmSettlementProjectionEffect::RecoveryPlanUpsert(plan.clone());
        let account_effect = if recovery_plan_releases_active_link(plan) {
            self.accounts
                .get_mut(&PmSettlementAccountKey {
                    wallet: plan.wallet,
                    underlying: plan.underlying.clone(),
                })
                .and_then(|account| {
                    if account.active_recovery_plan_id.as_deref() == Some(plan.plan_id.as_str()) {
                        account.active_recovery_plan_id = None;
                        account.updated_at_ms = command.timestamp_ms;
                        Some(PmSettlementProjectionEffect::AccountUpsert(account.clone()))
                    } else {
                        None
                    }
                })
        } else {
            None
        };
        self.record_request(
            command.request_id,
            command.input_digest,
            "ResolvePmRecoveryAction",
            command.timestamp_ms,
        );
        let mut effects = vec![effect];
        if let Some(effect) = account_effect {
            effects.push(effect);
        }
        Ok(effects)
    }

    fn idempotent_duplicate(
        &self,
        request_id: Uuid,
        input_digest: &str,
        command_type: &str,
    ) -> Result<Option<Vec<PmSettlementProjectionEffect>>, String> {
        let request_id = request_id.to_string();
        let Some(existing) = self.idempotency.get(&request_id) else {
            return Ok(None);
        };
        if existing.input_digest != input_digest || existing.command_type != command_type {
            return Err(format!(
                "PM settlement request idempotency mismatch for {}",
                request_id
            ));
        }
        Ok(Some(Vec::new()))
    }

    fn record_request(
        &mut self,
        request_id: Uuid,
        input_digest: String,
        command_type: &'static str,
        applied_at_ms: u64,
    ) {
        self.idempotency.insert(
            request_id.to_string(),
            PmSettlementRequestRecord {
                request_id,
                input_digest,
                output_digest: "ok".to_string(),
                command_type: command_type.to_string(),
                applied_at_ms,
            },
        );
    }
}

impl PmSettlementPoolState {
    pub fn empty(underlying: String) -> Self {
        Self {
            underlying,
            pool_available_usdc: Decimal::ZERO,
            pool_target_usdc: Decimal::ZERO,
            active_timing_bridge_usdc: Decimal::ZERO,
            active_settlement_debt_usdc: Decimal::ZERO,
            config_version: 0,
            policy_version: 0,
            config: None,
            utilization: None,
            updated_at_ms: 0,
        }
    }

    fn recompute_utilization(&mut self) -> Result<(), String> {
        let snapshot = self.snapshot();
        let capacity = pool_capacity_usdc(&snapshot).map_err(|error| error.to_string())?;
        self.utilization = if capacity == Decimal::ZERO {
            None
        } else {
            pool_utilization(&snapshot).map_err(|error| error.to_string())?
        };
        Ok(())
    }

    pub fn snapshot(&self) -> PmSettlementPoolSnapshot {
        PmSettlementPoolSnapshot {
            underlying: self.underlying.clone(),
            pool_available_usdc: self.pool_available_usdc,
            pool_target_usdc: self.pool_target_usdc,
            active_timing_bridge_usdc: self.active_timing_bridge_usdc,
            active_settlement_debt_usdc: self.active_settlement_debt_usdc,
            config_version: self.config_version,
            policy_version: self.policy_version,
        }
    }
}

impl PmSettlementAccountState {
    fn clean(
        wallet: WalletAddress,
        underlying: String,
        policy_version: u32,
        updated_at_ms: u64,
    ) -> Self {
        Self {
            wallet,
            underlying,
            bridge_principal_usdc: Decimal::ZERO,
            debt_principal_usdc: Decimal::ZERO,
            accrued_interest_usdc: Decimal::ZERO,
            last_interest_accrual_ms: updated_at_ms as i64,
            policy_version,
            bridge_deadline_ms: None,
            status: PmSettlementAccountStatus::Clean,
            active_recovery_plan_id: None,
            updated_at_ms,
        }
    }

    pub fn principal_liability_usdc(&self) -> Decimal {
        self.bridge_principal_usdc + self.debt_principal_usdc
    }

    fn status_after_repayment(&self) -> PmSettlementAccountStatus {
        if self.principal_liability_usdc() == Decimal::ZERO
            && self.accrued_interest_usdc == Decimal::ZERO
        {
            PmSettlementAccountStatus::Resolved
        } else if self.debt_principal_usdc > Decimal::ZERO {
            PmSettlementAccountStatus::Debt
        } else {
            PmSettlementAccountStatus::Bridged
        }
    }
}

fn format_event_key(key: &PmSettlementEventKey) -> String {
    format!(
        "{}:{}:{}:{}:{}",
        key.wallet, key.market_id, key.expiry_ts_ms, key.margin_mode, key.settlement_event_sequence
    )
}

fn validate_request_identity(
    request_id: Uuid,
    input_digest: &str,
    command_type: &str,
) -> Result<(), String> {
    if request_id.is_nil() {
        return Err(format!("{command_type} request_id must not be nil"));
    }
    if input_digest.trim().is_empty() {
        return Err(format!("{command_type} input_digest must be nonempty"));
    }
    Ok(())
}

fn reject_nonpositive(field: &'static str, amount: Decimal) -> Result<(), String> {
    if amount <= Decimal::ZERO {
        Err(format!("{field} must be positive"))
    } else {
        Ok(())
    }
}

fn reject_negative_amount(field: &'static str, amount: Decimal) -> Result<(), String> {
    if amount < Decimal::ZERO {
        Err(format!("{field} must be nonnegative"))
    } else {
        Ok(())
    }
}

fn recovery_action_type(action: &PmRecoveryActionKind) -> &'static str {
    match action {
        PmRecoveryActionKind::CancelOrder { .. } => "CancelOrder",
        PmRecoveryActionKind::ClosePerp { .. } => "ClosePerp",
        PmRecoveryActionKind::CloseOption { .. } => "CloseOption",
        PmRecoveryActionKind::TransferCollateral { .. } => "TransferCollateral",
        PmRecoveryActionKind::EscalateManual { .. } => "EscalateManual",
    }
}

fn recovery_action_target(action: &PmRecoveryActionKind) -> String {
    match action {
        PmRecoveryActionKind::CancelOrder { order_id, .. } => format!("order:{order_id}"),
        PmRecoveryActionKind::ClosePerp { asset, .. } => format!("perp:{asset}"),
        PmRecoveryActionKind::CloseOption { market_id, .. } => format!("option:{market_id}"),
        PmRecoveryActionKind::TransferCollateral { source, .. } => {
            format!("collateral:{source}")
        }
        PmRecoveryActionKind::EscalateManual { reason } => format!("manual:{reason:?}"),
    }
}

fn recovery_action_payload(action: &PmRecoveryActionKind) -> String {
    serde_json::to_string(action).expect("PM recovery action payload serialization must succeed")
}

fn recovery_initial_action_status(action: &PmRecoveryActionKind) -> &'static str {
    match action {
        PmRecoveryActionKind::EscalateManual { .. } => "EscalatedManual",
        _ => "Planned",
    }
}

fn validate_recovery_action_payload(action: &PmRecoveryActionKind) -> Result<(), String> {
    match action {
        PmRecoveryActionKind::CancelOrder { .. } | PmRecoveryActionKind::EscalateManual { .. } => {
            Ok(())
        }
        PmRecoveryActionKind::ClosePerp { size, .. } => reject_nonpositive("ClosePerp size", *size),
        PmRecoveryActionKind::CloseOption { size, .. } => {
            reject_nonpositive("CloseOption size", *size)
        }
        PmRecoveryActionKind::TransferCollateral { amount_usdc, .. } => {
            reject_nonpositive("TransferCollateral amount_usdc", *amount_usdc)
        }
    }
}

fn validate_recovery_plan_totals(
    plan_expected_usdc_recovered: Decimal,
    plan_expected_obligation_reduced: Decimal,
    plan_expected_impact_usdc: Decimal,
    actions: &[PmRecoveryActionState],
) -> Result<(), String> {
    let expected_usdc_recovered = actions
        .iter()
        .map(|action| action.expected_usdc_recovered)
        .sum();
    if plan_expected_usdc_recovered != expected_usdc_recovered {
        return Err(format!(
            "PM recovery plan expected_usdc_recovered {} must equal action total {}",
            plan_expected_usdc_recovered, expected_usdc_recovered
        ));
    }
    let expected_obligation_reduced = actions
        .iter()
        .map(|action| action.expected_obligation_reduced)
        .sum();
    if plan_expected_obligation_reduced != expected_obligation_reduced {
        return Err(format!(
            "PM recovery plan expected_obligation_reduced {} must equal action total {}",
            plan_expected_obligation_reduced, expected_obligation_reduced
        ));
    }
    let expected_impact_usdc = actions
        .iter()
        .map(|action| action.expected_impact_usdc)
        .sum();
    if plan_expected_impact_usdc != expected_impact_usdc {
        return Err(format!(
            "PM recovery plan expected_impact_usdc {} must equal action total {}",
            plan_expected_impact_usdc, expected_impact_usdc
        ));
    }
    Ok(())
}

fn recovery_plan_repaid_by_repayment(
    plan: &PmRecoveryPlanState,
    account: &PmSettlementAccountState,
    interest_paid: Decimal,
    debt_paid: Decimal,
    bridge_paid: Decimal,
) -> bool {
    let account_resolves = account.accrued_interest_usdc == interest_paid
        && account.debt_principal_usdc == debt_paid
        && account.bridge_principal_usdc == bridge_paid;
    if account_resolves {
        return true;
    }
    let debt_resolves =
        account.debt_principal_usdc > Decimal::ZERO && account.debt_principal_usdc == debt_paid;
    let bridge_resolves = account.bridge_principal_usdc > Decimal::ZERO
        && account.bridge_principal_usdc == bridge_paid;
    let confirmed_recovered_usdc: Decimal = plan
        .actions
        .iter()
        .filter(|action| action.status == "Confirmed")
        .map(|action| action.recovered_usdc)
        .sum();
    let confirmed_repayment_allocated = confirmed_recovered_usdc > Decimal::ZERO
        && !recovery_plan_has_progressable_actions(&plan.actions)
        && interest_paid + debt_paid + bridge_paid >= confirmed_recovered_usdc;
    if confirmed_repayment_allocated {
        return true;
    }
    match (plan.trigger.as_str(), plan.reason.as_str()) {
        ("Debt", _) | (_, "SettlementDebt") => debt_resolves,
        ("OverdueBridge", _) | (_, "TimingBridge") => bridge_resolves,
        _ => false,
    }
}

fn recovery_plan_payload_matches(
    existing: &PmRecoveryPlanState,
    incoming: &hypercall_engine::command::PmRecoveryPlanCommand,
    input_digest: &str,
) -> bool {
    existing.plan_id == incoming.plan_id
        && existing.wallet == incoming.wallet
        && existing.underlying == incoming.underlying
        && existing.trigger == format!("{:?}", incoming.trigger)
        && existing.reason == format!("{:?}", incoming.reason)
        && existing.policy_version == incoming.policy_version
        && existing.recovery_priority_version == incoming.recovery_priority_version
        && existing.input_digest == input_digest
        && existing.target_reduction_usdc == incoming.target_reduction_usdc
        && existing.expected_usdc_recovered == incoming.expected_usdc_recovered
        && existing.expected_obligation_reduced == incoming.expected_obligation_reduced
        && existing.expected_impact_usdc == incoming.expected_impact_usdc
        && existing.post_plan_utilization == incoming.post_plan_utilization
        && recovery_actions_payload_matches(&existing.actions, &incoming.actions)
}

fn recovery_actions_payload_matches(
    existing: &[PmRecoveryActionState],
    incoming: &[hypercall_engine::command::PmRecoveryActionCommand],
) -> bool {
    existing.len() == incoming.len()
        && existing
            .iter()
            .zip(incoming.iter())
            .all(|(existing, incoming)| {
                existing.action_index == incoming.action_index
                    && existing.action_type == recovery_action_type(&incoming.action)
                    && existing.target == recovery_action_target(&incoming.action)
                    && existing.action_payload == recovery_action_payload(&incoming.action)
                    && existing.expected_usdc_recovered == incoming.expected_usdc_recovered
                    && existing.expected_obligation_reduced == incoming.expected_obligation_reduced
                    && existing.expected_impact_usdc == incoming.expected_impact_usdc
            })
}

fn recovery_result_status(result: PmRecoveryActionResult) -> &'static str {
    match result {
        PmRecoveryActionResult::Confirmed => "Confirmed",
        PmRecoveryActionResult::Failed => "Failed",
        PmRecoveryActionResult::Canceled => "Canceled",
        PmRecoveryActionResult::TimedOut => "TimedOut",
    }
}

fn recovery_plan_status_after_action_resolution(actions: &[PmRecoveryActionState]) -> String {
    if actions.iter().any(|action| action.status == "Submitted") {
        return "Submitted".to_string();
    }
    if actions.iter().any(|action| action.status == "Confirmed")
        && actions.iter().any(|action| {
            matches!(
                action.status.as_str(),
                "Planned" | "Failed" | "TimedOut" | "Canceled"
            )
        })
    {
        return "PartiallyRepaid".to_string();
    }
    if actions.iter().all(|action| action.status == "Confirmed") {
        return "Confirmed".to_string();
    }
    if actions.iter().all(|action| action.status == "Canceled") {
        return "Canceled".to_string();
    }
    if actions
        .iter()
        .all(|action| matches!(action.status.as_str(), "Canceled" | "Failed" | "TimedOut"))
    {
        return "Failed".to_string();
    }
    "Planned".to_string()
}

fn recovery_plan_status_is_terminal(status: &str) -> bool {
    matches!(
        status,
        "Confirmed" | "Canceled" | "EscalatedManual" | "Repaid"
    )
}

fn recovery_plan_has_confirmed_actions(actions: &[PmRecoveryActionState]) -> bool {
    actions.iter().any(|action| action.status == "Confirmed")
}

fn recovery_plan_has_progressable_actions(actions: &[PmRecoveryActionState]) -> bool {
    actions.iter().any(|action| {
        matches!(
            action.status.as_str(),
            "Planned" | "Submitted" | "Failed" | "TimedOut"
        )
    })
}

fn recovery_plan_blocks_new_plan(plan: &PmRecoveryPlanState) -> bool {
    if plan.status == "Repaid" {
        return false;
    }
    recovery_plan_has_confirmed_actions(&plan.actions)
        || (!recovery_plan_status_is_terminal(&plan.status)
            && recovery_plan_has_progressable_actions(&plan.actions))
}

fn recovery_plan_releases_active_link(plan: &PmRecoveryPlanState) -> bool {
    !recovery_plan_has_confirmed_actions(&plan.actions)
        && (recovery_plan_status_is_terminal(&plan.status)
            || !recovery_plan_has_progressable_actions(&plan.actions))
}

fn reject_projection_version(field: &'static str, version: u32) -> Result<(), String> {
    if version == 0 {
        return Err(format!("{field} must be positive"));
    }
    if version > i32::MAX as u32 {
        return Err(format!(
            "{field} must fit in signed 32-bit projection integer"
        ));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use hypercall_engine::command::{
        AccruePmSettlementInterestCommand, ApplyPmSettlementRepaymentCommand,
        JournalPmRecoveryPlanCommand, MarkPmRecoveryActionSubmittedCommand,
        PmRecoveryActionCommand, PmRecoveryActionKind, PmRecoveryActionResult,
        PmRecoveryExternalKind, PmRecoveryPlanCommand, PmRecoveryReason, PmRecoveryTrigger,
        PmSettlementEventKey, RecordPmVaultDepositCommand, RequestPmVaultWithdrawalCommand,
        ResolvePmRecoveryActionCommand, SetPmSettlementPoolConfigCommand, TickExpiryPmSettlement,
    };
    use hypercall_margin::portfolio::{PmAccountSettlementFacts, PmSettlementObligation};
    use rust_decimal_macros::dec;
    use std::str::FromStr;

    fn wallet() -> WalletAddress {
        WalletAddress::from_str("0x1234567890123456789012345678901234567890")
            .expect("valid test wallet")
    }

    fn config() -> PmSettlementPoolConfig {
        PmSettlementPoolConfig {
            target_short_oi_notional_multiplier: dec!(0.10),
            utilization_kink: dec!(0.60),
            apr_at_kink: dec!(0.04),
            max_apr: dec!(4.00),
            normal_utilization_cap: dec!(0.80),
            crisis_utilization_cap: dec!(0.98),
            bridge_window_ms: 86_400_000,
            policy_version: 1,
        }
    }

    fn request_uuid(offset: u128) -> Uuid {
        Uuid::from_u128(0x018f_0000_0000_7000_8000_0000_0000_0000 + offset)
    }

    fn set_config(request_id: Uuid, version: u32) -> SetPmSettlementPoolConfigCommand {
        SetPmSettlementPoolConfigCommand {
            request_id,
            input_digest: format!("digest-{request_id}"),
            underlying: "BTC".to_string(),
            config_version: version,
            config: config(),
            timestamp_ms: 100,
        }
    }

    fn classify_command(
        request_id: Uuid,
        net_pnl_usdc: Decimal,
        liquid_usdc: Decimal,
        equity_usdc: Decimal,
        maintenance_usdc: Decimal,
        recoverable_usdc: Decimal,
        pool: &PmSettlementPoolState,
    ) -> TickExpiryPmSettlement {
        let event_key = PmSettlementEventKey {
            wallet: wallet(),
            market_id: "BTC-20260630-100000-C".to_string(),
            expiry_ts_ms: 1_780_200_000_000,
            margin_mode: "portfolio".to_string(),
            settlement_event_sequence: 0,
        };
        let obligation = PmSettlementObligation {
            wallet: wallet(),
            market_id: event_key.market_id.clone(),
            expiry_ts_ms: event_key.expiry_ts_ms,
            underlying: "BTC".to_string(),
            net_pnl_usdc,
            settlement_obligation_usdc: Decimal::ZERO.max(-net_pnl_usdc),
        };
        TickExpiryPmSettlement {
            request_id,
            event_key,
            wallet: wallet(),
            market_id: "BTC-20260630-100000-C".to_string(),
            underlying: "BTC".to_string(),
            expiry_ts_ms: 1_780_200_000_000,
            settlement_obligation_usdc: obligation.settlement_obligation_usdc,
            liquid_usdc,
            pm_facts: Some(PmAccountSettlementFacts {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                liquid_usdc,
                pm_equity_usdc: equity_usdc,
                pm_maintenance_requirement_usdc: maintenance_usdc,
                recoverable_collateral_usdc: recoverable_usdc,
                facts_as_of_ms: 1_780_100_000_000,
                stale: false,
            }),
            pool_snapshot: Some(pool.snapshot()),
            policy_version: pool.policy_version,
            facts_digest: format!("digest-{request_id}"),
            unavailable_reason: None,
            obligation,
        }
    }

    fn configured_state(pool_cash: Decimal) -> PmSettlementState {
        let mut state = PmSettlementState::default();
        state
            .apply_set_config(set_config(request_uuid(1), 1))
            .unwrap();
        state
            .set_test_pool_available_usdc("BTC".to_string(), pool_cash, 200)
            .unwrap();
        state
    }

    fn vault_deposit_command(
        request_id: Uuid,
        amount_usdc: Decimal,
        max_listed_expiry_ts_ms: i64,
        settlement_grace_ms: i64,
    ) -> RecordPmVaultDepositCommand {
        RecordPmVaultDepositCommand {
            request_id,
            input_digest: format!("digest-{request_id}"),
            depositor: wallet(),
            underlying: "BTC".to_string(),
            amount_usdc,
            chain_id: 42161,
            source_contract_address: wallet(),
            tx_hash: format!("0xtx{request_id}"),
            log_index: 0,
            max_listed_expiry_ts_ms,
            settlement_grace_ms,
            timestamp_ms: 1_000,
        }
    }

    #[test]
    fn pm_vault_deposit_locks_until_max_listed_expiry_plus_grace() {
        let mut state = configured_state(Decimal::ZERO);
        let request_id = request_uuid(20);
        let effects = state
            .apply_record_vault_deposit(vault_deposit_command(
                request_id,
                dec!(50_000),
                1_800_000_000_000,
                86_400_000,
            ))
            .expect("vault deposit should apply");

        let deposit = state
            .vault_deposits
            .get(&request_id)
            .expect("deposit should be durable engine state");
        assert_eq!(deposit.lock_until_ms, 1_800_086_400_000);
        assert_eq!(deposit.remaining_usdc, dec!(50_000));
        assert_eq!(
            state.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(50_000)
        );
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::VaultDepositUpsert(deposit)
                if deposit.deposit_id == request_id
        )));
    }

    #[test]
    fn pm_vault_deposit_normalizes_tx_hash_for_source_event_dedupe() {
        let mut state = configured_state(Decimal::ZERO);
        let first_id = request_uuid(121);
        let mut first = vault_deposit_command(first_id, dec!(1_000), 1_800_000_000_000, 0);
        first.tx_hash = "  0xAbCdEf  ".to_string();
        state
            .apply_record_vault_deposit(first)
            .expect("first vault deposit should apply");

        let stored = state
            .vault_deposits
            .get(&first_id)
            .expect("first deposit should be stored");
        assert_eq!(stored.tx_hash, "0xabcdef");

        let mut duplicate =
            vault_deposit_command(request_uuid(122), dec!(1_000), 1_800_000_000_000, 0);
        duplicate.tx_hash = "0xabcdef".to_string();
        let error = state
            .apply_record_vault_deposit(duplicate)
            .expect_err("same source event with normalized tx hash should reject");
        assert!(error.contains("source event already recorded"));
        assert_eq!(
            state.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(1_000)
        );
    }

    #[test]
    fn pm_vault_deposit_rejects_missing_listed_expiry() {
        let mut state = configured_state(Decimal::ZERO);
        let error = state
            .apply_record_vault_deposit(vault_deposit_command(
                request_uuid(21),
                dec!(1_000),
                0,
                86_400_000,
            ))
            .expect_err("deposit without max listed expiry should fail closed");
        assert!(error.contains("positive max listed expiry"));
    }

    #[test]
    fn pm_settlement_state_defaults_vault_maps_for_older_snapshots() {
        let state = configured_state(dec!(1_000));
        let mut value = serde_json::to_value(&state).expect("serialize settlement state");
        let object = value
            .as_object_mut()
            .expect("settlement state serializes as object");
        object.remove("vault_deposits");
        object.remove("vault_withdrawals");

        let decoded: PmSettlementState =
            serde_json::from_value(value).expect("older snapshot should deserialize");
        assert!(decoded.vault_deposits.is_empty());
        assert!(decoded.vault_withdrawals.is_empty());
        assert_eq!(
            decoded.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(1_000)
        );
    }

    #[test]
    fn pm_settlement_state_defaults_recovery_plans_for_older_msgpack_snapshots() {
        #[derive(serde::Serialize)]
        struct OlderPmSettlementState {
            pools: BTreeMap<String, PmSettlementPoolState>,
            accounts: BTreeMap<PmSettlementAccountKey, PmSettlementAccountState>,
            events: BTreeMap<PmSettlementEventKey, PmSettlementEventState>,
            idempotency: BTreeMap<String, PmSettlementRequestRecord>,
            vault_deposits: BTreeMap<Uuid, PmVaultDepositState>,
            vault_withdrawals: BTreeMap<Uuid, PmVaultWithdrawalState>,
        }

        let state = configured_state(dec!(1_000));
        let older = OlderPmSettlementState {
            pools: state.pools,
            accounts: state.accounts,
            events: state.events,
            idempotency: state.idempotency,
            vault_deposits: state.vault_deposits,
            vault_withdrawals: state.vault_withdrawals,
        };

        let payload = rmp_serde::to_vec(&older).expect("serialize older PM settlement state");
        let decoded: PmSettlementState =
            rmp_serde::from_slice(&payload).expect("older msgpack snapshot should deserialize");

        assert!(decoded.recovery_plans.is_empty());
        assert_eq!(
            decoded.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(1_000)
        );
    }

    #[test]
    fn pm_vault_withdrawal_requires_unlock_and_reserves_capacity() {
        let mut state = configured_state(Decimal::ZERO);
        let deposit_id = request_uuid(22);
        state
            .apply_record_vault_deposit(vault_deposit_command(
                deposit_id,
                dec!(10_000),
                1_800_000_000_000,
                1_000,
            ))
            .expect("vault deposit should apply");

        let locked_error = state
            .apply_request_vault_withdrawal(RequestPmVaultWithdrawalCommand {
                request_id: request_uuid(23),
                input_digest: "locked-withdrawal".to_string(),
                depositor: wallet(),
                underlying: "BTC".to_string(),
                deposit_id,
                amount_usdc: dec!(1_000),
                timestamp_ms: 1_800_000_000_999,
            })
            .expect_err("withdrawal before lock_until should reject");
        assert!(locked_error.contains("locked until"));

        let effects = state
            .apply_request_vault_withdrawal(RequestPmVaultWithdrawalCommand {
                request_id: request_uuid(24),
                input_digest: "unlocked-withdrawal".to_string(),
                depositor: wallet(),
                underlying: "BTC".to_string(),
                deposit_id,
                amount_usdc: dec!(1_000),
                timestamp_ms: 1_800_000_001_000,
            })
            .expect("unlocked withdrawal should reserve capacity");
        let deposit = state.vault_deposits.get(&deposit_id).unwrap();
        assert_eq!(deposit.remaining_usdc, dec!(9_000));
        assert_eq!(deposit.reserved_withdrawal_usdc, dec!(1_000));
        assert_eq!(
            state.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(9_000)
        );
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::VaultWithdrawalUpsert(withdrawal)
                if withdrawal.deposit_id == deposit_id
        )));
    }

    #[test]
    fn pm_settlement_event_state_rejects_missing_projection_fields() {
        let event = PmSettlementEventState {
            event_key: PmSettlementEventKey {
                wallet: wallet(),
                market_id: "BTC-20260630-100000-C".to_string(),
                expiry_ts_ms: 1_780_200_000_000,
                margin_mode: "portfolio".to_string(),
                settlement_event_sequence: 0,
            },
            input_digest: "input-digest".to_string(),
            output_digest: "output-digest".to_string(),
            status: "Paid".to_string(),
            request_id: request_uuid(10),
            underlying: "BTC".to_string(),
            event_type: "settlement_classified".to_string(),
            amount_usdc: dec!(123),
        };
        for field in ["underlying", "event_type", "amount_usdc"] {
            let mut value = serde_json::to_value(&event).expect("serialize event");
            value
                .as_object_mut()
                .expect("event serializes as object")
                .remove(field);

            let error = serde_json::from_value::<PmSettlementEventState>(value)
                .expect_err("missing persisted event field must fail deserialization");
            assert!(
                error.to_string().contains(field),
                "missing {field} should be named in error: {error}"
            );
        }
    }

    #[test]
    fn pm_recovery_plan_state_rejects_missing_identity_fields() {
        let plan = PmRecoveryPlanState {
            plan_id: "plan-1".to_string(),
            wallet: wallet(),
            underlying: "BTC".to_string(),
            trigger: "settlement_debt".to_string(),
            reason: "test".to_string(),
            policy_version: 1,
            recovery_priority_version: 1,
            status: "Pending".to_string(),
            input_digest: "digest".to_string(),
            target_reduction_usdc: dec!(1),
            expected_usdc_recovered: dec!(1),
            expected_obligation_reduced: dec!(1),
            expected_impact_usdc: Decimal::ZERO,
            post_plan_utilization: Some(dec!(0.1)),
            actions: vec![PmRecoveryActionState {
                action_index: 0,
                action_type: "notify".to_string(),
                target: "manual_recovery".to_string(),
                action_payload: "notify:manual_recovery".to_string(),
                status: "Planned".to_string(),
                expected_usdc_recovered: Decimal::ZERO,
                expected_obligation_reduced: Decimal::ZERO,
                expected_impact_usdc: Decimal::ZERO,
                attempt: 0,
                submitted_external_id: None,
                external_kind: None,
                result: None,
                result_external_id: None,
                recovered_usdc: Decimal::ZERO,
                liability_reduction_usdc: Decimal::ZERO,
                updated_at_ms: 1,
            }],
            updated_at_ms: 1,
        };
        for field in ["wallet", "underlying"] {
            let mut value = serde_json::to_value(&plan).expect("serialize recovery plan");
            value
                .as_object_mut()
                .expect("recovery plan serializes as object")
                .remove(field);

            let error = serde_json::from_value::<PmRecoveryPlanState>(value)
                .expect_err("missing persisted recovery identity field must fail deserialization");
            assert!(
                error.to_string().contains(field),
                "missing {field} should be named in error: {error}"
            );
        }
    }

    #[test]
    fn config_version_must_increase() {
        let mut state = PmSettlementState::default();
        state
            .apply_set_config(set_config(request_uuid(1), 1))
            .unwrap();

        let error = state
            .apply_set_config(set_config(request_uuid(2), 1))
            .expect_err("same config version should reject");
        assert!(error.contains("config_version must increase"));
    }

    #[test]
    fn config_versions_must_fit_projection_integer_range() {
        let mut state = PmSettlementState::default();

        let config_version_error = state
            .apply_set_config(set_config(request_uuid(1), i32::MAX as u32 + 1))
            .expect_err("oversized config version should reject before pool mutation");
        assert!(config_version_error.contains("config_version"));
        assert!(state.pools.is_empty());

        let mut oversized_policy_config = config();
        oversized_policy_config.policy_version = i32::MAX as u32 + 1;
        let policy_version_error = state
            .apply_set_config(SetPmSettlementPoolConfigCommand {
                request_id: request_uuid(2),
                input_digest: "oversized-policy-version".to_string(),
                underlying: "BTC".to_string(),
                config_version: 1,
                config: oversized_policy_config,
                timestamp_ms: 100,
            })
            .expect_err("oversized policy version should reject before pool mutation");
        assert!(policy_version_error.contains("policy_version"));
        assert!(state.pools.is_empty());
    }

    #[test]
    fn duplicate_request_with_mismatched_digest_rejects() {
        let mut state = PmSettlementState::default();
        let request_id = request_uuid(1);
        state.apply_set_config(set_config(request_id, 1)).unwrap();

        let error = state
            .apply_set_config(SetPmSettlementPoolConfigCommand {
                input_digest: "different".to_string(),
                config_version: 2,
                ..set_config(request_id, 2)
            })
            .expect_err("duplicate request with different digest should reject");

        assert!(error.contains("idempotency mismatch"));
    }

    #[test]
    fn classify_settlement_opens_timing_bridge_and_debits_pool() {
        let mut state = configured_state(dec!(10_000));
        let pool = state.pools.get("BTC").unwrap().clone();
        let command = classify_command(
            request_uuid(10),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );

        let effects = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect("bridge classification should apply");

        let pool = state.pools.get("BTC").unwrap();
        assert_eq!(pool.pool_available_usdc, dec!(9_100));
        assert_eq!(pool.active_timing_bridge_usdc, dec!(900));
        let account = state
            .accounts
            .get(&PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            })
            .unwrap();
        assert_eq!(account.bridge_principal_usdc, dec!(900));
        assert_eq!(account.status, PmSettlementAccountStatus::Bridged);
        assert_eq!(
            account.bridge_deadline_ms,
            Some(1_780_200_001_000_i64 + config().bridge_window_ms)
        );
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::EventUpsert(event)
                if event.status == "TimingBridge"
        )));
    }

    #[test]
    fn classify_settlement_debt_keeps_existing_active_recovery_plan() {
        let mut state = configured_state(dec!(10_000));
        let active_plan_id = "pm-recovery-existing-active";
        state.accounts.insert(
            PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            },
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1_000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1_000,
            },
        );
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(25),
                input_digest: "existing-plan-digest".to_string(),
                plan: recovery_plan(active_plan_id),
                timestamp_ms: 1_100,
            })
            .unwrap();
        let pool = state.pools.get("BTC").unwrap().clone();
        let command = classify_command(
            request_uuid(26),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(100),
            &pool,
        );
        let auto_plan_id = format!("pm-recovery-{}", command.request_id);

        let effects = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect("debt classification should apply");

        let account = state
            .accounts
            .get(&PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            })
            .unwrap();
        assert_eq!(
            account.active_recovery_plan_id.as_deref(),
            Some(active_plan_id)
        );
        assert!(!state.recovery_plans.contains_key(&auto_plan_id));
        assert!(!effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::RecoveryPlanUpsert(plan)
                if plan.plan_id == auto_plan_id
        )));
    }

    #[test]
    fn classify_settlement_rejects_liquid_usdc_mismatch() {
        let mut state = configured_state(dec!(10_000));
        let before = state.clone();
        let pool = state.pools.get("BTC").unwrap().clone();
        let mut command = classify_command(
            request_uuid(30),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        command
            .pm_facts
            .as_mut()
            .expect("test command includes PM facts")
            .liquid_usdc = dec!(1_000);

        let error = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect_err("liquid mismatch should reject before classification");

        assert!(error.contains("liquid_usdc mismatch"));
        assert_eq!(state, before);
    }

    #[test]
    fn classify_settlement_rejects_pm_facts_wallet_mismatch() {
        let mut state = configured_state(dec!(10_000));
        let before = state.clone();
        let pool = state.pools.get("BTC").unwrap().clone();
        let mut command = classify_command(
            request_uuid(31),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        command
            .pm_facts
            .as_mut()
            .expect("test command includes PM facts")
            .wallet = WalletAddress::from_str("0x2222222222222222222222222222222222222222")
            .expect("valid test wallet");

        let error = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect_err("wallet mismatch should reject before classification");

        assert!(error.contains("PM facts wallet mismatch"));
        assert_eq!(state, before);
    }

    #[test]
    fn classify_settlement_rejects_pm_facts_underlying_mismatch() {
        let mut state = configured_state(dec!(10_000));
        let before = state.clone();
        let pool = state.pools.get("BTC").unwrap().clone();
        let mut command = classify_command(
            request_uuid(32),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        command
            .pm_facts
            .as_mut()
            .expect("test command includes PM facts")
            .underlying = "ETH".to_string();

        let error = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect_err("underlying mismatch should reject before classification");

        assert!(error.contains("PM facts underlying mismatch"));
        assert_eq!(state, before);
    }

    #[test]
    fn classify_settlement_pool_shortfall_is_unavailable_without_mutating_liability() {
        let mut state = configured_state(dec!(500));
        let before = state.clone();
        let pool = state.pools.get("BTC").unwrap().clone();
        let command = classify_command(
            request_uuid(11),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        let event_key = command.event_key.clone();

        let effects = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .expect("shortfall should produce unavailable event");

        assert_eq!(state.pools, before.pools);
        assert!(state.accounts.is_empty());
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::EventUpsert(event)
                if event.status == "Unavailable"
        )));
        assert!(!state.events.contains_key(&event_key));
    }

    #[test]
    fn classify_settlement_unavailable_does_not_burn_retry_event_key() {
        let mut state = configured_state(dec!(500));
        let pool = state.pools.get("BTC").unwrap().clone();
        let unavailable = classify_command(
            request_uuid(12),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        let event_key = unavailable.event_key.clone();

        state
            .apply_classify_settlement(unavailable, 1_780_200_001_000)
            .expect("pool shortfall should emit unavailable event");
        assert!(!state.events.contains_key(&event_key));

        state
            .set_test_pool_available_usdc("BTC".to_string(), dec!(10_000), 200)
            .unwrap();
        let pool = state.pools.get("BTC").unwrap().clone();
        let mut retry = classify_command(
            request_uuid(13),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        retry.event_key = event_key.clone();
        retry.facts_digest = "digest-classify-after-unavailable".to_string();

        state
            .apply_classify_settlement(retry, 1_780_200_002_000)
            .expect("fresh retry should classify after unavailable event");
        assert!(state.events.contains_key(&event_key));
        assert_eq!(
            state.pools.get("BTC").unwrap().active_timing_bridge_usdc,
            dec!(900)
        );
    }

    #[test]
    fn classify_settlement_duplicate_same_digest_is_idempotent() {
        let mut state = configured_state(dec!(10_000));
        let pool = state.pools.get("BTC").unwrap().clone();
        let command = classify_command(
            request_uuid(14),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );

        state
            .apply_classify_settlement(command.clone(), 1_780_200_001_000)
            .unwrap();
        let effects = state
            .apply_classify_settlement(command, 1_780_200_001_000)
            .unwrap();

        assert!(effects.is_empty());
        assert_eq!(
            state.pools.get("BTC").unwrap().active_timing_bridge_usdc,
            dec!(900)
        );
    }

    #[test]
    fn classify_settlement_duplicate_event_key_with_changed_digest_rejects() {
        let mut state = configured_state(dec!(10_000));
        let pool = state.pools.get("BTC").unwrap().clone();
        let command = classify_command(
            request_uuid(15),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &pool,
        );
        state
            .apply_classify_settlement(command.clone(), 1_780_200_001_000)
            .unwrap();

        let mut changed = classify_command(
            request_uuid(16),
            dec!(-1_000),
            dec!(100),
            dec!(10_000),
            dec!(5_000),
            dec!(1_000),
            &state.pools.get("BTC").unwrap().clone(),
        );
        changed.event_key = command.event_key;
        changed.facts_digest = "different-digest".to_string();

        let error = state
            .apply_classify_settlement(changed, 1_780_200_001_000)
            .expect_err("same event key with changed digest must reject");
        assert!(error.contains("event digest mismatch"));
    }

    #[test]
    fn interest_rejects_backwards_cursor() {
        let mut state = configured_state(dec!(1_000));
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key,
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(100),
                debt_principal_usdc: Decimal::ZERO,
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Bridged,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );

        let error = state
            .apply_accrue_interest(AccruePmSettlementInterestCommand {
                request_id: request_uuid(3),
                input_digest: "interest-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                to_ms: 999,
                timestamp_ms: 1100,
            })
            .expect_err("backwards cursor should reject");

        assert!(error.contains("accrual end"));
    }

    #[test]
    fn interest_rejects_future_dated_window() {
        let mut state = configured_state(dec!(1_000));
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key,
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(100),
                debt_principal_usdc: Decimal::ZERO,
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Bridged,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );

        let error = state
            .apply_accrue_interest(AccruePmSettlementInterestCommand {
                request_id: request_uuid(4),
                input_digest: "future-interest-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                to_ms: 1200,
                timestamp_ms: 1100,
            })
            .expect_err("future-dated accrual window should reject");

        assert!(error.contains("after command timestamp"));
    }

    #[test]
    fn interest_derives_policy_utilization_apr_and_amount_from_pool_state() {
        let mut state = configured_state(dec!(10_000));
        {
            let pool = state.pools.get_mut("BTC").unwrap();
            pool.pool_available_usdc = dec!(7_500);
            pool.active_timing_bridge_usdc = dec!(2_500);
            pool.recompute_utilization().unwrap();
        }
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key.clone(),
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(100),
                debt_principal_usdc: Decimal::ZERO,
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 0,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Bridged,
                active_recovery_plan_id: None,
                updated_at_ms: 0,
            },
        );

        let effects = state
            .apply_accrue_interest(AccruePmSettlementInterestCommand {
                request_id: request_uuid(3),
                input_digest: "interest-derived-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                to_ms: INTEREST_YEAR_MS as i64,
                timestamp_ms: INTEREST_YEAR_MS,
            })
            .expect("engine-derived interest should accrue");

        let account = state.accounts.get(&key).unwrap();
        assert_eq!(account.policy_version, 1);
        assert_eq!(account.last_interest_accrual_ms, INTEREST_YEAR_MS as i64);
        assert_eq!(
            account.accrued_interest_usdc,
            dec!(100) * dec!(0.0166666666666666666666666667)
        );
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::InterestEvent(event)
                if event.utilization == dec!(0.25)
                    && event.apr == dec!(0.0166666666666666666666666667)
                    && event.interest_amount_usdc == account.accrued_interest_usdc
                    && event.policy_version == 1
        )));
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::AccountUpsert(account)
                if account.policy_version == 1
        )));
    }

    #[test]
    fn repayment_pays_interest_before_principal_and_replenishes_pool_for_principal() {
        let mut state = PmSettlementState::default();
        state
            .apply_set_config(set_config(request_uuid(1), 1))
            .unwrap();
        state
            .set_test_pool_available_usdc("BTC".to_string(), dec!(500), 200)
            .unwrap();
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_timing_bridge_usdc = dec!(100);
        pool.active_settlement_debt_usdc = dec!(50);
        pool.recompute_utilization().unwrap();
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key.clone(),
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(100),
                debt_principal_usdc: dec!(50),
                accrued_interest_usdc: dec!(10),
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );

        state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(3),
                input_digest: "repay-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(80),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1200,
            })
            .unwrap();

        let account = state.accounts.get(&key).unwrap();
        assert_eq!(account.accrued_interest_usdc, Decimal::ZERO);
        assert_eq!(account.bridge_principal_usdc, dec!(80));
        assert_eq!(account.debt_principal_usdc, Decimal::ZERO);
        assert_eq!(
            state.pools.get("BTC").unwrap().pool_available_usdc,
            dec!(570)
        );
    }

    #[test]
    fn repayment_rejects_overpayment_without_mutating_state() {
        let mut state = PmSettlementState::default();
        state
            .apply_set_config(set_config(request_uuid(1), 1))
            .unwrap();
        state
            .set_test_pool_available_usdc("BTC".to_string(), dec!(500), 200)
            .unwrap();
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_timing_bridge_usdc = dec!(100);
        pool.recompute_utilization().unwrap();
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key.clone(),
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(100),
                debt_principal_usdc: Decimal::ZERO,
                accrued_interest_usdc: dec!(10),
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Bridged,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );
        let before = state.clone();

        let error = state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(3),
                input_digest: "repay-too-much-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(111),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1200,
            })
            .expect_err("overpayment must reject");

        assert!(error.contains("exceeds outstanding"));
        assert_eq!(state, before);
        assert!(!state.idempotency.contains_key(&request_uuid(3).to_string()));
    }

    #[test]
    fn repayment_resolves_active_recovery_plan() {
        let mut state = configured_state(dec!(500));
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key.clone(),
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: dec!(5),
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_settlement_debt_usdc = dec!(100);
        pool.recompute_utilization().unwrap();
        let plan_id = "pm-recovery-repaid";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(20),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_100,
            })
            .unwrap();

        let effects = state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(21),
                input_digest: "repay-active-plan-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(105),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1_200,
            })
            .expect("full repayment should resolve active recovery plan");

        let account = state.accounts.get(&key).unwrap();
        assert_eq!(account.status, PmSettlementAccountStatus::Resolved);
        assert_eq!(account.active_recovery_plan_id, None);
        assert_eq!(state.recovery_plans.get(plan_id).unwrap().status, "Repaid");
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::RecoveryPlanUpsert(plan)
                if plan.plan_id == plan_id && plan.status == "Repaid"
        )));

        let before_late_submit = state.clone();
        let error = state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(22),
                input_digest: "late-submit-after-repaid-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "late-directive".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_300,
            })
            .expect_err("repaid plans must reject late submissions");

        assert!(error.contains("terminal plan"));
        assert_eq!(state, before_late_submit);
    }

    #[test]
    fn repayment_retires_debt_recovery_plan_when_bridge_remains() {
        let mut state = configured_state(dec!(500));
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key.clone(),
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: dec!(80),
                debt_principal_usdc: dec!(50),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_timing_bridge_usdc = dec!(80);
        pool.active_settlement_debt_usdc = dec!(50);
        pool.recompute_utilization().unwrap();
        let plan_id = "pm-recovery-debt-repaid";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(23),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_100,
            })
            .unwrap();

        let effects = state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(24),
                input_digest: "repay-debt-only-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(50),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1_200,
            })
            .expect("debt repayment should retire debt recovery plan");

        let account = state.accounts.get(&key).unwrap();
        assert_eq!(account.status, PmSettlementAccountStatus::Bridged);
        assert_eq!(account.debt_principal_usdc, Decimal::ZERO);
        assert_eq!(account.bridge_principal_usdc, dec!(80));
        assert_eq!(account.active_recovery_plan_id, None);
        assert_eq!(state.recovery_plans.get(plan_id).unwrap().status, "Repaid");
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::RecoveryPlanUpsert(plan)
                if plan.plan_id == plan_id && plan.status == "Repaid"
        )));
    }

    #[test]
    fn repaid_recovery_plan_rejects_late_resolution() {
        let mut state = configured_state(dec!(500));
        let key = PmSettlementAccountKey {
            wallet: wallet(),
            underlying: "BTC".to_string(),
        };
        state.accounts.insert(
            key,
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1000,
            },
        );
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_settlement_debt_usdc = dec!(100);
        pool.recompute_utilization().unwrap();
        let plan_id = "pm-recovery-late-resolution";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(27),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_100,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(28),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-late-resolution".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_150,
            })
            .unwrap();
        state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(29),
                input_digest: "repay-before-resolution-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(100),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1_200,
            })
            .unwrap();
        let before_late_resolution = state.clone();

        let error = state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(30),
                input_digest: "late-resolution-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                result: PmRecoveryActionResult::Confirmed,
                recovered_usdc: dec!(100),
                liability_reduction_usdc: dec!(100),
                result_external_id: Some("directive-late-resolution".to_string()),
                timestamp_ms: 1_300,
            })
            .expect_err("terminal repaid plan must reject late resolution");

        assert!(error.contains("terminal plan"));
        assert_eq!(state, before_late_resolution);
        assert_eq!(state.recovery_plans.get(plan_id).unwrap().status, "Repaid");
    }

    fn recovery_plan(plan_id: &str) -> PmRecoveryPlanCommand {
        PmRecoveryPlanCommand {
            plan_id: plan_id.to_string(),
            wallet: wallet(),
            underlying: "BTC".to_string(),
            trigger: PmRecoveryTrigger::Debt,
            reason: PmRecoveryReason::SettlementDebt,
            policy_version: 1,
            recovery_priority_version: 1,
            target_reduction_usdc: dec!(100),
            expected_usdc_recovered: dec!(100),
            expected_obligation_reduced: dec!(100),
            expected_impact_usdc: dec!(1),
            post_plan_utilization: Some(dec!(0.25)),
            actions: vec![PmRecoveryActionCommand {
                action_index: 0,
                action: PmRecoveryActionKind::ClosePerp {
                    asset: "BTC".to_string(),
                    size: dec!(0.1),
                    reduce_only: true,
                },
                expected_usdc_recovered: dec!(100),
                expected_obligation_reduced: dec!(100),
                expected_impact_usdc: dec!(1),
            }],
        }
    }

    #[test]
    fn duplicate_recovery_plan_rejournal_rejects_conflicting_payload() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-duplicate-payload";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(49),
                input_digest: "recovery-plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .expect("initial plan should journal");
        let before = state.clone();

        let mut conflicting_plan = recovery_plan(plan_id);
        conflicting_plan.actions[0].action = PmRecoveryActionKind::ClosePerp {
            asset: "BTC".to_string(),
            size: dec!(0.2),
            reduce_only: true,
        };
        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(50),
                input_digest: "recovery-plan-digest".to_string(),
                plan: conflicting_plan,
                timestamp_ms: 1_100,
            })
            .expect_err("same plan id and digest with different payload must reject");

        assert!(error.contains("payload mismatch"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(50).to_string()));
    }

    #[test]
    fn recovery_plan_journal_rejects_invalid_action_payload_amounts() {
        let mut state = configured_state(dec!(10_000));
        let before = state.clone();
        let mut bad_transfer = recovery_plan("pm-recovery-bad-transfer");
        bad_transfer.actions[0].action = PmRecoveryActionKind::TransferCollateral {
            source: "external".to_string(),
            amount_usdc: dec!(-1),
        };

        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(48),
                input_digest: "bad-transfer-digest".to_string(),
                plan: bad_transfer,
                timestamp_ms: 1_000,
            })
            .expect_err("negative transfer action amount must reject");

        assert!(error.contains("TransferCollateral amount_usdc"));
        assert_eq!(state, before);

        let mut bad_close = recovery_plan("pm-recovery-bad-close");
        bad_close.actions[0].action = PmRecoveryActionKind::CloseOption {
            market_id: "BTC-20260630-100000-C".to_string(),
            size: Decimal::ZERO,
            side: hypercall_types::Side::Buy,
        };
        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(47),
                input_digest: "bad-close-digest".to_string(),
                plan: bad_close,
                timestamp_ms: 1_001,
            })
            .expect_err("zero close action size must reject");

        assert!(error.contains("CloseOption size"));
        assert_eq!(state, before);
    }

    #[test]
    fn recovery_plan_journal_rejects_mismatched_plan_totals() {
        let mut state = configured_state(dec!(10_000));
        let mut plan = recovery_plan("pm-recovery-bad-total");
        plan.expected_usdc_recovered += dec!(1);
        let before = state.clone();

        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(46),
                input_digest: "bad-total-digest".to_string(),
                plan,
                timestamp_ms: 1_000,
            })
            .expect_err("plan totals must match action totals");

        assert!(error.contains("expected_usdc_recovered"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(46).to_string()));
    }

    #[test]
    fn recovery_plan_journal_rejects_second_active_plan_for_account() {
        let mut state = configured_state(dec!(10_000));
        let active_plan_id = "pm-recovery-active-plan";
        state.accounts.insert(
            PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            },
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1_000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1_000,
            },
        );
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(61),
                input_digest: "active-plan-digest".to_string(),
                plan: recovery_plan(active_plan_id),
                timestamp_ms: 1_000,
            })
            .expect("initial active plan should journal");
        let before = state.clone();

        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(62),
                input_digest: "second-plan-digest".to_string(),
                plan: recovery_plan("pm-recovery-second-plan"),
                timestamp_ms: 1_100,
            })
            .expect_err("second non-terminal plan must reject");

        assert!(error.contains("already has active plan"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(62).to_string()));
    }

    #[test]
    fn recovery_plan_journal_rejects_overlap_without_account_row() {
        let mut state = configured_state(dec!(10_000));
        let first_plan_id = "pm-recovery-no-account-first";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(75),
                input_digest: "no-account-first-digest".to_string(),
                plan: recovery_plan(first_plan_id),
                timestamp_ms: 1_000,
            })
            .expect("first no-account plan should journal");
        assert!(state
            .accounts
            .values()
            .all(|account| account.active_recovery_plan_id.as_deref() != Some(first_plan_id)));
        let before = state.clone();

        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(76),
                input_digest: "no-account-second-digest".to_string(),
                plan: recovery_plan("pm-recovery-no-account-second"),
                timestamp_ms: 1_100,
            })
            .expect_err("blocking no-account plan must reject overlap");

        assert!(error.contains("already has active plan"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(76).to_string()));
    }

    #[test]
    fn recovery_plan_journal_submit_and_resolve_updates_state_machine() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-test-1";

        let effects = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(50),
                input_digest: "recovery-plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .expect("plan should journal");
        assert!(matches!(
            effects.as_slice(),
            [PmSettlementProjectionEffect::RecoveryPlanUpsert(_)]
        ));
        let plan = state.recovery_plans.get(plan_id).unwrap();
        assert_eq!(plan.status, "Planned");
        assert_eq!(plan.actions[0].status, "Planned");

        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(51),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-1".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .expect("action should submit");
        let plan = state.recovery_plans.get(plan_id).unwrap();
        assert_eq!(plan.status, "Submitted");
        assert_eq!(plan.actions[0].attempt, 1);
        assert_eq!(
            plan.actions[0].submitted_external_id.as_deref(),
            Some("directive-1")
        );

        state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(52),
                input_digest: "resolve-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                result: PmRecoveryActionResult::Confirmed,
                recovered_usdc: dec!(75),
                liability_reduction_usdc: dec!(75),
                result_external_id: Some("directive-1".to_string()),
                timestamp_ms: 1_200,
            })
            .expect("action should resolve");
        let plan = state.recovery_plans.get(plan_id).unwrap();
        assert_eq!(plan.status, "Confirmed");
        assert_eq!(plan.actions[0].status, "Confirmed");
        assert_eq!(plan.actions[0].recovered_usdc, dec!(75));
        assert_eq!(plan.actions[0].liability_reduction_usdc, dec!(75));
    }

    #[test]
    fn recovery_duplicate_resolution_is_idempotent_without_double_counting() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-test-2";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(60),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(61),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-2".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .unwrap();
        let command = ResolvePmRecoveryActionCommand {
            request_id: request_uuid(62),
            input_digest: "resolve-digest".to_string(),
            wallet: wallet(),
            plan_id: plan_id.to_string(),
            action_index: 0,
            attempt: 1,
            result: PmRecoveryActionResult::Confirmed,
            recovered_usdc: dec!(75),
            liability_reduction_usdc: dec!(75),
            result_external_id: Some("directive-2".to_string()),
            timestamp_ms: 1_200,
        };

        state
            .apply_resolve_recovery_action(command.clone())
            .expect("first resolution applies");
        let after_first = state.clone();
        let effects = state
            .apply_resolve_recovery_action(command.clone())
            .expect("duplicate resolution is idempotent");

        assert!(effects.is_empty());
        assert_eq!(state, after_first);

        let mut conflicting = after_first.clone();
        let mut conflict_command = command.clone();
        conflict_command.request_id = request_uuid(63);
        conflict_command.input_digest = "resolve-conflict-digest".to_string();
        conflict_command.recovered_usdc = dec!(76);
        let error = conflicting
            .apply_resolve_recovery_action(conflict_command)
            .expect_err("conflicting duplicate resolution must reject");

        assert!(error.contains("conflicting PM recovery action resolution"));
        assert_eq!(conflicting, after_first);
    }

    #[test]
    fn failed_recovery_resolution_rejects_value_amounts() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-failed-with-value";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(69),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(70),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-failed-with-value".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .unwrap();
        let before = state.clone();

        let error = state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(71),
                input_digest: "failed-value-resolve-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                result: PmRecoveryActionResult::Failed,
                recovered_usdc: dec!(1),
                liability_reduction_usdc: Decimal::ZERO,
                result_external_id: Some("failure-ref".to_string()),
                timestamp_ms: 1_200,
            })
            .expect_err("failed resolution must not carry recovered value");

        assert!(error.contains("cannot carry recovered or liability reduction amounts"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(71).to_string()));
    }

    #[test]
    fn duplicate_recovery_submission_rejects_conflicting_external_kind() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-submit-kind";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(70),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(71),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "same-external-id".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .unwrap();
        let before = state.clone();

        let error = state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(72),
                input_digest: "submit-conflicting-kind-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "same-external-id".to_string(),
                external_kind: PmRecoveryExternalKind::EngineOrder,
                timestamp_ms: 1_200,
            })
            .expect_err("conflicting external kind must reject");

        assert!(error.contains("conflicting PM recovery action submission"));
        assert_eq!(state, before);
    }

    #[test]
    fn recovery_submission_rejects_projection_oversized_attempt() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-oversized-attempt";
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(73),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .unwrap();
        let before = state.clone();

        let error = state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(74),
                input_digest: "oversized-attempt-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: i32::MAX as u32 + 1,
                external_id: "oversized-attempt".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .expect_err("oversized attempt must reject before journaling");

        assert!(error.contains("attempt"));
        assert_eq!(state, before);
        assert!(!state
            .idempotency
            .contains_key(&request_uuid(74).to_string()));
    }

    #[test]
    fn failed_recovery_plan_remains_active_for_retry() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-test-failed";
        state.accounts.insert(
            PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            },
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1_000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1_000,
            },
        );
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(64),
                input_digest: "plan-digest".to_string(),
                plan: recovery_plan(plan_id),
                timestamp_ms: 1_000,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(65),
                input_digest: "submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-failed".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .unwrap();

        state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(66),
                input_digest: "resolve-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                result: PmRecoveryActionResult::Failed,
                recovered_usdc: dec!(0),
                liability_reduction_usdc: dec!(0),
                result_external_id: Some("failure-ref".to_string()),
                timestamp_ms: 1_200,
            })
            .expect("failed action should resolve and remain retryable");

        let plan = state.recovery_plans.get(plan_id).unwrap();
        assert_eq!(plan.status, "Failed");
        let account = state
            .accounts
            .get(&PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            })
            .unwrap();
        assert_eq!(account.active_recovery_plan_id.as_deref(), Some(plan_id));

        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(67),
                input_digest: "retry-submit-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 2,
                external_id: "directive-retry".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_300,
            })
            .expect("failed action should allow retry submission");
        assert_eq!(
            state.recovery_plans.get(plan_id).unwrap().actions[0].status,
            "Submitted"
        );
        let retried_action = &state.recovery_plans.get(plan_id).unwrap().actions[0];
        assert_eq!(retried_action.result, None);
        assert_eq!(retried_action.result_external_id, None);
        assert_eq!(retried_action.recovered_usdc, Decimal::ZERO);
        assert_eq!(retried_action.liability_reduction_usdc, Decimal::ZERO);

        state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(68),
                input_digest: "retry-resolve-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 2,
                result: PmRecoveryActionResult::Confirmed,
                recovered_usdc: dec!(25),
                liability_reduction_usdc: dec!(25),
                result_external_id: Some("directive-retry".to_string()),
                timestamp_ms: 1_400,
            })
            .expect("retry resolution should not conflict with previous failed result");
        assert_eq!(
            state.recovery_plans.get(plan_id).unwrap().actions[0].result,
            Some("Confirmed".to_string())
        );
    }

    #[test]
    fn confirmed_and_canceled_plan_stays_active_until_repayment() {
        let mut state = configured_state(dec!(10_000));
        let plan_id = "pm-recovery-confirmed-canceled";
        state.accounts.insert(
            PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            },
            PmSettlementAccountState {
                wallet: wallet(),
                underlying: "BTC".to_string(),
                bridge_principal_usdc: Decimal::ZERO,
                debt_principal_usdc: dec!(100),
                accrued_interest_usdc: Decimal::ZERO,
                last_interest_accrual_ms: 1_000,
                policy_version: 1,
                bridge_deadline_ms: None,
                status: PmSettlementAccountStatus::Debt,
                active_recovery_plan_id: None,
                updated_at_ms: 1_000,
            },
        );
        let pool = state.pools.get_mut("BTC").unwrap();
        pool.active_settlement_debt_usdc = dec!(100);
        pool.recompute_utilization().unwrap();
        let mut plan = recovery_plan(plan_id);
        let mut second_action = plan.actions[0].clone();
        second_action.action_index = 1;
        second_action.action = PmRecoveryActionKind::CancelOrder {
            order_id: 42,
            reason: PmRecoveryReason::SettlementDebt,
        };
        second_action.expected_usdc_recovered = Decimal::ZERO;
        second_action.expected_obligation_reduced = Decimal::ZERO;
        second_action.expected_impact_usdc = Decimal::ZERO;
        plan.actions.push(second_action);
        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(80),
                input_digest: "plan-digest".to_string(),
                plan,
                timestamp_ms: 1_000,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(81),
                input_digest: "submit-first-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                external_id: "directive-confirmed".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_100,
            })
            .unwrap();
        state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(82),
                input_digest: "resolve-first-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 0,
                attempt: 1,
                result: PmRecoveryActionResult::Confirmed,
                recovered_usdc: dec!(50),
                liability_reduction_usdc: dec!(50),
                result_external_id: Some("directive-confirmed".to_string()),
                timestamp_ms: 1_200,
            })
            .unwrap();
        state
            .apply_mark_recovery_action_submitted(MarkPmRecoveryActionSubmittedCommand {
                request_id: request_uuid(83),
                input_digest: "submit-second-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 1,
                attempt: 1,
                external_id: "directive-canceled".to_string(),
                external_kind: PmRecoveryExternalKind::Directive,
                timestamp_ms: 1_300,
            })
            .unwrap();
        let effects = state
            .apply_resolve_recovery_action(ResolvePmRecoveryActionCommand {
                request_id: request_uuid(84),
                input_digest: "resolve-second-digest".to_string(),
                wallet: wallet(),
                plan_id: plan_id.to_string(),
                action_index: 1,
                attempt: 1,
                result: PmRecoveryActionResult::Canceled,
                recovered_usdc: Decimal::ZERO,
                liability_reduction_usdc: Decimal::ZERO,
                result_external_id: Some("directive-canceled".to_string()),
                timestamp_ms: 1_400,
            })
            .unwrap();

        assert_eq!(
            state.recovery_plans.get(plan_id).unwrap().status,
            "PartiallyRepaid"
        );
        assert_eq!(
            state
                .accounts
                .get(&PmSettlementAccountKey {
                    wallet: wallet(),
                    underlying: "BTC".to_string()
                })
                .unwrap()
                .active_recovery_plan_id,
            Some(plan_id.to_string())
        );
        assert!(!effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::AccountUpsert(account)
                if account.active_recovery_plan_id.is_none()
        )));

        let before_duplicate_plan = state.clone();
        let error = state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(85),
                input_digest: "duplicate-before-repayment-digest".to_string(),
                plan: recovery_plan("pm-recovery-before-repayment"),
                timestamp_ms: 1_500,
            })
            .expect_err("confirmed unrepaid recovery plan must block duplicate planning");

        assert!(error.contains("already has active plan"));
        assert_eq!(state, before_duplicate_plan);

        let effects = state
            .apply_repayment(ApplyPmSettlementRepaymentCommand {
                request_id: request_uuid(86),
                input_digest: "partial-confirmed-repayment-digest".to_string(),
                wallet: wallet(),
                underlying: "BTC".to_string(),
                amount_usdc: dec!(50),
                reason: "admin_test".to_string(),
                source_event_id: "source-1".to_string(),
                timestamp_ms: 1_600,
            })
            .expect("confirmed cash repayment should retire the finished plan");
        let account = state
            .accounts
            .get(&PmSettlementAccountKey {
                wallet: wallet(),
                underlying: "BTC".to_string(),
            })
            .unwrap();
        assert_eq!(account.debt_principal_usdc, dec!(50));
        assert_eq!(account.active_recovery_plan_id, None);
        assert_eq!(state.recovery_plans.get(plan_id).unwrap().status, "Repaid");
        assert!(effects.iter().any(|effect| matches!(
            effect,
            PmSettlementProjectionEffect::RecoveryPlanUpsert(plan)
                if plan.plan_id == plan_id && plan.status == "Repaid"
        )));

        state
            .apply_journal_recovery_plan(JournalPmRecoveryPlanCommand {
                request_id: request_uuid(87),
                input_digest: "replacement-after-partial-repayment-digest".to_string(),
                plan: recovery_plan("pm-recovery-after-partial-repayment"),
                timestamp_ms: 1_700,
            })
            .expect("residual debt should allow a replacement recovery plan");
    }
}