v0.1.0-pre.052

kb-app-demo-desktop/src/app_state.rs

@@ -1,5 +1,5 @@
 // file: kb-app-demo-desktop/src/app_state.rs
-// version: 4
+// version: 5
 
 //! Shared Tauri application state and startup initialization.
 
@@ -10,10 +10,17 @@ pub(crate) struct AppState {
     active_profile: kb_config::ProfileConfig,
     logging_guard: std::sync::Mutex<kb_logging::LoggingGuard>,
     http_pool: kb_onchain_transport::HttpEndpointPool,
-    ws_pool: std::sync::Mutex<std::option::Option<std::sync::Arc<kb_onchain_transport::WsEndpointPool>>>,
-    demo_ws_session: tokio::sync::Mutex<std::option::Option<std::sync::Arc<kb_onchain_transport::WsSession>>>,
+    ws_pool:
+        std::sync::Mutex<std::option::Option<std::sync::Arc<kb_onchain_transport::WsEndpointPool>>>,
+    demo_ws_session:
+        tokio::sync::Mutex<std::option::Option<std::sync::Arc<kb_onchain_transport::WsSession>>>,
     demo_backfill_running: std::sync::atomic::AtomicBool,
     demo_backfill_cancel_requested: std::sync::atomic::AtomicBool,
+    demo_core_extraction_running: std::sync::atomic::AtomicBool,
+    demo_core_extraction_cancel_requested: std::sync::atomic::AtomicBool,
+    demo_decode_replay_running: std::sync::atomic::AtomicBool,
+    demo_decode_replay_cancel_requested: std::sync::atomic::AtomicBool,
+    demo_decode_replay_campaign_id: std::sync::Mutex<std::option::Option<std::string::String>>,
 }
 
 impl crate::AppState {
@@ -33,7 +40,8 @@ impl crate::AppState {
             std::result::Result::Ok(guard) => guard,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
-        let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile) {
+        let http_pool = match kb_onchain_transport::HttpEndpointPool::from_profile(&active_profile)
+        {
             std::result::Result::Ok(pool) => pool,
             std::result::Result::Err(error) => return std::result::Result::Err(error),
         };
@@ -47,6 +55,11 @@ impl crate::AppState {
             demo_ws_session: tokio::sync::Mutex::new(std::option::Option::None),
             demo_backfill_running: std::sync::atomic::AtomicBool::new(false),
             demo_backfill_cancel_requested: std::sync::atomic::AtomicBool::new(false),
+            demo_core_extraction_running: std::sync::atomic::AtomicBool::new(false),
+            demo_core_extraction_cancel_requested: std::sync::atomic::AtomicBool::new(false),
+            demo_decode_replay_running: std::sync::atomic::AtomicBool::new(false),
+            demo_decode_replay_cancel_requested: std::sync::atomic::AtomicBool::new(false),
+            demo_decode_replay_campaign_id: std::sync::Mutex::new(std::option::Option::None),
         });
     }
 
@@ -97,9 +110,8 @@ impl crate::AppState {
     /// Returns the persistent WebSocket demo session slot.
     pub(crate) fn demo_ws_session(
         &self,
-    ) -> &tokio::sync::Mutex<
-        std::option::Option<std::sync::Arc<kb_onchain_transport::WsSession>>,
-    > {
+    ) -> &tokio::sync::Mutex<std::option::Option<std::sync::Arc<kb_onchain_transport::WsSession>>>
+    {
         return &self.demo_ws_session;
     }
 
@@ -113,6 +125,33 @@ impl crate::AppState {
         return &self.demo_backfill_cancel_requested;
     }
 
+    /// Returns the core extraction running flag.
+    pub(crate) fn demo_core_extraction_running(&self) -> &std::sync::atomic::AtomicBool {
+        return &self.demo_core_extraction_running;
+    }
+
+    /// Returns the core extraction cancellation flag.
+    pub(crate) fn demo_core_extraction_cancel_requested(&self) -> &std::sync::atomic::AtomicBool {
+        return &self.demo_core_extraction_cancel_requested;
+    }
+
+    /// Returns the decode replay running flag.
+    pub(crate) fn demo_decode_replay_running(&self) -> &std::sync::atomic::AtomicBool {
+        return &self.demo_decode_replay_running;
+    }
+
+    /// Returns the decode replay cancellation flag.
+    pub(crate) fn demo_decode_replay_cancel_requested(&self) -> &std::sync::atomic::AtomicBool {
+        return &self.demo_decode_replay_cancel_requested;
+    }
+
+    /// Returns the active decode replay campaign slot.
+    pub(crate) fn demo_decode_replay_campaign_id(
+        &self,
+    ) -> &std::sync::Mutex<std::option::Option<std::string::String>> {
+        return &self.demo_decode_replay_campaign_id;
+    }
+
     /// Returns the number of logging routes held by the logging guard.
     pub(crate) fn logging_route_count(&self) -> usize {
         let lock_result = self.logging_guard.lock();

kb-app-demo-desktop/src/demo_config.rs

@@ -1,5 +1,5 @@
 // file: kb-app-demo-desktop/src/demo_config.rs
-// version: 1
+// version: 2
 
 //! Configuration demo payload and state projection.
 

kb-app-demo-desktop/src/demo_core_extraction.rs

@@ -0,0 +1,267 @@
+// file: kb-app-demo-desktop/src/demo_core_extraction.rs
+// version: 9
+
+//! Tauri commands and UI payloads for canonical transaction to core extraction.
+
+use tauri::Emitter; // rust-rules: trait-import
+use ts_rs::TS; // rust-rules: derive-import
+
+/// Initial options shown by the core extraction demo.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionOptionsPayload.ts"
+)]
+pub(crate) struct DemoCoreExtractionOptionsPayload {
+    /// Current extractor implementation version.
+    pub(crate) processor_version: std::string::String,
+    /// Default maximum selected canonical transactions.
+    pub(crate) default_limit: u32,
+    /// Default maximum concurrent extractions.
+    pub(crate) default_max_concurrent_extractions: u32,
+    /// Whether one extraction campaign is currently running.
+    pub(crate) running: bool,
+}
+
+/// UI request for one bounded core extraction campaign.
+#[derive(Clone, Debug, serde::Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionRequest.ts"
+)]
+pub(crate) struct DemoCoreExtractionRequest {
+    /// Source mode: signatures, pending or slot_range.
+    pub(crate) mode: std::string::String,
+    /// Newline-separated signatures for exact selection.
+    pub(crate) signatures_text: std::option::Option<std::string::String>,
+    /// Optional program id already present in core instructions.
+    pub(crate) program_id: std::option::Option<std::string::String>,
+    /// Optional inclusive minimum slot.
+    #[ts(type = "number | null")]
+    pub(crate) min_slot: std::option::Option<u64>,
+    /// Optional inclusive maximum slot.
+    #[ts(type = "number | null")]
+    pub(crate) max_slot: std::option::Option<u64>,
+    /// Maximum selected canonical transactions.
+    pub(crate) limit: u32,
+    /// Maximum concurrent extraction operations.
+    pub(crate) max_concurrent_extractions: u32,
+    /// Forces replacement of already current core rows.
+    pub(crate) force_replay: bool,
+}
+
+/// One progress event emitted to the core extraction window.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionProgressPayload.ts"
+)]
+pub(crate) struct DemoCoreExtractionProgressPayload {
+    /// RFC 3339 timestamp.
+    pub(crate) timestamp: std::string::String,
+    /// Stable severity code.
+    pub(crate) level: std::string::String,
+    /// Human-readable message.
+    pub(crate) message: std::string::String,
+    /// Number of terminal candidates.
+    #[ts(type = "number")]
+    pub(crate) completed: u64,
+    /// Total selected candidates.
+    #[ts(type = "number")]
+    pub(crate) total: u64,
+}
+
+/// Final UI-safe summary for one extraction campaign.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_core_extraction/DemoCoreExtractionSummaryPayload.ts"
+)]
+pub(crate) struct DemoCoreExtractionSummaryPayload {
+    /// Current extractor implementation version.
+    pub(crate) processor_version: std::string::String,
+    /// Number of selected canonical transactions.
+    #[ts(type = "number")]
+    pub(crate) selected: u64,
+    /// Number admitted to the bounded execution queue.
+    #[ts(type = "number")]
+    pub(crate) started: u64,
+    /// Number reaching a terminal result.
+    #[ts(type = "number")]
+    pub(crate) completed: u64,
+    /// Number skipped by version/hash idempotence.
+    #[ts(type = "number")]
+    pub(crate) skipped: u64,
+    /// Number extracted and committed.
+    #[ts(type = "number")]
+    pub(crate) extracted: u64,
+    /// Number failed.
+    #[ts(type = "number")]
+    pub(crate) failed: u64,
+    /// Number admitted but cancelled before a terminal result.
+    #[ts(type = "number")]
+    pub(crate) cancelled_candidates: u64,
+    /// Number selected but not started after cancellation.
+    #[ts(type = "number")]
+    pub(crate) not_started: u64,
+    /// Whether the campaign was cancelled.
+    pub(crate) cancelled: bool,
+    /// Campaign start timestamp.
+    pub(crate) started_at: std::string::String,
+    /// Campaign finish timestamp.
+    pub(crate) finished_at: std::string::String,
+}
+
+pub(crate) struct DemoCoreExtractionObserver<'a> {
+    pub(crate) app_handle: tauri::AppHandle,
+    pub(crate) cancel_requested: &'a std::sync::atomic::AtomicBool,
+}
+
+impl kb_pipeline::CoreExtractionObserver for crate::DemoCoreExtractionObserver<'_> {
+    fn on_progress(&self, event: &kb_pipeline::CoreExtractionProgressEvent) {
+        let payload = crate::DemoCoreExtractionProgressPayload {
+            timestamp: event.timestamp.clone(),
+            level: event.level.code().to_string(),
+            message: event.message.clone(),
+            completed: event.completed,
+            total: event.total,
+        };
+        let emit_result = self.app_handle.emit_to(
+            "demo_core_extraction",
+            "demo-core-extraction-progress",
+            payload,
+        );
+        if let std::result::Result::Err(error) = emit_result {
+            tracing::warn!(
+                target: crate::TRACING_TARGET,
+                "cannot emit core extraction progress: {error}"
+            );
+        }
+    }
+
+    fn is_cancelled(&self) -> bool {
+        return self.cancel_requested.load(std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
+pub(crate) struct DemoCoreExtractionRunGuard<'a> {
+    pub(crate) running: &'a std::sync::atomic::AtomicBool,
+}
+
+impl std::ops::Drop for crate::DemoCoreExtractionRunGuard<'_> {
+    fn drop(&mut self) {
+        self.running.store(false, std::sync::atomic::Ordering::Release);
+    }
+}
+
+pub(crate) fn build_demo_core_extraction_pipeline_request(
+    request: crate::DemoCoreExtractionRequest,
+) -> std::result::Result<kb_pipeline::CoreExtractionRequest, std::string::String> {
+    let source_result = match request.mode.trim() {
+        "signatures" => {
+            let signatures = split_signatures(request.signatures_text.as_deref());
+            std::result::Result::Ok(kb_pipeline::CoreExtractionSource::Signatures(signatures))
+        },
+        "pending" => std::result::Result::Ok(kb_pipeline::CoreExtractionSource::Pending),
+        "program_id" => {
+            let program_id = match request.program_id {
+                std::option::Option::Some(value) if !value.trim().is_empty() => value,
+                _ => {
+                    return std::result::Result::Err(
+                        "program id is required for program extraction".to_string(),
+                    );
+                },
+            };
+            std::result::Result::Ok(kb_pipeline::CoreExtractionSource::ProgramId { program_id })
+        },
+        "slot_range" => {
+            let min_slot = match request.min_slot {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => {
+                    return std::result::Result::Err(
+                        "minimum slot is required for slot range extraction".to_string(),
+                    );
+                },
+            };
+            let max_slot = match request.max_slot {
+                std::option::Option::Some(value) => value,
+                std::option::Option::None => {
+                    return std::result::Result::Err(
+                        "maximum slot is required for slot range extraction".to_string(),
+                    );
+                },
+            };
+            std::result::Result::Ok(kb_pipeline::CoreExtractionSource::SlotRange {
+                min_slot,
+                max_slot,
+            })
+        },
+        _ => std::result::Result::Err("unsupported core extraction mode".to_string()),
+    };
+    let source = match source_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let pipeline_request = kb_pipeline::CoreExtractionRequest {
+        source,
+        limit: request.limit,
+        max_concurrent_extractions: request.max_concurrent_extractions,
+        force_replay: request.force_replay,
+    };
+    let validation_result = pipeline_request.validate();
+    if let std::result::Result::Err(error) = validation_result {
+        return std::result::Result::Err(error.to_string());
+    }
+    return std::result::Result::Ok(pipeline_request);
+}
+
+fn split_signatures(text: std::option::Option<&str>) -> std::vec::Vec<std::string::String> {
+    let source = match text {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => "",
+    };
+    let mut unique = std::collections::BTreeSet::<std::string::String>::new();
+    let mut output = std::vec::Vec::new();
+    for line in source.lines() {
+        let signature = line.trim();
+        if signature.is_empty() {
+            continue;
+        }
+        if unique.insert(signature.to_string()) {
+            output.push(signature.to_string());
+        }
+    }
+    return output;
+}
+
+pub(crate) fn demo_core_extraction_summary_payload(
+    summary: kb_pipeline::CoreExtractionSummary,
+) -> crate::DemoCoreExtractionSummaryPayload {
+    return crate::DemoCoreExtractionSummaryPayload {
+        processor_version: summary.processor_version,
+        selected: summary.selected,
+        started: summary.started,
+        completed: summary.completed,
+        skipped: summary.skipped,
+        extracted: summary.extracted,
+        failed: summary.failed,
+        cancelled_candidates: summary.cancelled_candidates,
+        not_started: summary.not_started,
+        cancelled: summary.cancelled,
+        started_at: summary.started_at,
+        finished_at: summary.finished_at,
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    #[test]
+    fn signature_source_splits_deduplicates_and_ignores_empty_rows() {
+        let values = super::split_signatures(std::option::Option::Some("alpha\n\n beta \nalpha\n"));
+        assert_eq!(values, std::vec!["alpha".to_string(), "beta".to_string()]);
+    }
+}

kb-app-demo-desktop/src/demo_decode_replay.rs

@@ -0,0 +1,947 @@
+// file: kb-app-demo-desktop/src/demo_decode_replay.rs
+// version: 24
+
+//! Tauri commands and UI payloads for contextual instruction decode replay.
+
+use tauri::Emitter; // rust-rules: trait-import
+use ts_rs::TS; // rust-rules: derive-import
+
+/// One decoder selectable by the decode replay demo.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayDecoderOption.ts"
+)]
+pub(crate) struct DemoDecodeReplayDecoderOption {
+    /// Stable decoder name.
+    pub(crate) name: std::string::String,
+    /// Stable decoder version.
+    pub(crate) version: std::string::String,
+    /// Exact supported program identifiers.
+    pub(crate) program_ids: std::vec::Vec<std::string::String>,
+}
+
+/// Initial options shown by the contextual decode replay demo.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayOptionsPayload.ts"
+)]
+pub(crate) struct DemoDecodeReplayOptionsPayload {
+    /// Common orchestration version.
+    pub(crate) pipeline_version: std::string::String,
+    /// Available contextual decoders.
+    pub(crate) decoders: std::vec::Vec<crate::DemoDecodeReplayDecoderOption>,
+    /// Stable names of materializers currently registered by the demo.
+    pub(crate) materializer_names: std::vec::Vec<std::string::String>,
+    /// Default bounded selection limit.
+    pub(crate) default_limit: u32,
+    /// Default maximum concurrent contextual inputs.
+    pub(crate) default_max_concurrent_inputs: u32,
+    /// Whether one replay campaign is running.
+    pub(crate) running: bool,
+}
+
+/// UI request for one bounded contextual decode replay campaign.
+#[derive(Clone, Debug, serde::Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayRequest.ts"
+)]
+pub(crate) struct DemoDecodeReplayRequest {
+    /// Newline-separated exact transaction signatures.
+    pub(crate) signatures_text: std::option::Option<std::string::String>,
+    /// Optional exact program identifier.
+    pub(crate) program_id: std::option::Option<std::string::String>,
+    /// Instruction processing state code, actionable or incomplete_signatures.
+    pub(crate) instruction_state: std::string::String,
+    /// Newline-separated exact stable instruction paths.
+    pub(crate) instruction_paths_text: std::option::Option<std::string::String>,
+    /// Explicit selected decoder names.
+    pub(crate) decoder_names: std::vec::Vec<std::string::String>,
+    /// Maximum selected contextual inputs.
+    pub(crate) limit: u32,
+    /// Maximum concurrent contextual inputs.
+    pub(crate) max_concurrent_inputs: u32,
+    /// Whether every compatible decoder must run.
+    pub(crate) all_compatible: bool,
+    /// Replaces only processor-owned outputs for the same version and input.
+    pub(crate) force_replay: bool,
+    /// Explicitly authorizes a bounded force replay without exact signatures.
+    pub(crate) force_replay_all_matching: bool,
+    /// Runs compatible materializers after decoded event persistence.
+    pub(crate) materialize_after_decode: bool,
+}
+
+/// One progress event emitted to the decode replay window.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplayProgressPayload.ts"
+)]
+pub(crate) struct DemoDecodeReplayProgressPayload {
+    /// Stable process-local campaign identifier.
+    pub(crate) campaign_id: std::string::String,
+    /// RFC 3339 timestamp.
+    pub(crate) timestamp: std::string::String,
+    /// Stable severity code.
+    pub(crate) level: std::string::String,
+    /// Human-readable message.
+    pub(crate) message: std::string::String,
+    /// Number of terminal contextual inputs.
+    #[ts(type = "number")]
+    pub(crate) completed: u64,
+    /// Total selected contextual inputs.
+    #[ts(type = "number")]
+    pub(crate) total: u64,
+}
+
+/// One processor counter row returned to the UI.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeProcessorSummaryPayload.ts"
+)]
+pub(crate) struct DemoDecodeProcessorSummaryPayload {
+    /// Stable processor name.
+    pub(crate) processor_name: std::string::String,
+    /// Stable processor version.
+    pub(crate) processor_version: std::string::String,
+    /// Compatible dispatch count.
+    #[ts(type = "number")]
+    pub(crate) dispatched: u64,
+    /// Version and hash skip count.
+    #[ts(type = "number")]
+    pub(crate) skipped: u64,
+    /// Decoded input count.
+    #[ts(type = "number")]
+    pub(crate) decoded: u64,
+    /// Ignored input count.
+    #[ts(type = "number")]
+    pub(crate) ignored: u64,
+    /// Unsupported input count.
+    #[ts(type = "number")]
+    pub(crate) unsupported: u64,
+    /// Failed input count.
+    #[ts(type = "number")]
+    pub(crate) failed: u64,
+    /// Materialized output count.
+    #[ts(type = "number")]
+    pub(crate) materialized_outputs: u64,
+    /// Materialization refusal count.
+    #[ts(type = "number")]
+    pub(crate) materialization_refused: u64,
+}
+
+/// Final UI-safe summary for one contextual decode replay campaign.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeReplaySummaryPayload.ts"
+)]
+pub(crate) struct DemoDecodeReplaySummaryPayload {
+    /// Stable process-local campaign identifier.
+    pub(crate) campaign_id: std::string::String,
+    /// Common pipeline version.
+    pub(crate) pipeline_version: std::string::String,
+    /// Number of selected contextual inputs.
+    #[ts(type = "number")]
+    pub(crate) selected: u64,
+    /// Number admitted to execution.
+    #[ts(type = "number")]
+    pub(crate) started: u64,
+    /// Number reaching a terminal state.
+    #[ts(type = "number")]
+    pub(crate) completed: u64,
+    /// Number with no compatible enabled decoder.
+    #[ts(type = "number")]
+    pub(crate) unmatched: u64,
+    /// Number never started after cancellation.
+    #[ts(type = "number")]
+    pub(crate) not_started: u64,
+    /// Number of failed contextual inputs.
+    #[ts(type = "number")]
+    pub(crate) failed_inputs: u64,
+    /// Whether cancellation was observed.
+    pub(crate) cancelled: bool,
+    /// Per-processor counters.
+    pub(crate) processors: std::vec::Vec<crate::DemoDecodeProcessorSummaryPayload>,
+    /// Campaign start timestamp.
+    pub(crate) started_at: std::string::String,
+    /// Campaign finish timestamp.
+    pub(crate) finished_at: std::string::String,
+}
+
+/// One aggregated coverage row returned to the UI.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeCoverageSummaryPayload.ts"
+)]
+pub(crate) struct DemoDecodeCoverageSummaryPayload {
+    /// Stable processor name.
+    pub(crate) processor_name: std::string::String,
+    /// Stable processor version.
+    pub(crate) processor_version: std::string::String,
+    /// Exact program identifier.
+    pub(crate) program_id: std::string::String,
+    /// Optional stable surface code.
+    pub(crate) surface_code: std::option::Option<std::string::String>,
+    /// Stable entry classifier.
+    pub(crate) entry_code: std::string::String,
+    /// Declared count.
+    #[ts(type = "number")]
+    pub(crate) declared_count: i64,
+    /// Observed count.
+    #[ts(type = "number")]
+    pub(crate) observed_count: i64,
+    /// Recognized count.
+    #[ts(type = "number")]
+    pub(crate) recognized_count: i64,
+    /// Decoded observation count.
+    #[ts(type = "number")]
+    pub(crate) decoded_count: i64,
+    /// Materialized output count.
+    #[ts(type = "number")]
+    pub(crate) materialized_count: i64,
+    /// Error count.
+    #[ts(type = "number")]
+    pub(crate) error_count: i64,
+    /// Unknown or unsupported observation count.
+    #[ts(type = "number")]
+    pub(crate) unknown_count: i64,
+    /// Successful source transaction count.
+    #[ts(type = "number")]
+    pub(crate) successful_transaction_count: i64,
+    /// Failed source transaction count.
+    #[ts(type = "number")]
+    pub(crate) failed_transaction_count: i64,
+}
+
+/// Read-only decode, ledger and coverage diagnostics.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoDecodeDiagnosticsPayload.ts"
+)]
+pub(crate) struct DemoDecodeDiagnosticsPayload {
+    /// Decode, materialization and ledger table diagnostics.
+    pub(crate) tables: std::vec::Vec<crate::DemoSqlTableSnapshot>,
+    /// Aggregated declared and observed coverage.
+    pub(crate) coverage: std::vec::Vec<crate::DemoDecodeCoverageSummaryPayload>,
+}
+
+/// Bounded read request for committed transaction annotations.
+#[derive(Clone, Debug, serde::Deserialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRequest.ts"
+)]
+pub(crate) struct DemoTransactionAnnotationRequest {
+    /// Optional partial transaction signature.
+    pub(crate) signature_contains: std::option::Option<std::string::String>,
+    /// Maximum returned rows.
+    pub(crate) limit: u32,
+}
+
+/// UI-safe committed transaction annotation row.
+#[derive(Clone, Debug, serde::Serialize, TS)]
+#[serde(rename_all = "camelCase")]
+#[ts(
+    export,
+    export_to = "../frontend/ts/bindings/kb_app_demo_desktop/demo_decode_replay/DemoTransactionAnnotationRow.ts"
+)]
+pub(crate) struct DemoTransactionAnnotationRow {
+    /// Materializer processor version.
+    pub(crate) processor_version: std::string::String,
+    /// Source decoder and version.
+    pub(crate) decoder: std::string::String,
+    /// Transaction signature.
+    pub(crate) signature: std::string::String,
+    /// Decimal slot rendered as text to preserve JSON precision.
+    pub(crate) slot: std::string::String,
+    /// Stable outer or inner instruction path.
+    pub(crate) instruction_path: std::string::String,
+    /// Exact Memo generation.
+    pub(crate) generation: std::string::String,
+    /// Exact Memo Program ID.
+    pub(crate) program_id: std::string::String,
+    /// Complete bounded UTF-8 Memo text.
+    pub(crate) text: std::string::String,
+    /// Exact payload byte length.
+    pub(crate) payload_length_bytes: u32,
+    /// Canonical payload SHA-256.
+    pub(crate) payload_sha256: std::string::String,
+    /// Runtime-verified ordered signer keys.
+    pub(crate) verified_signers: std::vec::Vec<std::string::String>,
+    /// Stable materializer idempotence key.
+    pub(crate) idempotence_key: std::string::String,
+    /// Database creation timestamp.
+    pub(crate) created_at: std::string::String,
+    /// Database replacement timestamp.
+    pub(crate) updated_at: std::string::String,
+}
+
+pub(crate) struct DemoDecodeReplayObserver<'a> {
+    pub(crate) app_handle: tauri::AppHandle,
+    pub(crate) campaign_id: std::string::String,
+    pub(crate) cancel_requested: &'a std::sync::atomic::AtomicBool,
+}
+
+impl kb_pipeline::DecodeReplayObserver for crate::DemoDecodeReplayObserver<'_> {
+    fn on_progress(&self, event: &kb_pipeline::DecodeReplayProgressEvent) {
+        let payload = DemoDecodeReplayProgressPayload {
+            campaign_id: self.campaign_id.clone(),
+            timestamp: event.timestamp.clone(),
+            level: event.level.code().to_string(),
+            message: event.message.clone(),
+            completed: event.completed,
+            total: event.total,
+        };
+        let emit_result =
+            self.app_handle
+                .emit_to("demo_decode_replay", "demo-decode-replay-progress", payload);
+        if let std::result::Result::Err(error) = emit_result {
+            tracing::warn!(target: crate::TRACING_TARGET, action = "emit_progress", campaign_id = %self.campaign_id, error = %error, "cannot emit decode replay progress");
+        }
+    }
+
+    fn is_cancelled(&self) -> bool {
+        return self.cancel_requested.load(std::sync::atomic::Ordering::Relaxed);
+    }
+}
+
+pub(crate) struct DemoDecodeReplayRunGuard<'a> {
+    pub(crate) running: &'a std::sync::atomic::AtomicBool,
+    pub(crate) campaign_id: &'a std::sync::Mutex<std::option::Option<std::string::String>>,
+}
+
+impl std::ops::Drop for crate::DemoDecodeReplayRunGuard<'_> {
+    fn drop(&mut self) {
+        self.running.store(false, std::sync::atomic::Ordering::Release);
+        let lock_result = self.campaign_id.lock();
+        if let std::result::Result::Ok(mut active_campaign_id) = lock_result {
+            *active_campaign_id = std::option::Option::None;
+        }
+    }
+}
+
+pub(crate) fn register_active_campaign(
+    campaign_slot: &std::sync::Mutex<std::option::Option<std::string::String>>,
+    campaign_id: &str,
+) -> std::result::Result<(), std::string::String> {
+    let lock_result = campaign_slot.lock();
+    let mut active_campaign_id = match lock_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(_) => {
+            return std::result::Result::Err(
+                "cannot register active contextual decode campaign".to_string(),
+            );
+        },
+    };
+    *active_campaign_id = std::option::Option::Some(campaign_id.to_string());
+    return std::result::Result::Ok(());
+}
+
+pub(crate) fn available_materializers()
+-> std::vec::Vec<std::sync::Arc<dyn kb_lib::MtApiEventMaterializer>> {
+    return std::vec![
+        std::sync::Arc::new(kb_lib::MtAdminMaterializer),
+        std::sync::Arc::new(kb_lib::MtComplianceAuditMaterializer),
+        std::sync::Arc::new(kb_lib::MtFeesMaterializer),
+        std::sync::Arc::new(kb_lib::MtLifecycleMaterializer),
+        std::sync::Arc::new(kb_lib::MtStakingMaterializer),
+        std::sync::Arc::new(kb_lib::MtTokenAccountsMaterializer),
+        std::sync::Arc::new(kb_lib::MtRiskMaterializer),
+        std::sync::Arc::new(kb_lib::MtTransactionAnnotationMaterializer,),
+    ];
+}
+
+pub(crate) fn available_decoders()
+-> std::vec::Vec<std::sync::Arc<dyn kb_lib::DcApiInstructionDecoder>> {
+    return std::vec![
+        std::sync::Arc::new(kb_lib::DcSolanaCoreDecoder),
+        std::sync::Arc::new(kb_lib::DcSplAssociatedTokenAccountDecoder,),
+        std::sync::Arc::new(kb_lib::DcSplElgamalRegistryDecoder),
+        std::sync::Arc::new(kb_lib::DcSplMemoDecoder),
+        std::sync::Arc::new(kb_lib::DcSplTokenDecoder),
+        std::sync::Arc::new(kb_lib::DcSplToken2022Decoder),
+    ];
+}
+
+pub(crate) fn build_demo_decode_replay_pipeline_request(
+    request: crate::DemoDecodeReplayRequest,
+    campaign_id: std::string::String,
+) -> std::result::Result<kb_pipeline::DecodeReplayRequest, std::string::String> {
+    let states_result = processing_states(request.instruction_state.as_str());
+    let states = match states_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let signatures = split_lines(request.signatures_text.as_deref());
+    let instruction_paths = split_lines(request.instruction_paths_text.as_deref());
+    let incomplete_signatures = request.instruction_state.trim() == "incomplete_signatures";
+    let program_ids = match request.program_id {
+        std::option::Option::Some(value) if !value.trim().is_empty() => {
+            std::vec![value.trim().to_string()]
+        },
+        _ => std::vec::Vec::new(),
+    };
+    let selection_result = kb_store::DecodeSelectionFilter::new(
+        signatures,
+        states,
+        std::option::Option::None,
+        std::option::Option::None,
+        program_ids,
+        instruction_paths,
+        incomplete_signatures,
+        request.limit,
+    );
+    let selection = match selection_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    let dispatch_policy = if request.all_compatible {
+        kb_pipeline::DecodeDispatchPolicy::AllCompatible
+    } else {
+        kb_pipeline::DecodeDispatchPolicy::HighestPriority
+    };
+    let pipeline_request = kb_pipeline::DecodeReplayRequest {
+        campaign_id,
+        selection,
+        decoder_names: request.decoder_names,
+        dispatch_policy,
+        max_concurrent_inputs: request.max_concurrent_inputs,
+        force_replay: request.force_replay,
+        force_replay_all_matching: request.force_replay_all_matching,
+        materialize_after_decode: request.materialize_after_decode,
+    };
+    let validation_result = pipeline_request.validate();
+    if let std::result::Result::Err(error) = validation_result {
+        return std::result::Result::Err(error.to_string());
+    }
+    tracing::debug!(
+        target: crate::TRACING_TARGET,
+        action = "build_request",
+        campaign_id = %pipeline_request.campaign_id,
+        signature_count = pipeline_request.selection.signatures.len(),
+        signature_sample = ?text_sample(pipeline_request.selection.signatures.as_slice(), 5),
+        processing_states = ?pipeline_request.selection.processing_states,
+        program_ids = ?pipeline_request.selection.program_ids,
+        instruction_paths = ?pipeline_request.selection.instruction_paths,
+        incomplete_signatures = pipeline_request.selection.incomplete_signatures,
+        limit = pipeline_request.selection.limit,
+        decoder_names = ?pipeline_request.decoder_names,
+        dispatch_policy = ?pipeline_request.dispatch_policy,
+        max_concurrent_inputs = pipeline_request.max_concurrent_inputs,
+        force_replay = pipeline_request.force_replay,
+        force_replay_all_matching = pipeline_request.force_replay_all_matching,
+        materialize_after_decode = pipeline_request.materialize_after_decode,
+        "built normalized contextual decode replay request"
+    );
+    return std::result::Result::Ok(pipeline_request);
+}
+
+pub(crate) fn optional_line_count(value: &std::option::Option<std::string::String>) -> usize {
+    return match value {
+        std::option::Option::Some(text) => {
+            text.lines().filter(|line| return !line.trim().is_empty()).count()
+        },
+        std::option::Option::None => 0,
+    };
+}
+
+pub(crate) fn text_sample(values: &[std::string::String], limit: usize) -> std::vec::Vec<&str> {
+    return values.iter().take(limit).map(std::string::String::as_str).collect();
+}
+
+fn processing_states(
+    value: &str,
+) -> std::result::Result<std::vec::Vec<kb_store::CoreInstructionProcessingState>, std::string::String>
+{
+    return match value.trim() {
+        "incomplete_signatures" | "actionable" => std::result::Result::Ok(std::vec![
+            kb_store::CoreInstructionProcessingState::Pending,
+            kb_store::CoreInstructionProcessingState::Failed,
+            kb_store::CoreInstructionProcessingState::ReplayRequested,
+        ]),
+        "pending" => {
+            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Pending])
+        },
+        "failed" => {
+            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Failed])
+        },
+        "replay_requested" => std::result::Result::Ok(std::vec![
+            kb_store::CoreInstructionProcessingState::ReplayRequested
+        ]),
+        "decoded" => {
+            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Decoded])
+        },
+        "ignored" => {
+            std::result::Result::Ok(std::vec![kb_store::CoreInstructionProcessingState::Ignored])
+        },
+        "materialized" => std::result::Result::Ok(std::vec![
+            kb_store::CoreInstructionProcessingState::Materialized
+        ]),
+        _ => std::result::Result::Err("unsupported instruction processing state".to_string()),
+    };
+}
+
+fn split_lines(text: std::option::Option<&str>) -> std::vec::Vec<std::string::String> {
+    let source = match text {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => "",
+    };
+    let mut unique = std::collections::BTreeSet::<std::string::String>::new();
+    let mut output = std::vec::Vec::new();
+    for line in source.lines() {
+        let value = line.trim();
+        if value.is_empty() {
+            continue;
+        }
+        if unique.insert(value.to_string()) {
+            output.push(value.to_string());
+        }
+    }
+    return output;
+}
+
+pub(crate) fn demo_decode_replay_summary_payload(
+    summary: kb_pipeline::DecodeReplaySummary,
+) -> crate::DemoDecodeReplaySummaryPayload {
+    return crate::DemoDecodeReplaySummaryPayload {
+        campaign_id: summary.campaign_id,
+        pipeline_version: summary.pipeline_version,
+        selected: summary.selected,
+        started: summary.started,
+        completed: summary.completed,
+        unmatched: summary.unmatched,
+        not_started: summary.not_started,
+        failed_inputs: summary.failed_inputs,
+        cancelled: summary.cancelled,
+        processors: summary
+            .processors
+            .into_iter()
+            .map(|processor| {
+                return crate::DemoDecodeProcessorSummaryPayload {
+                    processor_name: processor.processor_name,
+                    processor_version: processor.processor_version,
+                    dispatched: processor.dispatched,
+                    skipped: processor.skipped,
+                    decoded: processor.decoded,
+                    ignored: processor.ignored,
+                    unsupported: processor.unsupported,
+                    failed: processor.failed,
+                    materialized_outputs: processor.materialized_outputs,
+                    materialization_refused: processor.materialization_refused,
+                };
+            })
+            .collect(),
+        started_at: summary.started_at,
+        finished_at: summary.finished_at,
+    };
+}
+
+pub(crate) fn coverage_payload(
+    value: kb_store::DecodeCoverageSummaryRow,
+) -> crate::DemoDecodeCoverageSummaryPayload {
+    return crate::DemoDecodeCoverageSummaryPayload {
+        processor_name: value.processor_name,
+        processor_version: value.processor_version,
+        program_id: value.program_id,
+        surface_code: value.surface_code,
+        entry_code: value.entry_code,
+        declared_count: value.declared_count,
+        observed_count: value.observed_count,
+        recognized_count: value.recognized_count,
+        decoded_count: value.decoded_count,
+        materialized_count: value.materialized_count,
+        error_count: value.error_count,
+        unknown_count: value.unknown_count,
+        successful_transaction_count: value.successful_transaction_count,
+        failed_transaction_count: value.failed_transaction_count,
+    };
+}
+
+pub(crate) fn annotation_payload(
+    row: kb_store::MaterializedEventQueryRow,
+) -> std::result::Result<crate::DemoTransactionAnnotationRow, std::string::String> {
+    if row.processor_name != "transaction_annotations"
+        || row.materialized_family != "transaction_annotation"
+    {
+        return std::result::Result::Err(
+            "materialized row is not a transaction annotation".to_string(),
+        );
+    }
+    let payload = &row.payload_json;
+    let byte_length = match payload.get("payloadLengthBytes").and_then(serde_json::Value::as_u64) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => {
+            return std::result::Result::Err(
+                "transaction annotation payloadLengthBytes is absent".to_string(),
+            );
+        },
+    };
+    let byte_length_result = u32::try_from(byte_length);
+    let payload_length_bytes = match byte_length_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => {
+            return std::result::Result::Err(format!(
+                "transaction annotation payloadLengthBytes is not UI-safe: {error}"
+            ));
+        },
+    };
+    let signers = match payload.get("signers") {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => {
+            return std::result::Result::Err(
+                "transaction annotation signers are absent".to_string(),
+            );
+        },
+    };
+    let verified_values = match signers.get("verified").and_then(serde_json::Value::as_array) {
+        std::option::Option::Some(value) => value,
+        std::option::Option::None => {
+            return std::result::Result::Err(
+                "transaction annotation verified signers are invalid".to_string(),
+            );
+        },
+    };
+    let mut verified_signers = std::vec::Vec::with_capacity(verified_values.len());
+    for value in verified_values {
+        match value.as_str() {
+            std::option::Option::Some(signer) if !signer.trim().is_empty() => {
+                verified_signers.push(signer.to_string());
+            },
+            _ => {
+                return std::result::Result::Err(
+                    "transaction annotation verified signer is invalid".to_string(),
+                );
+            },
+        }
+    }
+    let instruction_path = match required_annotation_text(payload, "instructionPath") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let generation = match required_annotation_text(payload, "generation") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let program_id = match required_annotation_text(payload, "programId") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let text = match required_annotation_text(payload, "text") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let payload_sha256 = match required_annotation_text(payload, "payloadSha256") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let idempotence_key = match required_annotation_text(payload, "idempotenceKey") {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    return std::result::Result::Ok(crate::DemoTransactionAnnotationRow {
+        processor_version: row.processor_version,
+        decoder: format!("{}@{}", row.source_decoder_name, row.source_decoder_version),
+        signature: row.signature,
+        slot: row.slot.to_string(),
+        instruction_path,
+        generation,
+        program_id,
+        text,
+        payload_length_bytes,
+        payload_sha256,
+        verified_signers,
+        idempotence_key,
+        created_at: row.created_at,
+        updated_at: row.updated_at,
+    });
+}
+
+fn required_annotation_text(
+    payload: &serde_json::Value,
+    field: &str,
+) -> std::result::Result<std::string::String, std::string::String> {
+    return match payload.get(field).and_then(serde_json::Value::as_str) {
+        std::option::Option::Some(value) => std::result::Result::Ok(value.to_string()),
+        std::option::Option::None => {
+            std::result::Result::Err(format!("transaction annotation {field} is absent"))
+        },
+    };
+}
+
+#[cfg(test)]
+mod tests {
+    fn ata_observation(entry: &str) -> kb_lib::DcApiDecodedObservation {
+        return kb_lib::DcApiDecodedObservation {
+            event_key: format!("ata:{entry}:0"),
+            event: kb_lib::MdDecodedProtocolEvent {
+                signature: kb_lib::MdSignature("signature".to_string()),
+                slot: kb_lib::MdSlot(1),
+                instruction_path: kb_lib::MdInstructionPath("0".to_string()),
+                program_id: kb_lib::MdProgramId(
+                    kb_program_ids::ASSOCIATED_TOKEN_PROGRAM_ID.to_string(),
+                ),
+                protocol_code: kb_lib::MdProtocolCode("spl_associated_token_account".to_string()),
+                surface_code: kb_lib::MdSurfaceCode("spl_associated_token_account".to_string()),
+                event_code: kb_lib::MdEventCode(format!("spl_associated_token_account.{entry}")),
+                event_name: kb_lib::MdEventName(entry.to_string()),
+                event_family: kb_lib::MdEventFamily::Lifecycle,
+                source_kind: kb_lib::MdEventSourceKind::Instruction,
+                confidence: kb_lib::MdDecoderConfidence::ManualExact,
+            },
+            payload_json: serde_json::json!({}),
+            transaction_failed: false,
+            transaction_error: std::option::Option::None,
+            observation_committed: true,
+            proof: kb_lib::DcApiDecoderProof {
+                kind: kb_lib::DcApiDecoderProofKind::Manual,
+                confidence: kb_lib::MdDecoderConfidence::ManualExact,
+                evidence: std::vec!["fixture".to_string()],
+            },
+        };
+    }
+
+    #[test]
+    fn native_memo_token_2022_elgamal_classic_token_and_ata_decoders_are_registered() {
+        let decoders = crate::available_decoders();
+        assert_eq!(decoders.len(), 6);
+        let names = decoders
+            .iter()
+            .map(|decoder| return decoder.identity().name)
+            .collect::<std::vec::Vec<_>>();
+        assert_eq!(
+            names,
+            std::vec![
+                "solana_native_classifier".to_string(),
+                "spl_associated_token_account".to_string(),
+                "spl_elgamal_registry".to_string(),
+                "spl_memo".to_string(),
+                "spl_token".to_string(),
+                "spl_token_2022".to_string(),
+            ]
+        );
+    }
+
+    #[test]
+    fn runtime_materializer_registry_is_complete_for_current_instructional_surfaces() {
+        let materializers = crate::available_materializers();
+        assert_eq!(materializers.len(), 8);
+        let names = materializers
+            .iter()
+            .map(|materializer| return materializer.identity().name)
+            .collect::<std::vec::Vec<_>>();
+        assert_eq!(
+            names,
+            std::vec![
+                "solana_native_admin".to_string(),
+                "solana_native_compliance_audit".to_string(),
+                "fees".to_string(),
+                "solana_native_lifecycle".to_string(),
+                "solana_native_staking".to_string(),
+                "spl_token_accounts".to_string(),
+                "spl_token_risk".to_string(),
+                "transaction_annotations".to_string(),
+            ]
+        );
+    }
+
+    #[test]
+    fn ata_materializer_ownership_is_exact_in_the_runtime_registry() {
+        for (entry, expected) in [
+            ("create", std::vec!["spl_token_accounts".to_string()]),
+            (
+                "recover_nested",
+                std::vec!["spl_token_accounts".to_string(), "spl_token_risk".to_string(),],
+            ),
+        ] {
+            let observation = ata_observation(entry);
+            let owners = crate::available_materializers()
+                .iter()
+                .filter(|materializer| return materializer.accepts_observation(&observation))
+                .map(|materializer| return materializer.identity().name)
+                .collect::<std::vec::Vec<_>>();
+            assert_eq!(owners, expected);
+        }
+    }
+
+    #[test]
+    fn line_selection_is_trimmed_and_deduplicated() {
+        let values = super::split_lines(std::option::Option::Some(" a \n\na\nb\n"));
+        assert_eq!(values, std::vec!["a".to_string(), "b".to_string()]);
+    }
+
+    fn request(
+        signatures_text: std::option::Option<&str>,
+        force_replay: bool,
+        force_replay_all_matching: bool,
+        materialize_after_decode: bool,
+    ) -> crate::DemoDecodeReplayRequest {
+        return crate::DemoDecodeReplayRequest {
+            signatures_text: signatures_text.map(|value| return value.to_string()),
+            program_id: std::option::Option::None,
+            instruction_state: "actionable".to_string(),
+            instruction_paths_text: std::option::Option::None,
+            decoder_names: std::vec!["solana_native_classifier".to_string()],
+            limit: 10,
+            max_concurrent_inputs: 2,
+            all_compatible: false,
+            force_replay,
+            force_replay_all_matching,
+            materialize_after_decode,
+        };
+    }
+
+    #[test]
+    fn force_replay_requires_signatures_or_all_matching_authorization() {
+        let result = crate::build_demo_decode_replay_pipeline_request(
+            request(std::option::Option::None, true, false, false),
+            "decode-test".to_string(),
+        );
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn force_replay_all_matching_is_forwarded() {
+        let result = crate::build_demo_decode_replay_pipeline_request(
+            request(std::option::Option::None, true, true, false),
+            "decode-test".to_string(),
+        );
+        let request = match result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("request mapping failed: {error}"),
+        };
+        assert!(request.force_replay_all_matching);
+        assert!(request.selection.signatures.is_empty());
+    }
+
+    #[test]
+    fn run_guard_clears_active_campaign_state() {
+        let running = std::sync::atomic::AtomicBool::new(true);
+        let campaign_id =
+            std::sync::Mutex::new(std::option::Option::Some("decode-test".to_string()));
+        {
+            let _guard = crate::DemoDecodeReplayRunGuard {
+                running: &running,
+                campaign_id: &campaign_id,
+            };
+        }
+        assert!(!running.load(std::sync::atomic::Ordering::Acquire));
+        let lock_result = campaign_id.lock();
+        let active_campaign_id = match lock_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"),
+        };
+        assert!(active_campaign_id.is_none());
+    }
+
+    #[test]
+    fn register_active_campaign_sets_slot_without_exposing_guard() {
+        let campaign_id = std::sync::Mutex::new(std::option::Option::None);
+        let result = crate::register_active_campaign(&campaign_id, "decode-test");
+        assert!(result.is_ok());
+        let lock_result = campaign_id.lock();
+        let active_campaign_id = match lock_result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(_) => panic!("campaign id mutex must not be poisoned"),
+        };
+        assert_eq!(active_campaign_id.as_deref(), std::option::Option::Some("decode-test"));
+    }
+
+    #[test]
+    fn incomplete_signature_state_uses_actionable_states_and_expansion() {
+        let mut request = request(std::option::Option::None, false, false, true);
+        request.instruction_state = "incomplete_signatures".to_string();
+        let result = crate::build_demo_decode_replay_pipeline_request(
+            request,
+            "decode-incomplete".to_string(),
+        );
+        let pipeline_request = match result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("request mapping failed: {error}"),
+        };
+        assert!(pipeline_request.selection.incomplete_signatures);
+        assert_eq!(pipeline_request.selection.processing_states.len(), 3);
+        assert!(!pipeline_request.force_replay);
+        assert!(pipeline_request.materialize_after_decode);
+    }
+
+    #[test]
+    fn actionable_state_contains_pending_failed_and_replay_requested() {
+        let result = super::processing_states("actionable");
+        let states = match result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("state mapping failed: {error}"),
+        };
+        assert_eq!(states.len(), 3);
+    }
+
+    #[test]
+    fn committed_annotation_row_is_mapped_to_ui_safe_contract() {
+        let row = kb_store::MaterializedEventQueryRow {
+            processor_name: "transaction_annotations".to_string(),
+            processor_version: "0.4.3".to_string(),
+            input_key: "input".to_string(),
+            output_key: "output".to_string(),
+            source_event_key: "memo:0".to_string(),
+            source_decoder_name: "spl_memo".to_string(),
+            source_decoder_version: "0.4.3".to_string(),
+            signature: "signature".to_string(),
+            slot: u64::MAX,
+            materialized_family: "transaction_annotation".to_string(),
+            payload_json: serde_json::json!({
+                "instructionPath": "1/0",
+                "generation": "v4",
+                "programId": kb_program_ids::SPL_MEMO_V4_PROGRAM_ID,
+                "text": "annotation",
+                "payloadLengthBytes": 10,
+                "payloadSha256": "11",
+                "signers": {"verified": ["signer"]},
+                "idempotenceKey": "stable"
+            }),
+            created_at: "created".to_string(),
+            updated_at: "updated".to_string(),
+        };
+        let result = crate::annotation_payload(row);
+        let mapped = match result {
+            std::result::Result::Ok(value) => value,
+            std::result::Result::Err(error) => panic!("annotation mapping failed: {error}"),
+        };
+        assert_eq!(mapped.slot, u64::MAX.to_string());
+        assert_eq!(mapped.text, "annotation");
+        assert_eq!(mapped.verified_signers, std::vec!["signer".to_string()]);
+    }
+
+    #[test]
+    fn malformed_annotation_payload_fails_closed() {
+        let row = kb_store::MaterializedEventQueryRow {
+            processor_name: "transaction_annotations".to_string(),
+            processor_version: "0.4.3".to_string(),
+            input_key: "input".to_string(),
+            output_key: "output".to_string(),
+            source_event_key: "memo:0".to_string(),
+            source_decoder_name: "spl_memo".to_string(),
+            source_decoder_version: "0.4.3".to_string(),
+            signature: "signature".to_string(),
+            slot: 1,
+            materialized_family: "transaction_annotation".to_string(),
+            payload_json: serde_json::json!({}),
+            created_at: "created".to_string(),
+            updated_at: "updated".to_string(),
+        };
+        assert!(crate::annotation_payload(row).is_err());
+    }
+}

kb-app-demo-desktop/src/demo_sql_common.rs

@@ -87,7 +87,6 @@ pub(crate) fn open_sql_demo_window(
     };
 }
 
-
 /// Builds PostgreSQL options from the active profile without coupling kb-store to kb-config.
 pub(crate) fn postgres_store_options_from_profile(
     profile: &kb_config::ProfileConfig,

kb-app-demo-desktop/src/demo_sql_replay_candidates.rs

@@ -326,10 +326,8 @@ pub(crate) fn program_scope_from_code(
 
 pub(crate) fn optional_entity_kind_from_code(
     code: std::option::Option<&str>,
-) -> std::result::Result<
-    std::option::Option<kb_store::PostgresReplayEntityKind>,
-    std::string::String,
-> {
+) -> std::result::Result<std::option::Option<kb_store::PostgresReplayEntityKind>, std::string::String>
+{
     return match code {
         std::option::Option::Some(value) => {
             let result = entity_kind_from_code(value);

kb-app-demo-desktop/src/demo_ws.rs

@@ -338,13 +338,19 @@ async fn ensure_demo_ws_session(
     } else {
         kb_onchain_transport::WsReconnectPolicy::disabled()
     };
-    let capabilities =
-        kb_onchain_transport::StandardWsCapabilities::from_endpoint(selected_client.endpoint_config());
-    let session =
-        match kb_onchain_transport::WsSession::connect(selected_client, capabilities, reconnect_policy).await {
-            std::result::Result::Ok(session) => std::sync::Arc::new(session),
-            std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
-        };
+    let capabilities = kb_onchain_transport::StandardWsCapabilities::from_endpoint(
+        selected_client.endpoint_config(),
+    );
+    let session = match kb_onchain_transport::WsSession::connect(
+        selected_client,
+        capabilities,
+        reconnect_policy,
+    )
+    .await
+    {
+        std::result::Result::Ok(session) => std::sync::Arc::new(session),
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
     spawn_demo_ws_event_bridge(app_handle.clone(), session.clone());
     {
         let mut guard = state.demo_ws_session().lock().await;
@@ -442,7 +448,10 @@ fn spawn_demo_ws_event_bridge(
                 kb_onchain_transport::WsSessionEvent::Diagnostic { code, message } => {
                     emit_demo_ws_message(&app_handle, &code, truncate_payload(message));
                 },
-                kb_onchain_transport::WsSessionEvent::Reconnecting { attempt, maximum_attempts } => {
+                kb_onchain_transport::WsSessionEvent::Reconnecting {
+                    attempt,
+                    maximum_attempts,
+                } => {
                     emit_demo_ws_message(
                         &app_handle,
                         "reconnecting",
@@ -473,7 +482,10 @@ fn spawn_demo_ws_event_bridge(
     });
 }
 
-fn emit_demo_ws_status(app_handle: &tauri::AppHandle, snapshot: kb_onchain_transport::WsSessionSnapshot) {
+fn emit_demo_ws_status(
+    app_handle: &tauri::AppHandle,
+    snapshot: kb_onchain_transport::WsSessionSnapshot,
+) {
     let window = match app_handle.get_webview_window("demo_ws") {
         std::option::Option::Some(window) => window,
         std::option::Option::None => return,
@@ -501,7 +513,9 @@ fn emit_demo_ws_message(
     }
 }
 
-fn status_from_snapshot(snapshot: kb_onchain_transport::WsSessionSnapshot) -> crate::DemoWsStatusPayload {
+fn status_from_snapshot(
+    snapshot: kb_onchain_transport::WsSessionSnapshot,
+) -> crate::DemoWsStatusPayload {
     let mut subscriptions = std::vec::Vec::new();
     for subscription in &snapshot.subscriptions {
         if let std::option::Option::Some(remote_subscription_id) =
@@ -602,7 +616,9 @@ pub(crate) fn build_ws_method_options() -> std::vec::Vec<crate::DemoWsMethodOpti
         options.push(crate::DemoWsMethodOption {
             method: specification.subscribe_method.to_string(),
             unsubscribe_method: specification.unsubscribe_method.to_string(),
-            request_kind: kb_onchain_transport::request_kind_from_method(specification.subscribe_method),
+            request_kind: kb_onchain_transport::request_kind_from_method(
+                specification.subscribe_method,
+            ),
             label: label.to_string(),
             requires_target,
             requires_filter_json,
@@ -624,10 +640,10 @@ fn build_standard_ws_request(
                 std::result::Result::Ok(target) => target,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
-            let config = match parse_optional_json_as::<kb_onchain_transport::WsAccountSubscribeConfig>(
-                config_json,
-                "configJson",
-            ) {
+            let config = match parse_optional_json_as::<
+                kb_onchain_transport::WsAccountSubscribeConfig,
+            >(config_json, "configJson")
+            {
                 std::result::Result::Ok(config) => config,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
@@ -644,11 +660,13 @@ fn build_standard_ws_request(
             if let std::result::Result::Err(error) = reject_present(target, "target", method) {
                 return std::result::Result::Err(error);
             }
-            let filter =
-                match parse_required_json_as::<kb_onchain_transport::WsBlockFilter>(filter_json, "filterJson") {
-                    std::result::Result::Ok(filter) => filter,
-                    std::result::Result::Err(error) => return std::result::Result::Err(error),
-                };
+            let filter = match parse_required_json_as::<kb_onchain_transport::WsBlockFilter>(
+                filter_json,
+                "filterJson",
+            ) {
+                std::result::Result::Ok(filter) => filter,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
             let config = match parse_optional_json_as::<kb_onchain_transport::WsBlockSubscribeConfig>(
                 config_json,
                 "configJson",
@@ -664,11 +682,13 @@ fn build_standard_ws_request(
             if let std::result::Result::Err(error) = reject_present(target, "target", method) {
                 return std::result::Result::Err(error);
             }
-            let filter =
-                match parse_required_json_as::<kb_onchain_transport::WsLogsFilter>(filter_json, "filterJson") {
-                    std::result::Result::Ok(filter) => filter,
-                    std::result::Result::Err(error) => return std::result::Result::Err(error),
-                };
+            let filter = match parse_required_json_as::<kb_onchain_transport::WsLogsFilter>(
+                filter_json,
+                "filterJson",
+            ) {
+                std::result::Result::Ok(filter) => filter,
+                std::result::Result::Err(error) => return std::result::Result::Err(error),
+            };
             let config = match parse_optional_json_as::<kb_onchain_transport::WsLogsSubscribeConfig>(
                 config_json,
                 "configJson",
@@ -676,10 +696,9 @@ fn build_standard_ws_request(
                 std::result::Result::Ok(config) => config,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
-            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Logs(kb_onchain_transport::LogsSubscribeRequest {
-                filter,
-                config,
-            }))
+            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Logs(
+                kb_onchain_transport::LogsSubscribeRequest { filter, config },
+            ))
         },
         "programSubscribe" => {
             let target = match required_target(target, method) {
@@ -691,10 +710,10 @@ fn build_standard_ws_request(
             {
                 return std::result::Result::Err(error);
             }
-            let config = match parse_optional_json_as::<kb_onchain_transport::WsProgramSubscribeConfig>(
-                config_json,
-                "configJson",
-            ) {
+            let config = match parse_optional_json_as::<
+                kb_onchain_transport::WsProgramSubscribeConfig,
+            >(config_json, "configJson")
+            {
                 std::result::Result::Ok(config) => config,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
@@ -712,10 +731,10 @@ fn build_standard_ws_request(
             {
                 return std::result::Result::Err(error);
             }
-            let config = match parse_optional_json_as::<kb_onchain_transport::WsSignatureSubscribeConfig>(
-                config_json,
-                "configJson",
-            ) {
+            let config = match parse_optional_json_as::<
+                kb_onchain_transport::WsSignatureSubscribeConfig,
+            >(config_json, "configJson")
+            {
                 std::result::Result::Ok(config) => config,
                 std::result::Result::Err(error) => return std::result::Result::Err(error),
             };
@@ -729,7 +748,9 @@ fn build_standard_ws_request(
             {
                 return std::result::Result::Err(error);
             }
-            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Root(kb_onchain_transport::RootSubscribeRequest))
+            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Root(
+                kb_onchain_transport::RootSubscribeRequest,
+            ))
         },
         "slotSubscribe" => {
             if let std::result::Result::Err(error) =
@@ -737,7 +758,9 @@ fn build_standard_ws_request(
             {
                 return std::result::Result::Err(error);
             }
-            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Slot(kb_onchain_transport::SlotSubscribeRequest))
+            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Slot(
+                kb_onchain_transport::SlotSubscribeRequest,
+            ))
         },
         "slotsUpdatesSubscribe" => {
             if let std::result::Result::Err(error) =
@@ -755,7 +778,9 @@ fn build_standard_ws_request(
             {
                 return std::result::Result::Err(error);
             }
-            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Vote(kb_onchain_transport::VoteSubscribeRequest))
+            std::result::Result::Ok(kb_onchain_transport::StandardWsRequest::Vote(
+                kb_onchain_transport::VoteSubscribeRequest,
+            ))
         },
         _ => std::result::Result::Err(format!("unsupported standard WebSocket method '{method}'")),
     };

kb-app-demo-desktop/src/frontend_log.rs

@@ -1,5 +1,5 @@
 // file: kb-app-demo-desktop/src/frontend_log.rs
-// version: 3
+// version: 4
 
 //! Frontend logging bridge used by Tauri WebView scripts.
 
@@ -23,11 +23,21 @@ pub(crate) struct FrontendLogPayload {
 pub(crate) fn emit_frontend_log(payload: crate::FrontendLogPayload) {
     let target = normalize_frontend_target(payload.target.as_str());
     match payload.level.trim().to_ascii_lowercase().as_str() {
-        "trace" => tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message),
-        "debug" => tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message),
-        "warn" => tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message),
-        "error" => tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message),
-        _ => tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message),
+        "trace" => {
+            tracing::trace!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message)
+        },
+        "debug" => {
+            tracing::debug!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message)
+        },
+        "warn" => {
+            tracing::warn!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message)
+        },
+        "error" => {
+            tracing::error!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message)
+        },
+        _ => {
+            tracing::info!(target: crate::TRACING_TARGET, action = "frontend_log", frontend_target = target, "{}", payload.message)
+        },
     }
 }
 
@@ -48,6 +58,12 @@ fn normalize_frontend_target(target: &str) -> std::string::String {
     if trimmed == "kb-app-demo-desktop.frontend.demo_ws" {
         return trimmed.to_string();
     }
+    if trimmed == "kb-app-demo-desktop.frontend.demo_core_extraction" {
+        return trimmed.to_string();
+    }
+    if trimmed == "kb-app-demo-desktop.frontend.demo_decode_replay" {
+        return trimmed.to_string();
+    }
     return "kb-app-demo-desktop.frontend".to_string();
 }
 
@@ -82,6 +98,13 @@ mod tests {
             super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_ws"),
             "kb-app-demo-desktop.frontend.demo_ws"
         );
+        assert_eq!(
+            super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_core_extraction"),
+            "kb-app-demo-desktop.frontend.demo_core_extraction"
+        );
+        assert_eq!(
+            super::normalize_frontend_target("kb-app-demo-desktop.frontend.demo_decode_replay"),
+            "kb-app-demo-desktop.frontend.demo_decode_replay"
+        );
     }
-
 }

kb-app-demo-desktop/src/lib.rs

@@ -1,5 +1,5 @@
 // file: kb-app-demo-desktop/src/lib.rs
-// version: 6
+// version: 7
 
 //! Tauri desktop demo application for `khadhroony-bot3`.
 
@@ -11,13 +11,15 @@ mod app_state;
 mod constants;
 mod demo_backfill;
 mod demo_config;
+mod demo_core_extraction;
+mod demo_decode_replay;
 mod demo_http;
-mod demo_ws;
 mod demo_sql_common;
 mod demo_sql_diag;
 mod demo_sql_pg_core;
 mod demo_sql_pg_raw;
 mod demo_sql_replay_candidates;
+mod demo_ws;
 mod frontend_log;
 mod main_window;
 mod splash;
@@ -68,36 +70,6 @@ pub(crate) use self::demo_http::build_http_method_options;
 pub(crate) use self::demo_http::build_http_role_options;
 /// Executes one raw HTTP JSON-RPC request.
 pub(crate) use self::demo_http::demo_http_execute_request_inner;
-/// WebSocket execution response payload.
-pub(crate) use self::demo_ws::DemoWsExecutionPayload;
-/// WebSocket message emitted to the frontend.
-pub(crate) use self::demo_ws::DemoWsMessagePayload;
-/// One selectable WebSocket method.
-pub(crate) use self::demo_ws::DemoWsMethodOption;
-/// WebSocket demo options payload.
-pub(crate) use self::demo_ws::DemoWsOptionsPayload;
-/// WebSocket subscription request.
-pub(crate) use self::demo_ws::DemoWsRequest;
-/// One selectable WebSocket role.
-pub(crate) use self::demo_ws::DemoWsRoleOption;
-/// Current WebSocket session status.
-pub(crate) use self::demo_ws::DemoWsStatusPayload;
-/// One active WebSocket subscription status.
-pub(crate) use self::demo_ws::DemoWsSubscriptionStatusPayload;
-/// WebSocket unsubscribe request.
-pub(crate) use self::demo_ws::DemoWsUnsubscribeRequest;
-/// Builds the standard WebSocket method inventory.
-pub(crate) use self::demo_ws::build_ws_method_options;
-/// Builds selectable WebSocket roles from pool snapshots.
-pub(crate) use self::demo_ws::build_ws_role_options;
-/// Connects and subscribes through the persistent WebSocket session.
-pub(crate) use self::demo_ws::demo_ws_connect_inner;
-/// Disconnects the persistent WebSocket session.
-pub(crate) use self::demo_ws::demo_ws_disconnect_inner;
-/// Returns the current WebSocket session status.
-pub(crate) use self::demo_ws::demo_ws_status_inner;
-/// Unsubscribes one active WebSocket subscription.
-pub(crate) use self::demo_ws::demo_ws_unsubscribe_inner;
 /// UI-safe SQL table diagnostic snapshot.
 pub(crate) use self::demo_sql_common::DemoSqlTableSnapshot;
 /// Connects to the configured PostgreSQL store.
@@ -152,6 +124,36 @@ pub(crate) use self::demo_sql_replay_candidates::program_scope_from_code;
 pub(crate) use self::demo_sql_replay_candidates::transaction_row_from_pg;
 /// Validates a CSV filename.
 pub(crate) use self::demo_sql_replay_candidates::validated_csv_file_name;
+/// WebSocket execution response payload.
+pub(crate) use self::demo_ws::DemoWsExecutionPayload;
+/// WebSocket message emitted to the frontend.
+pub(crate) use self::demo_ws::DemoWsMessagePayload;
+/// One selectable WebSocket method.
+pub(crate) use self::demo_ws::DemoWsMethodOption;
+/// WebSocket demo options payload.
+pub(crate) use self::demo_ws::DemoWsOptionsPayload;
+/// WebSocket subscription request.
+pub(crate) use self::demo_ws::DemoWsRequest;
+/// One selectable WebSocket role.
+pub(crate) use self::demo_ws::DemoWsRoleOption;
+/// Current WebSocket session status.
+pub(crate) use self::demo_ws::DemoWsStatusPayload;
+/// One active WebSocket subscription status.
+pub(crate) use self::demo_ws::DemoWsSubscriptionStatusPayload;
+/// WebSocket unsubscribe request.
+pub(crate) use self::demo_ws::DemoWsUnsubscribeRequest;
+/// Builds the standard WebSocket method inventory.
+pub(crate) use self::demo_ws::build_ws_method_options;
+/// Builds selectable WebSocket roles from pool snapshots.
+pub(crate) use self::demo_ws::build_ws_role_options;
+/// Connects and subscribes through the persistent WebSocket session.
+pub(crate) use self::demo_ws::demo_ws_connect_inner;
+/// Disconnects the persistent WebSocket session.
+pub(crate) use self::demo_ws::demo_ws_disconnect_inner;
+/// Returns the current WebSocket session status.
+pub(crate) use self::demo_ws::demo_ws_status_inner;
+/// Unsubscribes one active WebSocket subscription.
+pub(crate) use self::demo_ws::demo_ws_unsubscribe_inner;
 /// Frontend logging payload.
 pub(crate) use self::frontend_log::FrontendLogPayload;
 /// Emits one normalized frontend log event.
@@ -173,6 +175,64 @@ pub(crate) use self::splash::emit_splash_order;
 /// Waits for the minimum splash duration.
 pub(crate) use self::splash::wait_until_minimum;
 
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionObserver;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionOptionsPayload;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionProgressPayload;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionRequest;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionRunGuard;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::DemoCoreExtractionSummaryPayload;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::build_demo_core_extraction_pipeline_request;
+/// Internal demo core extraction item.
+pub(crate) use self::demo_core_extraction::demo_core_extraction_summary_payload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeCoverageSummaryPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeDiagnosticsPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeProcessorSummaryPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayDecoderOption;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayObserver;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayOptionsPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayProgressPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayRequest;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplayRunGuard;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoDecodeReplaySummaryPayload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoTransactionAnnotationRequest;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::DemoTransactionAnnotationRow;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::annotation_payload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::available_decoders;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::available_materializers;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::build_demo_decode_replay_pipeline_request;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::coverage_payload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::demo_decode_replay_summary_payload;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::optional_line_count;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::register_active_campaign;
+/// Internal demo decode replay item.
+pub(crate) use self::demo_decode_replay::text_sample;
 // Keep the canonical tracing target as the final facade export.
 /// Canonical tracing target for the desktop demo application.
 pub(crate) use self::constants::TRACING_TARGET;

kb-app-demo-desktop/src/tauri.rs

@@ -1,5 +1,5 @@
 // file: kb-app-demo-desktop/src/tauri.rs
-// version: 8
+// version: 9
 
 //! Tauri runtime assembly and private command wrappers.
 
@@ -64,6 +64,16 @@ pub fn run() -> kb_core::Result<()> {
         load_demo_sql_replay_programs,
         load_demo_sql_replay_entities,
         export_demo_sql_replay_csv,
+        open_demo_core_extraction_window,
+        demo_core_extraction_options,
+        demo_core_extraction_execute,
+        demo_core_extraction_cancel,
+        open_demo_decode_replay_window,
+        demo_decode_replay_options,
+        demo_decode_replay_execute,
+        demo_decode_replay_cancel,
+        demo_decode_replay_diagnostics,
+        demo_decode_replay_annotations,
         open_demo_config_window,
         load_demo_config,
     ]);
@@ -164,9 +174,9 @@ pub fn run() -> kb_core::Result<()> {
     let run_result = builder.run(tauri::generate_context!());
     return match run_result {
         std::result::Result::Ok(()) => std::result::Result::Ok(()),
-        std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri(format!(
-            "cannot run desktop demo application: {error:?}"
-        ))),
+        std::result::Result::Err(error) => std::result::Result::Err(kb_core::Error::tauri(
+            format!("cannot run desktop demo application: {error:?}"),
+        )),
     };
 }
 
@@ -182,6 +192,436 @@ fn install_default_rustls_provider() -> kb_core::Result<()> {
         )),
     };
 }
+/// Opens or focuses the canonical to core extraction demo window.
+#[tauri::command]
+fn open_demo_core_extraction_window(
+    app_handle: tauri::AppHandle,
+) -> std::result::Result<(), std::string::String> {
+    tracing::info!(target: crate::TRACING_TARGET, "open core extraction demo window");
+    let existing_window = app_handle.get_webview_window("demo_core_extraction");
+    if let std::option::Option::Some(window) = existing_window {
+        let show_result = window.show();
+        if let std::result::Result::Err(error) = show_result {
+            return std::result::Result::Err(error.to_string());
+        }
+        let focus_result = window.set_focus();
+        if let std::result::Result::Err(error) = focus_result {
+            return std::result::Result::Err(error.to_string());
+        }
+        return std::result::Result::Ok(());
+    }
+    let build_result = tauri::WebviewWindowBuilder::new(
+        &app_handle,
+        "demo_core_extraction",
+        tauri::WebviewUrl::App("demo_core_extraction.html".into()),
+    )
+    .title("Khadhroony Bot2 - Extraction canonical vers core")
+    .inner_size(1280.0, 860.0)
+    .min_inner_size(960.0, 620.0)
+    .resizable(true)
+    .visible(true)
+    .build();
+    return match build_result {
+        std::result::Result::Ok(window) => {
+            let focus_result = window.set_focus();
+            if let std::result::Result::Err(error) = focus_result {
+                return std::result::Result::Err(error.to_string());
+            }
+            std::result::Result::Ok(())
+        },
+        std::result::Result::Err(error) => std::result::Result::Err(error.to_string()),
+    };
+}
+
+/// Returns the initial core extraction demo options.
+#[tauri::command]
+fn demo_core_extraction_options(
+    state: tauri::State<'_, crate::AppState>,
+) -> crate::DemoCoreExtractionOptionsPayload {
+    return crate::DemoCoreExtractionOptionsPayload {
+        processor_version: kb_pipeline::CORE_EXTRACTION_PROCESSOR_VERSION.to_string(),
+        default_limit: 100,
+        default_max_concurrent_extractions: 4,
+        running: state.demo_core_extraction_running().load(std::sync::atomic::Ordering::Acquire),
+    };
+}
+
+/// Executes one bounded canonical transaction to core extraction campaign.
+#[tauri::command]
+async fn demo_core_extraction_execute(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, crate::AppState>,
+    request: crate::DemoCoreExtractionRequest,
+) -> std::result::Result<crate::DemoCoreExtractionSummaryPayload, std::string::String> {
+    let acquire_result = state.demo_core_extraction_running().compare_exchange(
+        false,
+        true,
+        std::sync::atomic::Ordering::AcqRel,
+        std::sync::atomic::Ordering::Acquire,
+    );
+    if acquire_result.is_err() {
+        return std::result::Result::Err(
+            "a core extraction campaign is already running".to_string(),
+        );
+    }
+    let _run_guard = crate::DemoCoreExtractionRunGuard {
+        running: state.demo_core_extraction_running(),
+    };
+    state
+        .demo_core_extraction_cancel_requested()
+        .store(false, std::sync::atomic::Ordering::Release);
+    let pipeline_request_result = crate::build_demo_core_extraction_pipeline_request(request);
+    let pipeline_request = match pipeline_request_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let store_result = crate::connect_postgres_store(state.active_profile()).await;
+    let store = match store_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let observer = crate::DemoCoreExtractionObserver {
+        app_handle,
+        cancel_requested: state.demo_core_extraction_cancel_requested(),
+    };
+    let summary_result =
+        kb_pipeline::execute_core_extraction(&store, &pipeline_request, &observer).await;
+    let summary = match summary_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    return std::result::Result::Ok(crate::demo_core_extraction_summary_payload(summary));
+}
+
+/// Requests cooperative cancellation of the current core extraction campaign.
+#[tauri::command]
+fn demo_core_extraction_cancel(state: tauri::State<'_, crate::AppState>) -> bool {
+    let running = state.demo_core_extraction_running().load(std::sync::atomic::Ordering::Acquire);
+    state
+        .demo_core_extraction_cancel_requested()
+        .store(true, std::sync::atomic::Ordering::Release);
+    return running;
+}
+
+/// Opens or focuses the contextual decode replay window.
+#[tauri::command]
+fn open_demo_decode_replay_window(
+    app_handle: tauri::AppHandle,
+) -> std::result::Result<(), std::string::String> {
+    tracing::info!(target: crate::TRACING_TARGET, action = "open_window", window = "demo_decode_replay", "open contextual decode replay window");
+    let existing_window = app_handle.get_webview_window("demo_decode_replay");
+    if let std::option::Option::Some(window) = existing_window {
+        let show_result = window.show();
+        if let std::result::Result::Err(error) = show_result {
+            return std::result::Result::Err(error.to_string());
+        }
+        let focus_result = window.set_focus();
+        if let std::result::Result::Err(error) = focus_result {
+            return std::result::Result::Err(error.to_string());
+        }
+        return std::result::Result::Ok(());
+    }
+    let build_result = tauri::WebviewWindowBuilder::new(
+        &app_handle,
+        "demo_decode_replay",
+        tauri::WebviewUrl::App("demo_decode_replay.html".into()),
+    )
+    .title("Khadhroony Bot2 - Décodage et matérialisation")
+    .inner_size(1320.0, 900.0)
+    .min_inner_size(980.0, 660.0)
+    .resizable(true)
+    .visible(true)
+    .build();
+    return match build_result {
+        std::result::Result::Ok(window) => {
+            let focus_result = window.set_focus();
+            if let std::result::Result::Err(error) = focus_result {
+                return std::result::Result::Err(error.to_string());
+            }
+            std::result::Result::Ok(())
+        },
+        std::result::Result::Err(error) => std::result::Result::Err(error.to_string()),
+    };
+}
+
+/// Returns available decoders and default replay bounds.
+#[tauri::command]
+fn demo_decode_replay_options(
+    state: tauri::State<'_, crate::AppState>,
+) -> crate::DemoDecodeReplayOptionsPayload {
+    let decoders = crate::available_decoders();
+    let decoder_options = decoders
+        .iter()
+        .map(|decoder| {
+            let identity = decoder.identity();
+            return crate::DemoDecodeReplayDecoderOption {
+                name: identity.name,
+                version: identity.version,
+                program_ids: decoder
+                    .surfaces()
+                    .iter()
+                    .map(|surface| return surface.program_id.to_string())
+                    .collect(),
+            };
+        })
+        .collect::<std::vec::Vec<_>>();
+    let materializers = crate::available_materializers();
+    let materializer_names = materializers
+        .iter()
+        .map(|materializer| {
+            let identity = materializer.identity();
+            return format!("{}@{}", identity.name, identity.version);
+        })
+        .collect::<std::vec::Vec<_>>();
+    let running = state.demo_decode_replay_running().load(std::sync::atomic::Ordering::Acquire);
+    tracing::debug!(
+        target: crate::TRACING_TARGET,
+        action = "load_options",
+        pipeline_version = kb_pipeline::DECODE_PIPELINE_VERSION,
+        decoder_count = decoder_options.len(),
+        materializer_names = ?materializer_names,
+        default_limit = 100_u32,
+        default_max_concurrent_inputs = 4_u32,
+        running,
+        "return contextual decode replay options"
+    );
+    return crate::DemoDecodeReplayOptionsPayload {
+        pipeline_version: kb_pipeline::DECODE_PIPELINE_VERSION.to_string(),
+        decoders: decoder_options,
+        materializer_names,
+        default_limit: 100,
+        default_max_concurrent_inputs: 4,
+        running,
+    };
+}
+
+/// Executes one bounded contextual decode and optional materialization campaign.
+#[tauri::command]
+async fn demo_decode_replay_execute(
+    app_handle: tauri::AppHandle,
+    state: tauri::State<'_, crate::AppState>,
+    request: crate::DemoDecodeReplayRequest,
+) -> std::result::Result<crate::DemoDecodeReplaySummaryPayload, std::string::String> {
+    let campaign_id = kb_pipeline::new_decode_campaign_id();
+    tracing::debug!(
+        target: crate::TRACING_TARGET,
+        action = "execute",
+        campaign_id = %campaign_id,
+        instruction_state = %request.instruction_state,
+        signature_line_count = crate::optional_line_count(&request.signatures_text),
+        program_id = ?request.program_id,
+        instruction_paths_text = ?request.instruction_paths_text,
+        decoder_names = ?request.decoder_names,
+        limit = request.limit,
+        max_concurrent_inputs = request.max_concurrent_inputs,
+        all_compatible = request.all_compatible,
+        force_replay = request.force_replay,
+        force_replay_all_matching = request.force_replay_all_matching,
+        materialize_after_decode = request.materialize_after_decode,
+        "received contextual decode replay command"
+    );
+    let acquire_result = state.demo_decode_replay_running().compare_exchange(
+        false,
+        true,
+        std::sync::atomic::Ordering::AcqRel,
+        std::sync::atomic::Ordering::Acquire,
+    );
+    if acquire_result.is_err() {
+        tracing::warn!(target: crate::TRACING_TARGET, action = "execute", campaign_id = %campaign_id, accepted = false, reason = "already_running", "reject contextual decode replay command");
+        return std::result::Result::Err(
+            "a contextual decode replay campaign is already running".to_string(),
+        );
+    }
+    let _run_guard = crate::DemoDecodeReplayRunGuard {
+        running: state.demo_decode_replay_running(),
+        campaign_id: state.demo_decode_replay_campaign_id(),
+    };
+    let register_result = crate::register_active_campaign(
+        state.demo_decode_replay_campaign_id(),
+        campaign_id.as_str(),
+    );
+    if let std::result::Result::Err(error) = register_result {
+        tracing::error!(target: crate::TRACING_TARGET, action = "register_campaign", campaign_id = %campaign_id, error = %error, "cannot lock active contextual decode campaign slot");
+        return std::result::Result::Err(error);
+    }
+    state
+        .demo_decode_replay_cancel_requested()
+        .store(false, std::sync::atomic::Ordering::Release);
+    let pipeline_request_result =
+        crate::build_demo_decode_replay_pipeline_request(request, campaign_id.clone());
+    let pipeline_request = match pipeline_request_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => {
+            tracing::error!(target: crate::TRACING_TARGET, action = "build_request", campaign_id = %campaign_id, error = %error, "cannot build contextual decode replay request");
+            return std::result::Result::Err(error);
+        },
+    };
+    tracing::debug!(
+        target: crate::TRACING_TARGET,
+        action = "execute_pipeline",
+        campaign_id = %pipeline_request.campaign_id,
+        signature_count = pipeline_request.selection.signatures.len(),
+        signature_sample = ?crate::text_sample(pipeline_request.selection.signatures.as_slice(), 5),
+        processing_states = ?pipeline_request.selection.processing_states,
+        min_slot = ?pipeline_request.selection.min_slot,
+        max_slot = ?pipeline_request.selection.max_slot,
+        program_ids = ?pipeline_request.selection.program_ids,
+        instruction_paths = ?pipeline_request.selection.instruction_paths,
+        incomplete_signatures = pipeline_request.selection.incomplete_signatures,
+        limit = pipeline_request.selection.limit,
+        decoder_names = ?pipeline_request.decoder_names,
+        dispatch_policy = ?pipeline_request.dispatch_policy,
+        max_concurrent_inputs = pipeline_request.max_concurrent_inputs,
+        force_replay = pipeline_request.force_replay,
+        force_replay_all_matching = pipeline_request.force_replay_all_matching,
+        materialize_after_decode = pipeline_request.materialize_after_decode,
+        "start contextual decode replay pipeline"
+    );
+    let store_result = crate::connect_postgres_store(state.active_profile()).await;
+    let store = match store_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => {
+            tracing::error!(target: crate::TRACING_TARGET, action = "connect_store", campaign_id = %pipeline_request.campaign_id, error = %error, "cannot connect contextual decode replay store");
+            return std::result::Result::Err(error);
+        },
+    };
+    let decoders = crate::available_decoders();
+    let materializers = crate::available_materializers();
+    let observer = crate::DemoDecodeReplayObserver {
+        app_handle,
+        campaign_id: pipeline_request.campaign_id.clone(),
+        cancel_requested: state.demo_decode_replay_cancel_requested(),
+    };
+    let summary_result = kb_pipeline::execute_decode_replay(
+        &store,
+        &pipeline_request,
+        decoders.as_slice(),
+        materializers.as_slice(),
+        &observer,
+    )
+    .await;
+    let summary = match summary_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => {
+            tracing::error!(target: crate::TRACING_TARGET, action = "execute_pipeline", campaign_id = %pipeline_request.campaign_id, error = %error, "contextual decode replay pipeline failed");
+            return std::result::Result::Err(error.to_string());
+        },
+    };
+    tracing::debug!(
+        target: crate::TRACING_TARGET,
+        action = "execute_pipeline",
+        campaign_id = %summary.campaign_id,
+        selected = summary.selected,
+        started = summary.started,
+        completed = summary.completed,
+        unmatched = summary.unmatched,
+        not_started = summary.not_started,
+        failed_inputs = summary.failed_inputs,
+        cancelled = summary.cancelled,
+        processors = ?summary.processors,
+        "contextual decode replay command completed"
+    );
+    return std::result::Result::Ok(crate::demo_decode_replay_summary_payload(summary));
+}
+
+/// Requests cooperative cancellation of the current decode replay campaign.
+#[tauri::command]
+fn demo_decode_replay_cancel(state: tauri::State<'_, crate::AppState>) -> bool {
+    let running = state.demo_decode_replay_running().load(std::sync::atomic::Ordering::Acquire);
+    state
+        .demo_decode_replay_cancel_requested()
+        .store(true, std::sync::atomic::Ordering::Release);
+    let campaign_lock_result = state.demo_decode_replay_campaign_id().lock();
+    let campaign_id = match campaign_lock_result {
+        std::result::Result::Ok(active_campaign_id) => active_campaign_id.clone(),
+        std::result::Result::Err(_) => std::option::Option::None,
+    };
+    tracing::debug!(target: crate::TRACING_TARGET, action = "cancel", campaign_id = ?campaign_id, running, cancellation_requested = true, "contextual decode replay cancellation command handled");
+    return running;
+}
+
+/// Loads read-only decode, materialization, ledger and coverage diagnostics.
+#[tauri::command]
+async fn demo_decode_replay_diagnostics(
+    state: tauri::State<'_, crate::AppState>,
+) -> std::result::Result<crate::DemoDecodeDiagnosticsPayload, std::string::String> {
+    tracing::debug!(target: crate::TRACING_TARGET, action = "load_diagnostics", coverage_limit = 500_u32, "load contextual decode replay diagnostics");
+    let store_result = crate::connect_postgres_store(state.active_profile()).await;
+    let store = match store_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let table_result = store.known_table_diagnostics().await;
+    let all_tables = match table_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    let selected_tables: std::vec::Vec<crate::DemoSqlTableSnapshot> = all_tables
+        .iter()
+        .filter(|table| {
+            return table.table_name.starts_with("kb_sol_decode_")
+                || table.table_name.starts_with("kb_sol_mat_")
+                || table.table_name == "kb_sol_ops_processing_ledger";
+        })
+        .map(crate::table_snapshot_from_pg)
+        .collect();
+    let coverage_result = kb_store::DecodePipelineStore::list_decode_coverage_summary(
+        &store,
+        std::option::Option::None,
+        std::option::Option::None,
+        500,
+    )
+    .await;
+    let coverage = match coverage_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    tracing::debug!(target: crate::TRACING_TARGET, action = "load_diagnostics", table_count = selected_tables.len(), coverage_count = coverage.len(), "contextual decode replay diagnostics loaded");
+    return std::result::Result::Ok(crate::DemoDecodeDiagnosticsPayload {
+        tables: selected_tables,
+        coverage: coverage.into_iter().map(crate::coverage_payload).collect(),
+    });
+}
+
+/// Loads a bounded journal of committed SPL Memo transaction annotations.
+#[tauri::command]
+async fn demo_decode_replay_annotations(
+    state: tauri::State<'_, crate::AppState>,
+    request: crate::DemoTransactionAnnotationRequest,
+) -> std::result::Result<std::vec::Vec<crate::DemoTransactionAnnotationRow>, std::string::String> {
+    let filter_result = kb_store::MaterializedEventFilter::new(
+        std::option::Option::Some("transaction_annotations".to_string()),
+        std::option::Option::Some("transaction_annotation".to_string()),
+        request.signature_contains,
+        request.limit,
+    );
+    let filter = match filter_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    tracing::debug!(target: crate::TRACING_TARGET, action = "load_transaction_annotations", signature_contains = ?filter.signature_contains, limit = filter.limit, "load bounded committed transaction annotation journal");
+    let store_result = crate::connect_postgres_store(state.active_profile()).await;
+    let store = match store_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error),
+    };
+    let rows_result =
+        kb_store::DecodePipelineStore::list_materialized_events(&store, &filter).await;
+    let rows = match rows_result {
+        std::result::Result::Ok(value) => value,
+        std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
+    };
+    let mut output = std::vec::Vec::with_capacity(rows.len());
+    for row in rows {
+        let mapped = crate::annotation_payload(row);
+        match mapped {
+            std::result::Result::Ok(value) => output.push(value),
+            std::result::Result::Err(error) => return std::result::Result::Err(error),
+        }
+    }
+    tracing::debug!(target: crate::TRACING_TARGET, action = "load_transaction_annotations", row_count = output.len(), "bounded committed transaction annotation journal loaded");
+    return std::result::Result::Ok(output);
+}
 
 #[tauri::command]
 fn open_demo_config_window(
@@ -217,15 +657,11 @@ fn open_demo_config_window(
 }
 
 #[tauri::command]
-fn load_demo_config(
-    state: tauri::State<'_, crate::AppState>,
-) -> crate::DemoConfigPayload {
+fn load_demo_config(state: tauri::State<'_, crate::AppState>) -> crate::DemoConfigPayload {
     return crate::demo_config_payload(state.inner());
 }
 
-fn into_ipc_result<T>(
-    result: kb_core::Result<T>,
-) -> std::result::Result<T, std::string::String> {
+fn into_ipc_result<T>(result: kb_core::Result<T>) -> std::result::Result<T, std::string::String> {
     return match result {
         std::result::Result::Ok(value) => std::result::Result::Ok(value),
         std::result::Result::Err(error) => std::result::Result::Err(error.to_string()),
@@ -244,9 +680,7 @@ fn load_project_readme() -> std::result::Result<std::string::String, std::string
 
 #[tauri::command]
 fn demo_backfill_cancel(state: tauri::State<'_, crate::AppState>) -> bool {
-    let running = state
-        .demo_backfill_running()
-        .load(std::sync::atomic::Ordering::Acquire);
+    let running = state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire);
     state
         .demo_backfill_cancel_requested()
         .store(true, std::sync::atomic::Ordering::Release);
@@ -305,9 +739,7 @@ fn demo_backfill_options(
         default_max_pages: 20,
         default_max_concurrent_requests: 4,
         default_max_retries: 2,
-        running: state
-            .demo_backfill_running()
-            .load(std::sync::atomic::Ordering::Acquire),
+        running: state.demo_backfill_running().load(std::sync::atomic::Ordering::Acquire),
     };
 }
 
@@ -327,9 +759,7 @@ async fn demo_backfill_execute(
     if acquire_result.is_err() {
         return std::result::Result::Err("a backfill campaign is already running".to_string());
     }
-    let _run_guard = crate::DemoBackfillRunGuard {
-        running: state.demo_backfill_running(),
-    };
+    let _run_guard = crate::DemoBackfillRunGuard { running: state.demo_backfill_running() };
     state
         .demo_backfill_cancel_requested()
         .store(false, std::sync::atomic::Ordering::Release);
@@ -414,9 +844,7 @@ fn demo_http_list_pool_clients(
 }
 
 #[tauri::command]
-fn demo_http_options(
-    state: tauri::State<'_, crate::AppState>,
-) -> crate::DemoHttpOptionsPayload {
+fn demo_http_options(state: tauri::State<'_, crate::AppState>) -> crate::DemoHttpOptionsPayload {
     return crate::DemoHttpOptionsPayload {
         roles: crate::build_http_role_options(state.http_pool().snapshot()),
         methods: crate::build_http_method_options(),
@@ -480,7 +908,10 @@ fn open_demo_ws_window(
 #[tauri::command]
 fn demo_ws_list_pool_clients(
     state: tauri::State<'_, crate::AppState>,
-) -> std::result::Result<std::vec::Vec<kb_onchain_transport::WsPoolClientSnapshot>, std::string::String> {
+) -> std::result::Result<
+    std::vec::Vec<kb_onchain_transport::WsPoolClientSnapshot>,
+    std::string::String,
+> {
     let pool = match state.demo_ws_pool() {
         std::result::Result::Ok(value) => value,
         std::result::Result::Err(error) => return std::result::Result::Err(error.to_string()),
@@ -876,4 +1307,3 @@ async fn load_demo_sql_replay_entities(
     }
     return std::result::Result::Ok(output);
 }
-