// file: kb-pipeline-demo-scenarios/src/solana_memo_execution.rs
// version: 2

//! Devnet SPL Memo v4 execution with canonical post-validation.

/// Complete request for one SPL Memo v4 Devnet execution.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct DevnetMemoExecutionRequest {
    /// Stable caller-provided execution identifier.
    pub intent_id: std::string::String,
    /// Endpoint role used for cluster, balance, blockhash, fee and hydration calls.
    pub query_role: std::string::String,
    /// Endpoint role used for simulation, submission and confirmation polling.
    pub transaction_role: std::string::String,
    /// Exact UTF-8 Memo payload.
    pub message: std::string::String,
    /// Supplies the persistent Devnet fee payer as a Memo signer account.
    pub include_wallet_as_memo_signer: bool,
    /// Explicitly authorizes signing and submission after successful simulation.
    pub submit: bool,
    /// Explicit operator confirmation required by the active profile.
    pub operator_confirmed: bool,
    /// Number of `getTransaction` retries after the first hydration attempt.
    pub post_validation_max_retries: u32,
    /// Replaces existing core and decode outputs for the submitted signature.
    pub force_post_validation_replay: bool,
}

impl DevnetMemoExecutionRequest {
    /// Creates a conservative simulation-only Memo v4 request.
    pub fn new(
        intent_id: impl std::convert::Into<std::string::String>,
        message: impl std::convert::Into<std::string::String>,
    ) -> Self {
        return Self {
            intent_id: intent_id.into(),
            query_role: "http_queries".to_string(),
            transaction_role: "http_transactions".to_string(),
            message: message.into(),
            include_wallet_as_memo_signer: true,
            submit: false,
            operator_confirmed: false,
            post_validation_max_retries: 10,
            force_post_validation_replay: false,
        };
    }

    /// Validates request-local bounds independently from one profile.
    pub fn validate(&self) -> kb_core::Result<()> {
        if self.intent_id.trim().is_empty() {
            return std::result::Result::Err(kb_core::Error::config(
                "Devnet Memo execution intent id must not be empty",
            ));
        }
        if self.query_role.trim().is_empty() || self.transaction_role.trim().is_empty() {
            return std::result::Result::Err(kb_core::Error::config(
                "Devnet Memo execution endpoint roles must not be empty",
            ));
        }
        if self.message.len() > kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES {
            return std::result::Result::Err(kb_core::Error::config(format!(
                "Devnet Memo payload length {} exceeds the executor limit {}",
                self.message.len(),
                kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES
            )));
        }
        if self.post_validation_max_retries > 20 {
            return std::result::Result::Err(kb_core::Error::config(
                "post-execution getTransaction retries must not exceed 20",
            ));
        }
        return std::result::Result::Ok(());
    }
}

/// Complete result of one SPL Memo v4 Devnet execution.
#[derive(Clone, Debug, PartialEq)]
pub struct DevnetMemoExecutionSummary {
    /// Profile used by the orchestration.
    pub profile_name: std::string::String,
    /// Exact classified cluster.
    pub cluster: kb_lib::ExApiExecutionCluster,
    /// Genesis hash returned by the selected endpoint.
    pub genesis_hash: std::string::String,
    /// Non-secret persistent wallet description.
    pub wallet: kb_wallet::WalletSummary,
    /// Wallet balance observed before planning.
    pub balance_lamports: u64,
    /// Exact prepared Memo plan.
    pub plan: kb_lib::ExApiPreparedExecutionPlan,
    /// Recent blockhash used by the exact transaction.
    pub latest_blockhash: kb_onchain_transport::LatestBlockhashResult,
    /// Fee estimate for the exact compiled message.
    pub fee: kb_onchain_transport::FeeForMessageResult,
    /// Exact simulation result bound to the compiled message.
    pub simulation: kb_lib::ExApiExecutionSimulationResult,
    /// Submission result when explicitly authorized.
    pub send_result: std::option::Option<kb_lib::ExApiExecutionSendResult>,
    /// Confirmation result when submitted.
    pub confirmation: std::option::Option<kb_lib::ExApiExecutionConfirmationResult>,
    /// Canonical hydration result for the exact signature.
    pub backfill: std::option::Option<kb_pipeline::BackfillSummary>,
    /// Core extraction result for the exact signature.
    pub core_extraction: std::option::Option<kb_pipeline::CoreExtractionSummary>,
    /// First Memo decode and materialization replay.
    pub decode_replay: std::option::Option<kb_pipeline::DecodeReplaySummary>,
    /// Second replay proving that the same decoder version and input are idempotent.
    pub idempotence_replay: std::option::Option<kb_pipeline::DecodeReplaySummary>,
    /// Exact persisted transaction annotation rows for the submitted signature.
    pub annotations: std::vec::Vec<kb_store::MaterializedEventQueryRow>,
    /// Aggregated post-execution validation diagnostic.
    pub post_execution: std::option::Option<kb_lib::ExApiPostExecutionDiagnostic>,
}

/// Executes one Memo v4 Devnet simulation or explicitly authorized submission.
#[allow(clippy::too_many_arguments)]
pub async fn execute_devnet_memo<S, O>(
    http_pool: &kb_onchain_transport::HttpEndpointPool,
    store: &S,
    profile: &kb_config::ProfileConfig,
    workspace_root: &std::path::Path,
    request: &crate::DevnetMemoExecutionRequest,
    decoders: &[std::sync::Arc<dyn kb_lib::DcApiInstructionDecoder>],
    materializers: &[std::sync::Arc<dyn kb_lib::MtApiEventMaterializer>],
    observer: &O,
) -> kb_core::Result<crate::DevnetMemoExecutionSummary>
where
    S: kb_store::RawTransactionStore
        + kb_store::CoreExtractionStore
        + kb_store::DecodePipelineStore
        + Sync,
    O: crate::SolanaExecutionObserver,
{
    if let std::result::Result::Err(error) = request.validate() {
        return std::result::Result::Err(error);
    }
    if let std::result::Result::Err(error) = validate_profile(profile, request) {
        return std::result::Result::Err(error);
    }
    if let std::result::Result::Err(error) = crate::ensure_not_cancelled(observer, "validate") {
        return std::result::Result::Err(error);
    }
    let genesis = match http_pool.get_genesis_hash_for_role(request.query_role.as_str()).await {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if genesis.classified_cluster
        != std::option::Option::Some(kb_lib::ExApiExecutionCluster::Devnet)
    {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_cluster_mismatch",
            format!(
                "expected Devnet genesis hash but endpoint returned {} classified as {:?}",
                genesis.genesis_hash, genesis.classified_cluster
            ),
        ));
    }
    let wallet = match crate::load_profile_wallet(profile, workspace_root).await {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let wallet_summary = wallet.summary();
    let fee_payer = kb_lib::MdPubkey(wallet_summary.public_key.clone());
    let balance = match http_pool
        .get_balance_for_role(
            request.query_role.as_str(),
            &fee_payer,
            &kb_onchain_transport::GetBalanceConfig::confirmed(),
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if balance.lamports < profile.execution.max_fee_lamports {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_balance_insufficient",
            format!(
                "Devnet wallet balance {} is below the configured fee ceiling {}",
                balance.lamports, profile.execution.max_fee_lamports
            ),
        ));
    }
    let plan = match build_plan(profile, request, fee_payer.clone()) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let plan_evaluation = match kb_lib::ExSafetyChecker.evaluate_prepared_plan(&plan) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if plan_evaluation.decision == kb_lib::ExSafetyDecision::Deny {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_plan_denied",
            crate::violation_message(plan_evaluation.violations.as_slice()),
        ));
    }
    let latest_blockhash = match http_pool
        .get_latest_blockhash_for_role(
            request.query_role.as_str(),
            &kb_onchain_transport::GetLatestBlockhashConfig::confirmed(),
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let unsigned = match kb_lib::executor_solana_build_legacy_transaction(
        &plan,
        latest_blockhash.blockhash.as_str(),
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let message_base64 = unsigned.message_base64();
    let fee = match http_pool
        .get_fee_for_message_for_role(
            request.query_role.as_str(),
            message_base64.as_str(),
            &kb_onchain_transport::GetFeeForMessageConfig::new(
                kb_onchain_transport::RpcCommitmentLevel::Confirmed,
                std::option::Option::Some(latest_blockhash.context.slot),
            ),
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if fee.fee_lamports.is_none() {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_fee_unavailable",
            "getFeeForMessage returned null for the selected recent blockhash",
        ));
    }
    let unsigned_base64 = match unsigned.transaction_base64() {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let simulation_config = match kb_onchain_transport::SimulateTransactionConfig::new(
        kb_onchain_transport::RpcCommitmentLevel::Confirmed,
        false,
        false,
        std::option::Option::Some(latest_blockhash.context.slot),
        true,
        std::option::Option::None,
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    crate::emit(
        observer,
        crate::SolanaExecutionProgressLevel::Info,
        "memo_simulation",
        format!("simulating exact Memo message {}", unsigned.message_hash()),
        std::option::Option::None,
    );
    let simulation_rpc = match http_pool
        .simulate_transaction_for_role(
            request.transaction_role.as_str(),
            unsigned_base64.as_str(),
            &simulation_config,
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let simulation = simulation_rpc.to_execution_result(
        kb_lib::ExApiExecutionCluster::Devnet,
        kb_lib::ExApiExecutionBlockhashKind::Latest,
        std::option::Option::Some(
            simulation_rpc.context.slot.saturating_sub(latest_blockhash.context.slot),
        ),
        std::option::Option::None,
        std::option::Option::None,
        std::option::Option::Some(&fee),
    );
    let evidence = unsigned.bind_simulation(simulation.clone());
    let mut summary = crate::DevnetMemoExecutionSummary {
        profile_name: profile.name.clone(),
        cluster: kb_lib::ExApiExecutionCluster::Devnet,
        genesis_hash: genesis.genesis_hash,
        wallet: wallet_summary,
        balance_lamports: balance.lamports,
        plan,
        latest_blockhash,
        fee,
        simulation,
        send_result: std::option::Option::None,
        confirmation: std::option::Option::None,
        backfill: std::option::Option::None,
        core_extraction: std::option::Option::None,
        decode_replay: std::option::Option::None,
        idempotence_replay: std::option::Option::None,
        annotations: std::vec::Vec::new(),
        post_execution: std::option::Option::None,
    };
    if !request.submit {
        return std::result::Result::Ok(summary);
    }
    if !summary.simulation.success {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_simulation_failed",
            crate::simulation_failure_message(&summary.simulation),
        ));
    }
    let send_evaluation =
        match kb_lib::ExSafetyChecker.evaluate_send(&summary.plan, &summary.simulation) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
    if send_evaluation.decision == kb_lib::ExSafetyDecision::Deny {
        return std::result::Result::Err(kb_core::Error::new(
            "execution_send_denied",
            crate::violation_message(send_evaluation.violations.as_slice()),
        ));
    }
    let signed = match unsigned.sign_after_simulation(&evidence, &[wallet.as_signer()]) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    if let std::result::Result::Err(error) = signed.verify_signatures() {
        return std::result::Result::Err(error);
    }
    let signature = signed.primary_signature().clone();
    let mut diagnostic = kb_lib::ExApiPostExecutionDiagnostic {
        signature: signature.clone(),
        canonical_inserted: false,
        core_extracted: false,
        decode_replayed: false,
        materialized: false,
        diagnostics: std::vec::Vec::new(),
    };
    let send_config = match kb_onchain_transport::SendTransactionConfig::from_execution_config(
        &profile.execution,
        std::option::Option::Some(summary.latest_blockhash.context.slot),
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    let signed_base64 = signed.transaction_base64();
    let sent = match http_pool
        .send_transaction_for_role(
            request.transaction_role.as_str(),
            signed_base64.as_str(),
            &signature,
            &send_config,
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            diagnostic.diagnostics.push(format!("Memo submission failed: {error}"));
            summary.post_execution = std::option::Option::Some(diagnostic);
            return std::result::Result::Ok(summary);
        },
    };
    summary.send_result =
        std::option::Option::Some(sent.to_execution_result(kb_lib::ExApiExecutionCluster::Devnet));
    let confirmation_config =
        match kb_onchain_transport::ConfirmTransactionConfig::from_execution_config(
            &profile.execution,
            std::option::Option::Some(summary.latest_blockhash.last_valid_block_height),
            std::option::Option::Some(summary.latest_blockhash.context.slot),
        ) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
    let confirmation = match http_pool
        .confirm_transaction_for_roles(
            request.transaction_role.as_str(),
            request.query_role.as_str(),
            kb_lib::ExApiExecutionCluster::Devnet,
            &signature,
            &confirmation_config,
        )
        .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            diagnostic.diagnostics.push(format!("Memo confirmation failed: {error}"));
            summary.post_execution = std::option::Option::Some(diagnostic);
            return std::result::Result::Ok(summary);
        },
    };
    let confirmation_status = confirmation.status;
    summary.confirmation = std::option::Option::Some(confirmation);
    if !matches!(
        confirmation_status,
        kb_lib::ExApiExecutionConfirmationStatus::Confirmed
            | kb_lib::ExApiExecutionConfirmationStatus::Finalized
    ) {
        diagnostic.diagnostics.push(format!(
            "Memo post-validation stopped at confirmation status {confirmation_status:?}"
        ));
        summary.post_execution = std::option::Option::Some(diagnostic);
        return std::result::Result::Ok(summary);
    }
    let backfill =
        match hydrate_signature(http_pool, store, profile, request, observer, &signature).await {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                diagnostic.diagnostics.push(format!("Memo hydration failed: {error}"));
                summary.post_execution = std::option::Option::Some(diagnostic);
                return std::result::Result::Ok(summary);
            },
        };
    diagnostic.canonical_inserted = canonical_available(&backfill);
    summary.backfill = std::option::Option::Some(backfill);
    if !diagnostic.canonical_inserted {
        diagnostic
            .diagnostics
            .push("confirmed Memo transaction was unavailable for canonical hydration".to_string());
        summary.post_execution = std::option::Option::Some(diagnostic);
        return std::result::Result::Ok(summary);
    }
    let extraction = match kb_pipeline::execute_core_extraction(
        store,
        &kb_pipeline::CoreExtractionRequest {
            source: kb_pipeline::CoreExtractionSource::Signatures(std::vec![signature.0.clone()]),
            limit: 1,
            max_concurrent_extractions: 1,
            force_replay: request.force_post_validation_replay,
        },
        observer,
    )
    .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            diagnostic.diagnostics.push(format!("Memo core extraction failed: {error}"));
            summary.post_execution = std::option::Option::Some(diagnostic);
            return std::result::Result::Ok(summary);
        },
    };
    diagnostic.core_extracted = extraction.failed == 0
        && !extraction.cancelled
        && extraction.selected == 1
        && extraction.extracted.saturating_add(extraction.skipped) >= 1;
    summary.core_extraction = std::option::Option::Some(extraction);
    if !diagnostic.core_extracted {
        diagnostic.diagnostics.push("Memo core extraction did not complete".to_string());
        summary.post_execution = std::option::Option::Some(diagnostic);
        return std::result::Result::Ok(summary);
    }
    let first_replay =
        match replay_memo(store, request, &signature, false, decoders, materializers, observer)
            .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                diagnostic.diagnostics.push(format!("Memo decode replay failed: {error}"));
                summary.post_execution = std::option::Option::Some(diagnostic);
                return std::result::Result::Ok(summary);
            },
        };
    diagnostic.decode_replayed = decode_completed(&first_replay);
    summary.decode_replay = std::option::Option::Some(first_replay);
    let filter = match kb_store::MaterializedEventFilter::new(
        std::option::Option::Some("transaction_annotations".to_string()),
        std::option::Option::Some("transaction_annotation".to_string()),
        std::option::Option::Some(signature.0.clone()),
        8,
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    summary.annotations =
        match kb_store::DecodePipelineStore::list_materialized_events(store, &filter).await {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                diagnostic.diagnostics.push(format!("Memo annotation query failed: {error}"));
                summary.post_execution = std::option::Option::Some(diagnostic);
                return std::result::Result::Ok(summary);
            },
        };
    diagnostic.materialized = diagnostic.decode_replayed
        && summary
            .annotations
            .iter()
            .any(|row| return row.signature.as_str() == signature.0.as_str());
    let second_replay = match replay_memo(
        store,
        request,
        &signature,
        true,
        decoders,
        materializers,
        observer,
    )
    .await
    {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => {
            diagnostic.diagnostics.push(format!("Memo idempotence replay failed: {error}"));
            summary.post_execution = std::option::Option::Some(diagnostic);
            return std::result::Result::Ok(summary);
        },
    };
    let idempotent = second_replay.failed_inputs == 0
        && second_replay.processors.iter().all(|processor| {
            return processor.failed == 0
                && processor.materialized_outputs == 0
                && processor.materialization_refused == 0;
        });
    summary.idempotence_replay = std::option::Option::Some(second_replay);
    if !idempotent {
        diagnostic
            .diagnostics
            .push("second Memo replay did not prove a clean idempotent skip".to_string());
    } else if diagnostic.canonical_inserted
        && diagnostic.core_extracted
        && diagnostic.decode_replayed
        && diagnostic.materialized
    {
        diagnostic.diagnostics.push(
            "Memo completed canonical hydration, core extraction, decode, annotation projection and idempotence validation"
                .to_string(),
        );
    }
    summary.post_execution = std::option::Option::Some(diagnostic);
    return std::result::Result::Ok(summary);
}

fn validate_profile(
    profile: &kb_config::ProfileConfig,
    request: &crate::DevnetMemoExecutionRequest,
) -> kb_core::Result<()> {
    if profile.wallet.cluster != "devnet" {
        return std::result::Result::Err(kb_core::Error::config(
            "Memo Devnet orchestration requires a Devnet wallet profile",
        ));
    }
    if !profile.wallet.temporary_wallet_enabled || !profile.wallet.temporary_wallet_persist {
        return std::result::Result::Err(kb_core::Error::config(
            "Memo Devnet orchestration requires an enabled persistent temporary wallet",
        ));
    }
    if !profile.execution.require_simulation {
        return std::result::Result::Err(kb_core::Error::config(
            "Memo Devnet orchestration requires simulation",
        ));
    }
    if request.submit && !profile.wallet.devnet_send_enabled {
        return std::result::Result::Err(kb_core::Error::config(
            "Devnet transaction submission is disabled by the wallet profile",
        ));
    }
    if request.submit
        && profile.execution.require_operator_confirmation
        && !request.operator_confirmed
    {
        return std::result::Result::Err(kb_core::Error::config(
            "Memo Devnet submission requires explicit operator confirmation",
        ));
    }
    return std::result::Result::Ok(());
}

fn build_plan(
    profile: &kb_config::ProfileConfig,
    request: &crate::DevnetMemoExecutionRequest,
    fee_payer: kb_lib::MdPubkey,
) -> kb_core::Result<kb_lib::ExApiPreparedExecutionPlan> {
    let signers = if request.include_wallet_as_memo_signer {
        std::vec![kb_lib::ExSplMemoSigner { pubkey: fee_payer.clone() }]
    } else {
        std::vec::Vec::new()
    };
    let intent = kb_lib::ExSplMemoExecutionIntent {
        intent_id: request.intent_id.clone(),
        fee_payer: fee_payer.clone(),
        policy: kb_lib::ExApiExecutionPolicy {
            cluster: kb_lib::ExApiExecutionClusterPolicy {
                expected_cluster: kb_lib::ExApiExecutionCluster::Devnet,
                allow_mainnet: false,
                mainnet_confirmation: false,
            },
            simulation: kb_lib::ExApiExecutionSimulationPolicy::Required,
            blockhash: kb_lib::ExApiExecutionBlockhashPolicy {
                kind: kb_lib::ExApiExecutionBlockhashKind::Latest,
                max_age_slots: std::option::Option::Some(
                    profile.execution.recent_blockhash_max_age_slots,
                ),
                nonce_account: std::option::Option::None,
                nonce_authority: std::option::Option::None,
            },
            cost_limit: kb_lib::ExApiExecutionCostLimit {
                max_spend_lamports: std::option::Option::Some(0),
                max_fee_lamports: std::option::Option::Some(profile.execution.max_fee_lamports),
                max_compute_unit_price_micro_lamports: std::option::Option::Some(
                    profile.execution.max_compute_unit_price_micro_lamports,
                ),
            },
            authorized_signers: std::vec![fee_payer.clone()],
            dry_run: !request.submit,
            post_execution_validation: kb_lib::ExApiPostExecutionValidationPolicy {
                canonical_insert_required: true,
                core_extraction_required: true,
                decode_replay_required: true,
                materialization_required: true,
            },
        },
        operation: kb_lib::ExSplMemoOperation::AddMemo {
            generation: kb_lib::ExSplMemoGeneration::V4,
            message: request.message.clone(),
            signers,
        },
    };
    return kb_lib::ExApiTypedInstructionExecutor::build_prepared_plan(
        &kb_lib::ExSplMemoExecutor,
        &intent,
    );
}

async fn hydrate_signature<S, O>(
    http_pool: &kb_onchain_transport::HttpEndpointPool,
    store: &S,
    profile: &kb_config::ProfileConfig,
    request: &crate::DevnetMemoExecutionRequest,
    observer: &O,
    signature: &kb_lib::MdSignature,
) -> kb_core::Result<kb_pipeline::BackfillSummary>
where
    S: kb_store::RawTransactionStore + Sync,
    O: crate::SolanaExecutionObserver,
{
    let mut retry = 0_u32;
    loop {
        let result = match kb_pipeline::execute_http_backfill(
            http_pool,
            store,
            &kb_pipeline::BackfillRequest {
                role: request.query_role.clone(),
                commitment: "confirmed".to_string(),
                source: kb_pipeline::BackfillSource::ExplicitSignatures(std::vec![
                    signature.0.clone()
                ]),
                page_size: 1,
                max_pages: 1,
                max_concurrent_requests: 1,
                max_retries: 0,
            },
            observer,
        )
        .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => return std::result::Result::Err(error),
        };
        if canonical_available(&result)
            || retry >= request.post_validation_max_retries
            || observer.is_execution_cancelled()
        {
            return std::result::Result::Ok(result);
        }
        retry = retry.saturating_add(1);
        tokio::time::sleep(std::time::Duration::from_millis(std::cmp::max(
            profile.execution.confirmation_poll_interval_ms,
            500,
        )))
        .await;
    }
}

fn canonical_available(summary: &kb_pipeline::BackfillSummary) -> bool {
    return summary.failed == 0
        && summary.missing == 0
        && summary.candidates_completed == 1
        && summary.candidates_cancelled == 0
        && summary.candidates_not_started == 0
        && summary
            .canonical_inserted
            .saturating_add(summary.canonical_skipped)
            .saturating_add(summary.existing_skipped)
            >= 1;
}

async fn replay_memo<S, O>(
    store: &S,
    request: &crate::DevnetMemoExecutionRequest,
    signature: &kb_lib::MdSignature,
    include_materialized_state: bool,
    decoders: &[std::sync::Arc<dyn kb_lib::DcApiInstructionDecoder>],
    materializers: &[std::sync::Arc<dyn kb_lib::MtApiEventMaterializer>],
    observer: &O,
) -> kb_core::Result<kb_pipeline::DecodeReplaySummary>
where
    S: kb_store::DecodePipelineStore + Sync,
    O: crate::SolanaExecutionObserver,
{
    let mut states = std::vec![
        kb_store::CoreInstructionProcessingState::Pending,
        kb_store::CoreInstructionProcessingState::Failed,
        kb_store::CoreInstructionProcessingState::ReplayRequested,
    ];
    if include_materialized_state {
        states.push(kb_store::CoreInstructionProcessingState::Materialized);
    }
    let selection = match kb_store::DecodeSelectionFilter::new(
        std::vec![signature.0.clone()],
        states,
        std::option::Option::None,
        std::option::Option::None,
        std::vec![kb_program_ids::SPL_MEMO_V4_PROGRAM_ID.to_string()],
        std::vec::Vec::new(),
        false,
        8,
    ) {
        std::result::Result::Ok(value) => value,
        std::result::Result::Err(error) => return std::result::Result::Err(error),
    };
    return kb_pipeline::execute_decode_replay(
        store,
        &kb_pipeline::DecodeReplayRequest {
            campaign_id: kb_pipeline::new_decode_campaign_id(),
            selection,
            decoder_names: std::vec::Vec::new(),
            dispatch_policy: kb_pipeline::DecodeDispatchPolicy::HighestPriority,
            max_concurrent_inputs: 1,
            force_replay: if include_materialized_state {
                false
            } else {
                request.force_post_validation_replay
            },
            force_replay_all_matching: false,
            materialize_after_decode: true,
        },
        decoders,
        materializers,
        observer,
    )
    .await;
}

fn decode_completed(summary: &kb_pipeline::DecodeReplaySummary) -> bool {
    return summary.failed_inputs == 0
        && summary.unmatched == 0
        && !summary.cancelled
        && summary.completed >= 1
        && summary.processors.iter().all(|processor| {
            return processor.failed == 0
                && processor.unsupported == 0
                && processor.materialization_refused == 0;
        })
        && summary.processors.iter().map(|processor| return processor.decoded).sum::<u64>() >= 1;
}

#[cfg(test)]
mod tests {
    fn local_devnet_profile() -> kb_config::ProfileConfig {
        let config =
            match kb_config::parse_config_json(include_str!("../../config/example.config.json")) {
                std::result::Result::Ok(value) => value,
                std::result::Result::Err(error) => panic!("example config parse failed: {error}"),
            };
        for profile in config.profiles {
            if profile.name == "local_devnet" {
                return profile;
            }
        }
        panic!("local_devnet profile missing");
    }

    #[test]
    fn request_is_simulation_only_and_bounded_by_default() {
        let request = crate::DevnetMemoExecutionRequest::new("memo-1", "audit annotation");
        assert!(!request.submit);
        assert!(request.include_wallet_as_memo_signer);
        assert!(request.validate().is_ok());
        let oversized = "x".repeat(kb_lib::EX_SPL_MEMO_MAX_MESSAGE_BYTES + 1);
        assert!(crate::DevnetMemoExecutionRequest::new("memo-2", oversized).validate().is_err());
    }

    #[test]
    fn exact_v4_plan_has_zero_spend_and_wallet_signer() {
        let profile = local_devnet_profile();
        let fee_payer = kb_lib::MdPubkey(kb_program_ids::SYSTEM_PROGRAM_ID.to_string());
        let mut request = crate::DevnetMemoExecutionRequest::new("memo-3", "hello");
        request.submit = true;
        request.operator_confirmed = true;
        let plan = match super::build_plan(&profile, &request, fee_payer.clone()) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("Memo plan failed: {error}"),
        };
        assert_eq!(plan.requested_spend_lamports, 0);
        assert_eq!(plan.fee_payer, fee_payer);
        assert_eq!(plan.instructions.len(), 1);
        assert_eq!(plan.instructions[0].program_id.0, kb_program_ids::SPL_MEMO_V4_PROGRAM_ID);
        assert_eq!(plan.instructions[0].accounts.len(), 1);
        assert!(plan.instructions[0].accounts[0].is_signer);
        assert!(!plan.instructions[0].accounts[0].is_writable);
        assert!(kb_lib::ExSafetyChecker.evaluate_prepared_plan(&plan).is_ok());
    }

    #[test]
    fn submission_requires_profile_enablement_and_confirmation() {
        let profile = local_devnet_profile();
        let mut request = crate::DevnetMemoExecutionRequest::new("memo-4", "hello");
        assert!(super::validate_profile(&profile, &request).is_ok());
        request.submit = true;
        assert!(super::validate_profile(&profile, &request).is_err());
        request.operator_confirmed = true;
        assert!(super::validate_profile(&profile, &request).is_ok());
    }

    #[tokio::test]
    async fn optional_devnet_memo_execution_from_env() {
        if std::env::var("KB_DEVNET_MEMO_EXECUTION_TEST").ok().as_deref()
            != std::option::Option::Some("1")
        {
            return;
        }
        let database_url = match std::env::var("KB_POSTGRES_TEST_URL") {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => {
                panic!("KB_POSTGRES_TEST_URL is required: {error}");
            },
        };
        let mut profile = local_devnet_profile();
        profile.database.backend = "postgres".to_string();
        profile.database.postgres.url = database_url;
        if let std::result::Result::Ok(directory) = std::env::var("KB_DEVNET_WALLET_DIR") {
            profile.wallet.wallet_dir = directory;
        }
        let pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&profile) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("HTTP pool creation failed: {error}"),
        };
        let store_options = match kb_store::PostgresStoreOptions::new(
            profile.database.postgres.url.clone(),
            profile.database.postgres.max_connections,
            profile.database.postgres.connect_timeout_ms,
            false,
        ) {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("PostgreSQL options failed: {error}"),
        };
        let store = match kb_store::PostgresStore::connect(store_options).await {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("PostgreSQL connection failed: {error}"),
        };
        if let std::result::Result::Err(error) = store.initialize_store_schema().await {
            panic!("PostgreSQL schema initialization failed: {error}");
        }
        let mut request = crate::DevnetMemoExecutionRequest::new(
            format!("devnet-memo-test-{}", uuid::Uuid::new_v4()),
            format!("khadhroony-bot3 memo validation {}", uuid::Uuid::new_v4()),
        );
        request.post_validation_max_retries = 20;
        if std::env::var("KB_DEVNET_MEMO_SUBMIT").ok().as_deref() == std::option::Option::Some("1")
        {
            request.submit = true;
            request.operator_confirmed = true;
        }
        let decoders: std::vec::Vec<std::sync::Arc<dyn kb_lib::DcApiInstructionDecoder>> =
            std::vec![std::sync::Arc::new(kb_lib::DcSplMemoDecoder)];
        let materializers: std::vec::Vec<std::sync::Arc<dyn kb_lib::MtApiEventMaterializer>> =
            std::vec![std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,)];
        let workspace_root = match std::path::Path::new(env!("CARGO_MANIFEST_DIR")).parent() {
            std::option::Option::Some(value) => value,
            std::option::Option::None => panic!("workspace root cannot be resolved"),
        };
        let summary = match crate::execute_devnet_memo(
            &pool,
            &store,
            &profile,
            workspace_root,
            &request,
            decoders.as_slice(),
            materializers.as_slice(),
            &crate::NoopSolanaExecutionObserver,
        )
        .await
        {
            std::result::Result::Ok(value) => value,
            std::result::Result::Err(error) => panic!("Devnet Memo execution failed: {error}"),
        };
        assert!(summary.simulation.success);
        if request.submit {
            let post_execution = match summary.post_execution {
                std::option::Option::Some(value) => value,
                std::option::Option::None => panic!("Memo post-execution diagnostic missing"),
            };
            assert!(post_execution.canonical_inserted);
            assert!(post_execution.core_extracted);
            assert!(post_execution.decode_replayed);
            assert!(post_execution.materialized);
            assert!(!summary.annotations.is_empty());
            let idempotence = match summary.idempotence_replay {
                std::option::Option::Some(value) => value,
                std::option::Option::None => panic!("Memo idempotence replay missing"),
            };
            assert_eq!(
                idempotence
                    .processors
                    .iter()
                    .map(|processor| return processor.materialized_outputs)
                    .sum::<u64>(),
                0
            );
        }
    }
}