duroxide 0.1.27

// SQLite provider: Mutex/lock operations should panic on poison
#![allow(clippy::expect_used)]
#![allow(clippy::unwrap_used)]

use sqlx::sqlite::{SqlitePool, SqlitePoolOptions};
use sqlx::{Row, Sqlite, Transaction};
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tracing::debug;

use super::{
    DeleteInstanceResult, DispatcherCapabilityFilter, ExecutionInfo, InstanceFilter, InstanceInfo, OrchestrationItem,
    Provider, ProviderAdmin, ProviderError, PruneOptions, PruneResult, QueueDepths, ScheduledActivityIdentifier,
    SessionFetchConfig, SystemMetrics, TagFilter, WorkItem,
};
use crate::{Event, EventKind};

/// Default limit for bulk operations when not specified by caller
const DEFAULT_BULK_OPERATION_LIMIT: u32 = 1000;

/// Configuration options for SQLiteProvider
#[derive(Debug, Clone, Default)]
pub struct SqliteOptions {
    // Currently empty - lock timeout moved to RuntimeOptions
    // Kept for future provider-specific options
}

/// SQLite-backed provider with full transactional support
///
/// This provider offers true ACID guarantees across all operations,
/// eliminating the race conditions present in the filesystem provider.
pub struct SqliteProvider {
    pool: SqlitePool,
}

impl SqliteProvider {
    /// Convert sqlx error to ProviderError with appropriate retry classification
    fn sqlx_to_provider_error(operation: &str, e: sqlx::Error) -> ProviderError {
        let error_msg = e.to_string();

        // Check for SQLITE_BUSY (database locked) - retryable
        if error_msg.contains("database is locked") || error_msg.contains("SQLITE_BUSY") {
            return ProviderError::retryable(operation, format!("Database locked: {error_msg}"));
        }

        // Check for constraint violations (duplicate events, etc.) - permanent
        if error_msg.contains("UNIQUE constraint") || error_msg.contains("PRIMARY KEY") {
            return ProviderError::permanent(operation, format!("Constraint violation: {error_msg}"));
        }

        // Check for connection errors - retryable
        if error_msg.contains("connection") || error_msg.contains("timeout") {
            return ProviderError::retryable(operation, format!("Connection error: {error_msg}"));
        }

        // Default: treat as retryable (conservative approach)
        ProviderError::retryable(operation, error_msg)
    }

    /// Internal method to enqueue orchestrator work with optional visibility delay
    async fn enqueue_orchestrator_work_with_delay(
        &self,
        item: WorkItem,
        delay: Option<Duration>,
    ) -> Result<(), ProviderError> {
        let work_item = serde_json::to_string(&item)
            .map_err(|e| ProviderError::permanent("enqueue_for_orchestrator", format!("Serialization error: {e}")))?;
        let instance = match &item {
            WorkItem::StartOrchestration { instance, .. }
            | WorkItem::ActivityCompleted { instance, .. }
            | WorkItem::ActivityFailed { instance, .. }
            | WorkItem::TimerFired { instance, .. }
            | WorkItem::ExternalRaised { instance, .. }
            | WorkItem::QueueMessage { instance, .. }
            | WorkItem::CancelInstance { instance, .. }
            | WorkItem::ContinueAsNew { instance, .. } => instance,
            #[cfg(feature = "replay-version-test")]
            WorkItem::ExternalRaised2 { instance, .. } => instance,
            WorkItem::SubOrchCompleted { parent_instance, .. } | WorkItem::SubOrchFailed { parent_instance, .. } => {
                parent_instance
            }
            _ => {
                return Err(ProviderError::permanent(
                    "enqueue_for_orchestrator",
                    "Invalid work item type",
                ));
            }
        };
        tracing::debug!(
            target: "duroxide::providers::sqlite",
            ?item,
            instance = %instance,
            delay = ?delay,
            "enqueue_orchestrator_work_with_delay"
        );

        // Calculate visible_at based on delay
        let visible_at = if let Some(delay) = delay {
            let delay_ms = delay.as_millis().min(i64::MAX as u128) as i64;
            Self::now_millis().saturating_add(delay_ms)
        } else {
            Self::now_millis()
        };

        sqlx::query("INSERT INTO orchestrator_queue (instance_id, work_item, visible_at) VALUES (?, ?, ?)")
            .bind(instance)
            .bind(work_item)
            .bind(visible_at)
            .execute(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("enqueue_for_orchestrator", e))?;

        Ok(())
    }

    /// Create a new SQLite provider
    ///
    /// # Arguments
    /// * `database_url` - SQLite connection string (e.g., "sqlite:data.db" or "sqlite::memory:")
    /// * `options` - Optional configuration (currently unused, kept for future options)
    ///
    /// # Errors
    ///
    /// Returns an error if database connection or schema initialization fails.
    pub async fn new(database_url: &str, _options: Option<SqliteOptions>) -> Result<Self, sqlx::Error> {
        // Configure SQLite for better concurrency
        let is_memory = database_url.contains(":memory:") || database_url.contains("mode=memory");
        let pool = SqlitePoolOptions::new()
            .max_connections(5)
            .after_connect(move |conn, _meta| {
                Box::pin({
                    let is_memory = is_memory;
                    async move {
                        // Journal mode: WAL for file DBs; MEMORY for in-memory DBs
                        if is_memory {
                            sqlx::query("PRAGMA journal_mode = MEMORY").execute(&mut *conn).await?;
                            // For in-memory DB, durability is not required
                            sqlx::query("PRAGMA synchronous = OFF").execute(&mut *conn).await?;
                        } else {
                            // Enable WAL mode for better concurrent access
                            sqlx::query("PRAGMA journal_mode = WAL").execute(&mut *conn).await?;
                            // Set synchronous mode to WAL for better performance with WAL mode
                            // WAL mode: only sync the WAL file, not the main database
                            sqlx::query("PRAGMA synchronous = WAL").execute(&mut *conn).await?;
                            // Increase WAL checkpoint interval for better write batching
                            sqlx::query("PRAGMA wal_autocheckpoint = 10000")
                                .execute(&mut *conn)
                                .await?;
                            // Increase cache size for better performance (64MB)
                            sqlx::query("PRAGMA cache_size = -64000").execute(&mut *conn).await?;
                        }

                        // Set busy timeout to 60 seconds to retry on locks
                        sqlx::query("PRAGMA busy_timeout = 60000").execute(&mut *conn).await?;

                        // Enable foreign keys
                        sqlx::query("PRAGMA foreign_keys = ON").execute(&mut *conn).await?;

                        Ok(())
                    }
                })
            })
            .connect(database_url)
            .await?;

        // If using in-memory database (for tests), create schema directly
        if database_url.contains(":memory:") || database_url.contains("mode=memory") {
            Self::create_schema(&pool).await?;
        } else {
            // For file-based databases, try migrations first, fall back to direct schema creation
            match sqlx::migrate!("./migrations").run(&pool).await {
                Ok(_) => {
                    tracing::debug!("Successfully ran migrations");
                }
                Err(e) => {
                    tracing::debug!("Migration failed: {}, falling back to create_schema", e);
                    // Migrations not available (e.g., in tests), create schema directly
                    Self::create_schema(&pool).await?;
                }
            }
        }

        Ok(Self { pool })
    }

    /// Convenience: create a shared in-memory SQLite store for tests
    /// Uses a shared cache so multiple pooled connections see the same DB
    ///
    /// # Errors
    ///
    /// Returns an error if database connection or schema initialization fails.
    pub async fn new_in_memory() -> Result<Self, sqlx::Error> {
        Self::new_in_memory_with_options(None).await
    }

    /// Create an in-memory SQLite store with custom options
    ///
    /// # Errors
    ///
    /// Returns an error if database connection or schema initialization fails.
    pub async fn new_in_memory_with_options(options: Option<SqliteOptions>) -> Result<Self, sqlx::Error> {
        // use shared-cache memory to allow pool > 1
        // ref: https://www.sqlite.org/inmemorydb.html
        let url = "sqlite::memory:?cache=shared";
        Self::new(url, options).await
    }

    /// Debug helper: dump current queue states and small samples
    /// Force a WAL checkpoint to ensure all changes are written to main database file
    ///
    /// # Errors
    ///
    /// Returns an error if the checkpoint operation fails.
    pub async fn checkpoint(&self) -> Result<(), sqlx::Error> {
        sqlx::query("PRAGMA wal_checkpoint(FULL)").execute(&self.pool).await?;
        Ok(())
    }

    pub async fn debug_dump(&self) -> String {
        let mut out = String::new();
        let mut conn = match self.pool.acquire().await {
            Ok(c) => c,
            Err(e) => return format!("<debug_dump: acquire error: {e}>"),
        };

        // Orchestrator queue count and sample
        if let Ok((cnt,)) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM orchestrator_queue")
            .fetch_one(&mut *conn)
            .await
        {
            out.push_str(&format!("orchestrator_queue.count = {cnt}\n"));
        }
        if let Ok(rows) = sqlx::query(
            r#"SELECT id, instance_id, lock_token, locked_until, work_item FROM orchestrator_queue ORDER BY id LIMIT 10"#
        ).fetch_all(&mut *conn).await {
            out.push_str("orchestrator_queue.sample:\n");
            for r in rows { let id: i64 = r.try_get("id").unwrap_or_default(); let inst: String = r.try_get("instance_id").unwrap_or_default(); let lock: Option<String> = r.try_get("lock_token").unwrap_or_default(); let until: Option<i64> = r.try_get("locked_until").unwrap_or_default(); let item: String = r.try_get("work_item").unwrap_or_default(); out.push_str(&format!("  id={id}, inst={inst}, lock={lock:?}, until={until:?}, item={item}\n")); }
        }

        // Worker queue count and sample
        if let Ok((cnt,)) = sqlx::query_as::<_, (i64,)>("SELECT COUNT(*) FROM worker_queue")
            .fetch_one(&mut *conn)
            .await
        {
            out.push_str(&format!("worker_queue.count = {cnt}\n"));
        }
        if let Ok(rows) =
            sqlx::query(r#"SELECT id, lock_token, locked_until, work_item FROM worker_queue ORDER BY id LIMIT 10"#)
                .fetch_all(&mut *conn)
                .await
        {
            out.push_str("worker_queue.sample:\n");
            for r in rows {
                let id: i64 = r.try_get("id").unwrap_or_default();
                let lock: Option<String> = r.try_get("lock_token").unwrap_or_default();
                let until: Option<i64> = r.try_get("locked_until").unwrap_or_default();
                let item: String = r.try_get("work_item").unwrap_or_default();
                out.push_str(&format!("  id={id}, lock={lock:?}, until={until:?}, item={item}\n"));
            }
        }

        out
    }

    /// Create schema directly (for in-memory databases)
    async fn create_schema(pool: &SqlitePool) -> Result<(), sqlx::Error> {
        // Create all tables
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS instances (
                instance_id TEXT PRIMARY KEY,
                orchestration_name TEXT NOT NULL,
                orchestration_version TEXT,
                current_execution_id INTEGER NOT NULL DEFAULT 1,
                custom_status TEXT,
                custom_status_version INTEGER NOT NULL DEFAULT 0,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                parent_instance_id TEXT REFERENCES instances(instance_id)
            )
            "#,
        )
        .execute(pool)
        .await?;

        // Index for efficient parent-child lookups during cascading delete
        sqlx::query(r#"CREATE INDEX IF NOT EXISTS idx_instances_parent ON instances(parent_instance_id)"#)
            .execute(pool)
            .await?;

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS executions (
                instance_id TEXT NOT NULL,
                execution_id INTEGER NOT NULL,
                status TEXT NOT NULL DEFAULT 'Running',
                output TEXT,
                duroxide_version_major INTEGER,
                duroxide_version_minor INTEGER,
                duroxide_version_patch INTEGER,
                started_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                completed_at TIMESTAMP,
                PRIMARY KEY (instance_id, execution_id)
            )
            "#,
        )
        .execute(pool)
        .await?;

        // Migration: Add output column if it doesn't exist (for existing databases)
        // Check if column exists first to avoid errors
        let column_exists: bool =
            sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM pragma_table_info('executions') WHERE name = 'output'")
                .fetch_one(pool)
                .await
                .unwrap_or(0)
                > 0;

        if !column_exists {
            sqlx::query("ALTER TABLE executions ADD COLUMN output TEXT")
                .execute(pool)
                .await?;
            debug!("Added output column to executions table");
        }

        // Migration: Add custom_status columns to instances table if they don't exist
        // (was previously on executions; moved to instances so status survives ContinueAsNew)
        let custom_status_on_instances: bool = sqlx::query_scalar::<_, i64>(
            "SELECT COUNT(*) FROM pragma_table_info('instances') WHERE name = 'custom_status'",
        )
        .fetch_one(pool)
        .await
        .unwrap_or(0)
            > 0;

        if !custom_status_on_instances {
            sqlx::query("ALTER TABLE instances ADD COLUMN custom_status TEXT")
                .execute(pool)
                .await?;
            sqlx::query("ALTER TABLE instances ADD COLUMN custom_status_version INTEGER NOT NULL DEFAULT 0")
                .execute(pool)
                .await?;
            debug!("Added custom_status columns to instances table");
        }

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS history (
                instance_id TEXT NOT NULL,
                execution_id INTEGER NOT NULL,
                event_id INTEGER NOT NULL,
                event_type TEXT NOT NULL,
                event_data TEXT NOT NULL,
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                PRIMARY KEY (instance_id, execution_id, event_id)
            )
            "#,
        )
        .execute(pool)
        .await?;

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS orchestrator_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                instance_id TEXT NOT NULL,
                work_item TEXT NOT NULL,
                visible_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                lock_token TEXT,
                locked_until TIMESTAMP,
                attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
            "#,
        )
        .execute(pool)
        .await?;

        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS worker_queue (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                work_item TEXT NOT NULL,
                visible_at INTEGER NOT NULL DEFAULT 0,
                lock_token TEXT,
                locked_until TIMESTAMP,
                attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count >= 0),
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
                instance_id TEXT,
                execution_id TEXT,
                activity_id INTEGER,
                session_id TEXT,
                tag TEXT
            )
            "#,
        )
        .execute(pool)
        .await?;

        // Instance-level locks for concurrent dispatcher coordination
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS instance_locks (
                instance_id TEXT PRIMARY KEY,
                lock_token TEXT NOT NULL,
                locked_until INTEGER NOT NULL,
                locked_at INTEGER NOT NULL
            )
            "#,
        )
        .execute(pool)
        .await?;

        // Create indexes
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_orch_visible ON orchestrator_queue(visible_at, lock_token)")
            .execute(pool)
            .await?;
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_orch_instance ON orchestrator_queue(instance_id)")
            .execute(pool)
            .await?;
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_orch_lock ON orchestrator_queue(lock_token)")
            .execute(pool)
            .await?;
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_worker_available ON worker_queue(lock_token, id)")
            .execute(pool)
            .await?;
        sqlx::query(
            "CREATE INDEX IF NOT EXISTS idx_worker_identity ON worker_queue(instance_id, execution_id, activity_id)",
        )
        .execute(pool)
        .await?;

        // Session routing index
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_worker_queue_session ON worker_queue(session_id)")
            .execute(pool)
            .await?;

        // Activity tag routing index
        sqlx::query("CREATE INDEX IF NOT EXISTS idx_worker_queue_tag ON worker_queue(tag)")
            .execute(pool)
            .await?;

        // Sessions table for worker-affinity routing
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS sessions (
                session_id     TEXT PRIMARY KEY,
                worker_id      TEXT NOT NULL,
                locked_until   INTEGER NOT NULL,
                last_activity_at INTEGER NOT NULL
            )
            "#,
        )
        .execute(pool)
        .await?;

        // KV store: instance-scoped key-value pairs, written at execution completion.
        // Contains the merged state from all prior completed executions.
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS kv_store (
                instance_id TEXT NOT NULL,
                key TEXT NOT NULL,
                value TEXT NOT NULL,
                last_updated_at_ms INTEGER NOT NULL DEFAULT 0,
                PRIMARY KEY (instance_id, key)
            )
            "#,
        )
        .execute(pool)
        .await?;

        // KV delta: mutations from the current execution only.
        // Written every turn (ack). Merged into kv_store on execution completion.
        // NULL value = tombstone (key was explicitly cleared).
        sqlx::query(
            r#"
            CREATE TABLE IF NOT EXISTS kv_delta (
                instance_id TEXT NOT NULL,
                key TEXT NOT NULL,
                value TEXT,
                last_updated_at_ms INTEGER NOT NULL,
                PRIMARY KEY (instance_id, key)
            )
            "#,
        )
        .execute(pool)
        .await?;

        Ok(())
    }

    /// Generate a unique lock token
    fn generate_lock_token() -> String {
        let now = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock should be after UNIX epoch")
            .as_nanos();
        format!("lock_{now}_{}", std::process::id())
    }

    /// Get current timestamp in milliseconds
    fn now_millis() -> i64 {
        SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .expect("system clock should be after UNIX epoch")
            .as_millis() as i64
    }

    /// Get future timestamp in milliseconds
    fn timestamp_after(duration: Duration) -> i64 {
        Self::now_millis() + duration.as_millis() as i64
    }

    /// Build a SQL WHERE clause fragment for tag filtering.
    ///
    /// Returns a SQL expression to be used as `AND ({clause})`.
    /// Uses positional parameters that must be bound AFTER any existing parameters
    /// in the query.
    fn build_tag_clause(filter: &TagFilter, start_param: usize) -> String {
        match filter {
            TagFilter::DefaultOnly => "q.tag IS NULL".to_string(),
            TagFilter::Tags(set) => {
                let placeholders: Vec<_> = (0..set.len()).map(|i| format!("?{}", start_param + i)).collect();
                format!("q.tag IN ({})", placeholders.join(", "))
            }
            TagFilter::DefaultAnd(set) => {
                let placeholders: Vec<_> = (0..set.len()).map(|i| format!("?{}", start_param + i)).collect();
                format!("(q.tag IS NULL OR q.tag IN ({}))", placeholders.join(", "))
            }
            TagFilter::Any => "1".to_string(),  // Always match
            TagFilter::None => "0".to_string(), // Never match
        }
    }

    /// Collect tag values for binding to positional parameters.
    /// Returns a stable-ordered Vec of tag strings to bind.
    fn collect_tag_values(filter: &TagFilter) -> Vec<String> {
        match filter {
            TagFilter::Tags(set) | TagFilter::DefaultAnd(set) => {
                let mut v: Vec<String> = set.iter().cloned().collect();
                v.sort(); // Stable ordering for deterministic bind order
                v
            }
            _ => Vec::new(),
        }
    }

    /// Read history within a transaction
    async fn read_history_in_tx(
        &self,
        tx: &mut Transaction<'_, Sqlite>,
        instance: &str,
        execution_id: Option<u64>,
    ) -> Result<Vec<Event>, sqlx::Error> {
        let execution_id = match execution_id {
            Some(id) => id as i64,
            None => {
                // Get latest execution
                sqlx::query_scalar::<_, i64>(
                    "SELECT COALESCE(MAX(execution_id), 1) FROM executions WHERE instance_id = ?",
                )
                .bind(instance)
                .fetch_one(&mut **tx)
                .await?
            }
        };

        let rows = sqlx::query(
            r#"
            SELECT event_data 
            FROM history 
            WHERE instance_id = ? AND execution_id = ?
            ORDER BY event_id
            "#,
        )
        .bind(instance)
        .bind(execution_id)
        .fetch_all(&mut **tx)
        .await?;

        let mut events = Vec::new();
        for (idx, row) in rows.iter().enumerate() {
            let event_data: String = row.try_get("event_data")?;
            let event: Event = serde_json::from_str::<Event>(&event_data).map_err(|e| {
                // Deserialization failures must NOT be silently dropped.
                // Unknown event types (e.g., from a newer duroxide version) produce a hard error
                // so the caller can surface it properly and the poison path can eventually terminate
                // the orchestration. See provider-capability-filtering.md requirement 2.
                sqlx::Error::Protocol(format!(
                    "Failed to deserialize history event at position {idx} for instance '{instance}' execution {execution_id}: {e}"
                ))
            })?;
            events.push(event);
        }

        Ok(events)
    }

    /// Load KV snapshot within an existing transaction.
    async fn get_kv_snapshot_in_tx(
        &self,
        tx: &mut Transaction<'_, Sqlite>,
        instance: &str,
    ) -> Result<std::collections::HashMap<String, crate::providers::KvEntry>, ProviderError> {
        let rows = sqlx::query("SELECT key, value, last_updated_at_ms FROM kv_store WHERE instance_id = ?")
            .bind(instance)
            .fetch_all(&mut **tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_kv_snapshot_in_tx", e))?;

        let mut map = std::collections::HashMap::new();
        for row in rows {
            let k: String = row
                .try_get("key")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_snapshot_in_tx", e))?;
            let v: String = row
                .try_get("value")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_snapshot_in_tx", e))?;
            let ts: i64 = row
                .try_get("last_updated_at_ms")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_snapshot_in_tx", e))?;
            map.insert(
                k,
                crate::providers::KvEntry {
                    value: v,
                    last_updated_at_ms: ts as u64,
                },
            );
        }
        Ok(map)
    }

    /// Append history within a transaction
    async fn append_history_in_tx(
        &self,
        tx: &mut Transaction<'_, Sqlite>,
        instance: &str,
        execution_id: u64,
        events: Vec<Event>,
    ) -> Result<(), sqlx::Error> {
        // Validate that runtime provided concrete event_ids
        // The provider must NOT generate event_ids - they come from the runtime
        for event in &events {
            if event.event_id() == 0 {
                return Err(sqlx::Error::Protocol("event_id must be set by runtime".into()));
            }
        }

        // Insert events
        for event in &events {
            let event_type = match &event.kind {
                EventKind::OrchestrationStarted { .. } => "OrchestrationStarted",
                EventKind::OrchestrationCompleted { .. } => "OrchestrationCompleted",
                EventKind::OrchestrationFailed { .. } => "OrchestrationFailed",
                EventKind::OrchestrationContinuedAsNew { .. } => "OrchestrationContinuedAsNew",
                EventKind::ActivityScheduled { .. } => "ActivityScheduled",
                EventKind::ActivityCompleted { .. } => "ActivityCompleted",
                EventKind::ActivityFailed { .. } => "ActivityFailed",
                EventKind::ActivityCancelRequested { .. } => "ActivityCancelRequested",
                EventKind::TimerCreated { .. } => "TimerCreated",
                EventKind::TimerFired { .. } => "TimerFired",
                EventKind::ExternalSubscribed { .. } => "ExternalSubscribed",
                EventKind::ExternalEvent { .. } => "ExternalEvent",
                EventKind::SubOrchestrationScheduled { .. } => "SubOrchestrationScheduled",
                EventKind::SubOrchestrationCompleted { .. } => "SubOrchestrationCompleted",
                EventKind::SubOrchestrationFailed { .. } => "SubOrchestrationFailed",
                EventKind::SubOrchestrationCancelRequested { .. } => "SubOrchestrationCancelRequested",
                EventKind::OrchestrationCancelRequested { .. } => "OrchestrationCancelRequested",
                EventKind::ExternalSubscribedCancelled { .. } => "ExternalSubscribedCancelled",
                EventKind::QueueSubscribed { .. } => "ExternalSubscribedPersistent",
                EventKind::QueueEventDelivered { .. } => "ExternalEventPersistent",
                EventKind::QueueSubscriptionCancelled { .. } => "ExternalSubscribedPersistentCancelled",
                EventKind::OrchestrationChained { .. } => "OrchestrationChained",
                EventKind::CustomStatusUpdated { .. } => "CustomStatusUpdated",
                EventKind::KeyValueSet { .. } => "KeyValueSet",
                EventKind::KeyValueCleared { .. } => "KeyValueCleared",
                EventKind::KeyValuesCleared => "KeyValuesCleared",
                #[cfg(feature = "replay-version-test")]
                EventKind::ExternalSubscribed2 { .. } => "ExternalSubscribed2",
                #[cfg(feature = "replay-version-test")]
                EventKind::ExternalEvent2 { .. } => "ExternalEvent2",
            };

            let event_data = serde_json::to_string(&event)
                .expect("Event serialization should never fail - this is a programming error");
            let event_id = event.event_id() as i64;

            sqlx::query(
                r#"
                INSERT INTO history (instance_id, execution_id, event_id, event_type, event_data)
                VALUES (?, ?, ?, ?, ?)
                "#,
            )
            .bind(instance)
            .bind(execution_id as i64)
            .bind(event_id)
            .bind(event_type)
            .bind(event_data)
            .execute(&mut **tx)
            .await?;
        }

        Ok(())
    }

    pub fn get_pool(&self) -> &sqlx::SqlitePool {
        &self.pool
    }
}

#[async_trait::async_trait]
impl Provider for SqliteProvider {
    fn name(&self) -> &str {
        "sqlite"
    }

    fn version(&self) -> &str {
        env!("CARGO_PKG_VERSION")
    }

    async fn fetch_orchestration_item(
        &self,
        lock_timeout: Duration,
        _poll_timeout: Duration,
        filter: Option<&DispatcherCapabilityFilter>,
    ) -> Result<Option<(OrchestrationItem, String, u32)>, ProviderError> {
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
        let now_ms = Self::now_millis();

        // Implementation ordering (capability filtering contract):
        //   1. Select candidate instance (with version filter applied via SQL)
        //   2. Acquire instance lock
        //   3. Load and deserialize history (via read_history_in_tx)
        //
        // History deserialization intentionally happens AFTER filtering. The capability
        // filter excludes incompatible executions at the SQL level so the provider never
        // loads or deserializes history events from an incompatible execution.

        // Find an instance that has visible messages AND is not locked (or lock expired).
        // When a capability filter is provided, also join instances+executions to filter
        // on the execution's pinned duroxide version before acquiring the lock.
        let row = if let Some(cap_filter) = filter {
            // Phase 1: single range only. Multi-range would need an OR chain in the query.
            let range = match cap_filter.supported_duroxide_versions.first() {
                Some(r) => r,
                None => {
                    // Empty supported_duroxide_versions = "supports nothing" → no candidate.
                    tx.commit()
                        .await
                        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
                    return Ok(None);
                }
            };

            // Packed integer comparison: major * 1_000_000 + minor * 1_000 + patch.
            // Works as long as minor and patch are < 1000, which is true in practice.
            let min_packed =
                range.min.major as i64 * 1_000_000 + range.min.minor as i64 * 1_000 + range.min.patch as i64;
            let max_packed =
                range.max.major as i64 * 1_000_000 + range.max.minor as i64 * 1_000 + range.max.patch as i64;

            sqlx::query(
                r#"
                SELECT q.instance_id
                FROM orchestrator_queue q
                LEFT JOIN instance_locks il ON q.instance_id = il.instance_id
                LEFT JOIN instances i ON q.instance_id = i.instance_id
                LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
                WHERE q.visible_at <= ?1
                  AND (il.instance_id IS NULL OR il.locked_until <= ?1)
                  AND (
                    e.duroxide_version_major IS NULL
                    OR (e.duroxide_version_major * 1000000 + e.duroxide_version_minor * 1000 + e.duroxide_version_patch) BETWEEN ?2 AND ?3
                  )
                ORDER BY q.id
                LIMIT 1
                "#,
            )
            .bind(now_ms)
            .bind(min_packed)
            .bind(max_packed)
            .fetch_optional(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?
        } else {
            // No filter — legacy behavior, return any available instance
            sqlx::query(
                r#"
                SELECT q.instance_id
                FROM orchestrator_queue q
                LEFT JOIN instance_locks il ON q.instance_id = il.instance_id
                WHERE q.visible_at <= ?1
                  AND (il.instance_id IS NULL OR il.locked_until <= ?1)
                ORDER BY q.id
                LIMIT 1
                "#,
            )
            .bind(now_ms)
            .fetch_optional(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?
        };

        if row.is_none() {
            // Check if there are any messages at all
            let msg_count: Option<i64> = sqlx::query_scalar("SELECT COUNT(*) FROM orchestrator_queue")
                .fetch_one(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
            tracing::debug!(
                target = "duroxide::providers::sqlite",
                total_messages=?msg_count,
                "No available instances"
            );
            tx.rollback().await.ok();
            return Ok(None);
        }

        let instance_id: String = row
            .ok_or_else(|| ProviderError::permanent("fetch_orchestration_item", "No instance found"))?
            .try_get("instance_id")
            .map_err(|e| {
                ProviderError::permanent("fetch_orchestration_item", format!("Failed to get instance_id: {e}"))
            })?;
        tracing::debug!(target="duroxide::providers::sqlite", instance_id=%instance_id, "Selected available instance");

        let lock_token = Self::generate_lock_token();
        let locked_until = Self::timestamp_after(lock_timeout);

        // Atomically acquire instance lock (INSERT OR REPLACE to handle expired locks)
        let lock_result = sqlx::query(
            r#"
            INSERT INTO instance_locks (instance_id, lock_token, locked_until, locked_at)
            VALUES (?1, ?2, ?3, ?4)
            ON CONFLICT(instance_id) DO UPDATE
            SET lock_token = ?2, locked_until = ?3, locked_at = ?4
            WHERE locked_until <= ?4
            "#,
        )
        .bind(&instance_id)
        .bind(&lock_token)
        .bind(locked_until)
        .bind(now_ms)
        .execute(&mut *tx)
        .await;

        match lock_result {
            Ok(result) => {
                let affected = result.rows_affected();
                tracing::debug!(target="duroxide::providers::sqlite", instance=%instance_id, rows_affected=affected, "Instance lock result");
                if affected == 0 {
                    // Failed to acquire lock (lock still held by another worker)
                    tracing::debug!(target="duroxide::providers::sqlite", instance=%instance_id, "Failed to acquire instance lock (already locked)");
                    tx.rollback().await.ok();
                    return Ok(None);
                }
            }
            Err(e) => {
                tracing::debug!(target="duroxide::providers::sqlite", instance=%instance_id, error=%e.to_string(), "Error acquiring instance lock");
                tx.rollback().await.ok();
                return Err(Self::sqlx_to_provider_error("fetch_orchestration_item", e));
            }
        }

        // Fetch ALL messages for this instance and mark them with our lock_token for later deletion
        // We mark them so we can distinguish between messages we fetched vs new messages that arrive later
        // This includes messages that were previously fetched but the lock expired (they still have old lock_token)
        // We update ALL visible messages for this instance, regardless of their current lock_token
        // This allows recovery of messages that were fetched but the lock expired without ack
        // Also increment attempt_count for poison message detection
        let update_result = sqlx::query(
            r#"
            UPDATE orchestrator_queue
            SET lock_token = ?1, attempt_count = attempt_count + 1
            WHERE instance_id = ?2 AND visible_at <= ?3
            "#,
        )
        .bind(&lock_token)
        .bind(&instance_id)
        .bind(now_ms)
        .execute(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;

        let messages = sqlx::query(
            r#"
            SELECT id, work_item, attempt_count
            FROM orchestrator_queue
            WHERE lock_token = ?1
            ORDER BY id
            "#,
        )
        .bind(&lock_token)
        .fetch_all(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
        tracing::debug!(target="duroxide::providers::sqlite", message_count=%messages.len(), rows_updated=%update_result.rows_affected(), instance=%instance_id, "Fetched and marked messages for locked instance");

        if messages.is_empty() {
            // No messages for instance - this can happen if:
            // 1. Messages were already acked/deleted by another worker
            // 2. Race condition where messages were consumed between SELECT and UPDATE
            // Release lock and rollback
            sqlx::query("DELETE FROM instance_locks WHERE instance_id = ?")
                .bind(&instance_id)
                .execute(&mut *tx)
                .await
                .ok();
            tx.rollback().await.ok();
            return Ok(None);
        }

        // Deserialize work items and track max attempt_count
        let mut max_attempt_count: u32 = 0;
        let work_items: Vec<WorkItem> = messages
            .iter()
            .filter_map(|r| {
                // Track the max attempt_count across all messages in the batch
                if let Ok(attempt_count) = r.try_get::<i64, _>("attempt_count") {
                    max_attempt_count = max_attempt_count.max(attempt_count as u32);
                }
                r.try_get::<String, _>("work_item")
                    .ok()
                    .and_then(|s| serde_json::from_str(&s).ok())
            })
            .collect();

        // Get instance metadata
        let instance_info = sqlx::query(
            r#"
            SELECT i.orchestration_name, i.orchestration_version, i.current_execution_id
            FROM instances i
            WHERE i.instance_id = ?1
            "#,
        )
        .bind(&instance_id)
        .fetch_optional(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;

        let (orchestration_name, orchestration_version, current_execution_id, history) = if let Some(info) =
            instance_info
        {
            // Instance exists - get metadata and history for current execution
            let name: String = info.try_get("orchestration_name").map_err(|e| {
                ProviderError::permanent(
                    "fetch_orchestration_item",
                    format!("Failed to get orchestration_name: {e}"),
                )
            })?;
            let version: Option<String> = info.try_get("orchestration_version").ok();
            let exec_id: i64 = info.try_get("current_execution_id").map_err(|e| {
                ProviderError::permanent(
                    "fetch_orchestration_item",
                    format!("Failed to get current_execution_id: {e}"),
                )
            })?;

            // Normal case: always read history for current execution; runtime decides CAN semantics
            let hist_result = self
                .read_history_in_tx(&mut tx, &instance_id, Some(exec_id as u64))
                .await;

            match hist_result {
                Ok(hist) => {
                    // If version is NULL in database, use "unknown"
                    // If instance exists but version is NULL, there shouldn't be an OrchestrationStarted event
                    // in history (version should have been set when instance was created via metadata)
                    let version = version.unwrap_or_else(|| {
                        debug_assert!(
                            !hist
                                .iter()
                                .any(|e| matches!(&e.kind, EventKind::OrchestrationStarted { .. })),
                            "Instance exists with NULL version but history contains OrchestrationStarted event"
                        );
                        "unknown".to_string()
                    });

                    (name, version, exec_id as u64, hist)
                }
                Err(e) => {
                    // History deserialization failed. Return Ok(Some(...)) with
                    // history_error set so the runtime can reach its poison check.
                    // The lock is held (committed below) and attempt_count was
                    // already incremented, so the normal max_attempts cycle works.
                    let error_msg = format!("Failed to deserialize history: {e}");
                    tracing::warn!(
                        target = "duroxide::providers::sqlite",
                        instance = %instance_id,
                        error = %error_msg,
                        "History deserialization failed, returning item with history_error"
                    );

                    let version = version.unwrap_or_else(|| "unknown".to_string());

                    tx.commit()
                        .await
                        .map_err(|ce| Self::sqlx_to_provider_error("fetch_orchestration_item", ce))?;

                    return Ok(Some((
                        OrchestrationItem {
                            instance: instance_id,
                            orchestration_name: name,
                            execution_id: exec_id as u64,
                            version,
                            history: vec![],
                            messages: work_items,
                            history_error: Some(error_msg),
                            kv_snapshot: std::collections::HashMap::new(),
                        },
                        lock_token,
                        max_attempt_count,
                    )));
                }
            }
        } else {
            // Fallback: try to derive from history (e.g., ActivityCompleted arriving before we see instance row)
            let hist = self
                .read_history_in_tx(&mut tx, &instance_id, None)
                .await
                .unwrap_or_default();
            if let Some(first_started) = hist.iter().find_map(|e| {
                if let EventKind::OrchestrationStarted { name, version, .. } = &e.kind {
                    Some((name.clone(), version.clone()))
                } else {
                    None
                }
            }) {
                let (name, version) = first_started;
                (name, version, 1u64, hist)
            } else if let Some(start_item) = work_items.iter().find(|item| {
                matches!(
                    item,
                    WorkItem::StartOrchestration { .. } | WorkItem::ContinueAsNew { .. }
                )
            }) {
                // Brand new instance - find StartOrchestration or ContinueAsNew in work items
                // Version may be None - runtime will resolve and set via metadata
                let (orchestration, version) = match start_item {
                    WorkItem::StartOrchestration {
                        orchestration, version, ..
                    }
                    | WorkItem::ContinueAsNew {
                        orchestration, version, ..
                    } => (orchestration.clone(), version.clone()),
                    _ => unreachable!(),
                };
                (
                    orchestration,
                    version.unwrap_or_else(|| "unknown".to_string()),
                    1u64,
                    Vec::new(),
                )
            } else {
                // No instance info, no history, no StartOrchestration/ContinueAsNew.
                // Check if the batch contains only QueueMessage items (persistent events
                // enqueued before the orchestration started). These are orphans — drop them.
                // Other work items (CancelInstance, ActivityCompleted, etc.) may legitimately
                // race with StartOrchestration and must be kept in the queue.
                let all_queue_messages = work_items
                    .iter()
                    .all(|item| matches!(item, WorkItem::QueueMessage { .. }));
                if all_queue_messages {
                    let message_count = work_items.len();
                    tracing::warn!(
                        target="duroxide::providers::sqlite",
                        instance=%instance_id,
                        message_count,
                        "Dropping orphan queue messages — events enqueued before orchestration started are not supported"
                    );
                    // Messages are already marked with our lock_token; delete them.
                    sqlx::query("DELETE FROM orchestrator_queue WHERE lock_token = ?1")
                        .bind(&lock_token)
                        .execute(&mut *tx)
                        .await
                        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
                    // Release instance lock
                    sqlx::query("DELETE FROM instance_locks WHERE lock_token = ?1")
                        .bind(&lock_token)
                        .execute(&mut *tx)
                        .await
                        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
                    tx.commit()
                        .await
                        .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;
                    return Ok(None);
                }

                // Non-QueueMessage items racing with StartOrchestration — release locks
                // and leave messages in queue for the next poll.
                tracing::debug!(
                    target="duroxide::providers::sqlite",
                    instance=%instance_id,
                    message_count=work_items.len(),
                    "Work items without orchestration context; releasing for retry"
                );
                tx.rollback().await.ok();
                return Ok(None);
            }
        };

        // Load KV snapshot for seeding orchestration context
        let kv_snapshot = self.get_kv_snapshot_in_tx(&mut tx, &instance_id).await?;

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_orchestration_item", e))?;

        debug!(
            instance = %instance_id,
            messages = work_items.len(),
            history_len = history.len(),
            kv_keys = kv_snapshot.len(),
            "Fetched orchestration item"
        );

        Ok(Some((
            OrchestrationItem {
                instance: instance_id,
                orchestration_name,
                execution_id: current_execution_id,
                version: orchestration_version,
                messages: work_items,
                history,
                history_error: None,
                kv_snapshot,
            },
            lock_token,
            max_attempt_count,
        )))
    }

    async fn ack_orchestration_item(
        &self,
        lock_token: &str,
        execution_id: u64,
        history_delta: Vec<Event>,
        worker_items: Vec<WorkItem>,
        orchestrator_items: Vec<WorkItem>,
        metadata: crate::providers::ExecutionMetadata,
        cancelled_activities: Vec<ScheduledActivityIdentifier>,
    ) -> Result<(), ProviderError> {
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        // Get instance from instance_locks table (we now use instance-level locking)
        let row = sqlx::query("SELECT instance_id FROM instance_locks WHERE lock_token = ?")
            .bind(lock_token)
            .fetch_optional(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?
            .ok_or_else(|| ProviderError::permanent("ack_orchestration_item", "Invalid lock token"))?;

        let instance_id: String = row.try_get("instance_id").map_err(|e| {
            ProviderError::permanent("ack_orchestration_item", format!("Failed to decode instance_id: {e}"))
        })?;

        // Delete only the messages we fetched (marked with our lock_token)
        // New messages that arrived after fetch will remain in queue for next turn
        sqlx::query("DELETE FROM orchestrator_queue WHERE lock_token = ?")
            .bind(lock_token)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        debug!(
            instance = %instance_id,
            execution_id = %execution_id,
            history_delta_len = %history_delta.len(),
            "Acking with explicit execution_id"
        );

        // Create or update instance metadata from runtime-provided metadata
        // Runtime resolves version from registry and provides it via metadata
        if let (Some(name), Some(version)) = (&metadata.orchestration_name, &metadata.orchestration_version) {
            // First, ensure instance exists (INSERT OR IGNORE if new)
            // Include parent_instance_id for sub-orchestration tracking (used for cascading delete)
            sqlx::query(
                r#"
                INSERT OR IGNORE INTO instances
                (instance_id, orchestration_name, orchestration_version, current_execution_id, parent_instance_id)
                VALUES (?, ?, ?, ?, ?)
                "#,
            )
            .bind(&instance_id)
            .bind(name)
            .bind(version.as_str()) // version is &String, so use as_str()
            .bind(execution_id as i64)
            .bind(&metadata.parent_instance_id)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

            // Then update with resolved version (will update if instance exists, no-op if just created)
            // Note: parent_instance_id is immutable after creation, so we don't update it here
            sqlx::query(
                r#"
                UPDATE instances
                SET orchestration_name = ?, orchestration_version = ?
                WHERE instance_id = ?
                "#,
            )
            .bind(name)
            .bind(version)
            .bind(&instance_id)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;
        }

        // Create execution record if it doesn't exist (idempotent).
        // INSERT OR IGNORE is correct here: the execution may already exist from a previous
        // ack cycle (e.g., second turn of the same execution). We only need to ensure the
        // row exists; we never want to overwrite an existing execution's status.
        sqlx::query(
            r#"
            INSERT OR IGNORE INTO executions (instance_id, execution_id, status)
            VALUES (?, ?, 'Running')
            "#,
        )
        .bind(&instance_id)
        .bind(execution_id as i64)
        .execute(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

        // If pinned duroxide version is provided, store it on the execution row.
        // The provider unconditionally updates when told — the runtime is responsible
        // for only setting this on the first turn (enforced via debug_assert in the
        // orchestration dispatcher).
        if let Some(pinned) = &metadata.pinned_duroxide_version {
            sqlx::query(
                r#"
                UPDATE executions
                SET duroxide_version_major = ?, duroxide_version_minor = ?, duroxide_version_patch = ?
                WHERE instance_id = ? AND execution_id = ?
                "#,
            )
            .bind(pinned.major as i64)
            .bind(pinned.minor as i64)
            .bind(pinned.patch as i64)
            .bind(&instance_id)
            .bind(execution_id as i64)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
        }

        // Update instances.current_execution_id if this is a newer execution
        sqlx::query(
            r#"
            UPDATE instances 
            SET current_execution_id = MAX(current_execution_id, ?)
            WHERE instance_id = ?
            "#,
        )
        .bind(execution_id as i64)
        .bind(&instance_id)
        .execute(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        // Always append history_delta to the specified execution
        if !history_delta.is_empty() {
            debug!(
                instance = %instance_id,
                events = history_delta.len(),
                first_event = ?history_delta.first().map(std::mem::discriminant),
                "Appending history delta"
            );
            self.append_history_in_tx(&mut tx, &instance_id, execution_id, history_delta.clone())
                .await
                .map_err(|e| {
                    ProviderError::permanent("ack_orchestration_item", format!("Failed to append history: {e}"))
                })?;
        }

        // Update execution status and output from pre-computed metadata (no event inspection!)
        // This runs regardless of whether there's a history delta — needed for cases like
        // poison termination of items with corrupted history (history_error) where we can't
        // write history events but still need to mark the execution as Failed.
        if let Some(status) = &metadata.status {
            let now_ms = Self::now_millis();
            sqlx::query(
                r#"
                UPDATE executions 
                SET status = ?, output = ?, completed_at = ?
                WHERE instance_id = ? AND execution_id = ?
                "#,
            )
            .bind(status)
            .bind(&metadata.output)
            .bind(now_ms)
            .bind(&instance_id)
            .bind(execution_id as i64)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

            debug!(
                instance = %instance_id,
                execution_id = %execution_id,
                status = %status,
                "Updated execution status and output from metadata"
            );
        }

        // Derive custom_status from history_delta events.
        // Custom status is now tracked via CustomStatusUpdated events in history.
        // The provider scans the delta for the last such event to update the column.
        let custom_status_from_delta = history_delta.iter().rev().find_map(|e| match &e.kind {
            EventKind::CustomStatusUpdated { status } => Some(status.clone()),
            _ => None,
        });

        match custom_status_from_delta {
            Some(Some(custom_status)) => {
                sqlx::query(
                    r#"
                    UPDATE instances 
                    SET custom_status = ?, custom_status_version = custom_status_version + 1
                    WHERE instance_id = ?
                    "#,
                )
                .bind(&custom_status)
                .bind(&instance_id)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

                debug!(
                    instance = %instance_id,
                    custom_status = %custom_status,
                    "Updated custom_status from history event"
                );
            }
            Some(None) => {
                sqlx::query(
                    r#"
                    UPDATE instances 
                    SET custom_status = NULL, custom_status_version = custom_status_version + 1
                    WHERE instance_id = ?
                    "#,
                )
                .bind(&instance_id)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

                debug!(
                    instance = %instance_id,
                    "Cleared custom_status from history event"
                );
            }
            None => {
                // No CustomStatusUpdated in delta — preserve existing value
            }
        }

        // Materialize KV mutations from history_delta into the kv_delta table.
        // kv_delta captures current-execution mutations only. On execution completion,
        // kv_delta is merged into kv_store and cleared.
        for event in &history_delta {
            match &event.kind {
                EventKind::KeyValueSet {
                    key,
                    value,
                    last_updated_at_ms,
                } => {
                    sqlx::query(
                        r#"
                        INSERT INTO kv_delta (instance_id, key, value, last_updated_at_ms)
                        VALUES (?, ?, ?, ?)
                        ON CONFLICT(instance_id, key)
                        DO UPDATE SET value = excluded.value, last_updated_at_ms = excluded.last_updated_at_ms
                        "#,
                    )
                    .bind(&instance_id)
                    .bind(key)
                    .bind(value)
                    .bind(*last_updated_at_ms as i64)
                    .execute(&mut *tx)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
                }
                EventKind::KeyValueCleared { key } => {
                    // Upsert tombstone (value = NULL)
                    sqlx::query(
                        r#"
                        INSERT INTO kv_delta (instance_id, key, value, last_updated_at_ms)
                        VALUES (?, ?, NULL, ?)
                        ON CONFLICT(instance_id, key)
                        DO UPDATE SET value = NULL, last_updated_at_ms = excluded.last_updated_at_ms
                        "#,
                    )
                    .bind(&instance_id)
                    .bind(key)
                    .bind(Self::now_millis())
                    .execute(&mut *tx)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
                }
                EventKind::KeyValuesCleared => {
                    let clear_ts = Self::now_millis();
                    // Tombstone all existing delta rows
                    sqlx::query("UPDATE kv_delta SET value = NULL, last_updated_at_ms = ? WHERE instance_id = ?")
                        .bind(clear_ts)
                        .bind(&instance_id)
                        .execute(&mut *tx)
                        .await
                        .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
                    // Tombstone kv_store keys not already represented in delta
                    sqlx::query(
                        r#"
                        INSERT OR IGNORE INTO kv_delta (instance_id, key, value, last_updated_at_ms)
                        SELECT instance_id, key, NULL, ? FROM kv_store WHERE instance_id = ?
                        "#,
                    )
                    .bind(clear_ts)
                    .bind(&instance_id)
                    .execute(&mut *tx)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
                }
                _ => {}
            }
        }

        // On execution completion (terminal turn), merge kv_delta into kv_store
        // and clear kv_delta for this instance.
        let is_terminal = metadata
            .status
            .as_deref()
            .is_some_and(|s| s == "Completed" || s == "ContinuedAsNew" || s == "Failed");
        if is_terminal {
            // Apply non-tombstone values from delta → store
            sqlx::query(
                r#"
                INSERT INTO kv_store (instance_id, key, value, last_updated_at_ms)
                SELECT instance_id, key, value, last_updated_at_ms
                FROM kv_delta
                WHERE instance_id = ? AND value IS NOT NULL
                ON CONFLICT(instance_id, key)
                DO UPDATE SET value = excluded.value, last_updated_at_ms = excluded.last_updated_at_ms
                "#,
            )
            .bind(&instance_id)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

            // Apply tombstones: delete from kv_store where delta has NULL value
            sqlx::query(
                r#"
                DELETE FROM kv_store
                WHERE instance_id = ?
                AND key IN (SELECT key FROM kv_delta WHERE instance_id = ? AND value IS NULL)
                "#,
            )
            .bind(&instance_id)
            .bind(&instance_id)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;

            // Clear kv_delta for this instance
            sqlx::query("DELETE FROM kv_delta WHERE instance_id = ?")
                .bind(&instance_id)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
        }

        // Enqueue worker items with identity columns for cancellation support
        debug!(
            instance = %instance_id,
            count = worker_items.len(),
            "Enqueuing worker items"
        );
        let now_ms = Self::now_millis();
        for item in worker_items {
            // Extract identity fields from ActivityExecute
            let (activity_instance, activity_execution_id, activity_id, session_id, tag) = match &item {
                WorkItem::ActivityExecute {
                    instance,
                    execution_id,
                    id,
                    session_id,
                    tag,
                    ..
                } => (
                    Some(instance.as_str()),
                    Some(*execution_id),
                    Some(*id),
                    session_id.as_deref(),
                    tag.as_deref(),
                ),
                _ => (None, None, None, None, None),
            };

            let work_item = serde_json::to_string(&item).map_err(|e| {
                ProviderError::permanent("enqueue_for_orchestrator", format!("Serialization error: {e}"))
            })?;
            sqlx::query(
                r#"
                INSERT INTO worker_queue (work_item, visible_at, instance_id, execution_id, activity_id, session_id, tag)
                VALUES (?, ?, ?, ?, ?, ?, ?)
                "#,
            )
            .bind(work_item)
            .bind(now_ms)
            .bind(activity_instance)
            .bind(activity_execution_id.map(|e| e as i64))
            .bind(activity_id.map(|a| a as i64))
            .bind(session_id)
            .bind(tag)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;
        }

        // Delete cancelled activities (lock stealing) - batch operation
        if !cancelled_activities.is_empty() {
            debug!(
                instance = %instance_id,
                count = cancelled_activities.len(),
                "Cancelling activities via lock stealing"
            );

            // Build batch DELETE with VALUES clause
            let placeholders: Vec<String> = cancelled_activities
                .iter()
                .enumerate()
                .map(|(i, _)| format!("(?{}, ?{}, ?{})", i * 3 + 1, i * 3 + 2, i * 3 + 3))
                .collect();
            let sql = format!(
                "DELETE FROM worker_queue WHERE (instance_id, execution_id, activity_id) IN (VALUES {})",
                placeholders.join(", ")
            );

            let mut query = sqlx::query(&sql);
            for activity in &cancelled_activities {
                query = query
                    .bind(&activity.instance)
                    .bind(activity.execution_id as i64)
                    .bind(activity.activity_id as i64);
            }
            query
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_orchestration_item", e))?;
        }

        // Enqueue orchestrator items within the transaction
        for item in orchestrator_items {
            let work_item = serde_json::to_string(&item).map_err(|e| {
                ProviderError::permanent("enqueue_for_orchestrator", format!("Serialization error: {e}"))
            })?;
            let instance = match &item {
                WorkItem::StartOrchestration { instance, .. }
                | WorkItem::ActivityCompleted { instance, .. }
                | WorkItem::ActivityFailed { instance, .. }
                | WorkItem::TimerFired { instance, .. }
                | WorkItem::ExternalRaised { instance, .. }
                | WorkItem::QueueMessage { instance, .. }
                | WorkItem::CancelInstance { instance, .. }
                | WorkItem::ContinueAsNew { instance, .. } => instance,
                #[cfg(feature = "replay-version-test")]
                WorkItem::ExternalRaised2 { instance, .. } => instance,
                WorkItem::SubOrchCompleted { parent_instance, .. }
                | WorkItem::SubOrchFailed { parent_instance, .. } => parent_instance,
                _ => continue,
            };
            tracing::debug!(target = "duroxide::providers::sqlite", instance=%instance, ?item, "enqueue orchestrator item in ack");

            // Set visible_at based on item type
            // TimerFired uses fire_at_ms to delay visibility until timer should fire
            let visible_at = match &item {
                WorkItem::TimerFired { fire_at_ms, .. } => *fire_at_ms as i64,
                _ => Self::now_millis(),
            };

            sqlx::query("INSERT INTO orchestrator_queue (instance_id, work_item, visible_at) VALUES (?, ?, ?)")
                .bind(instance)
                .bind(work_item)
                .bind(visible_at)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;
        }

        // Validate instance lock is still valid before committing
        let now_ms = Self::now_millis();
        let lock_valid = sqlx::query_scalar::<_, i64>(
            r#"
            SELECT COUNT(*) FROM instance_locks 
            WHERE instance_id = ? AND lock_token = ? AND locked_until > ?
            "#,
        )
        .bind(&instance_id)
        .bind(lock_token)
        .bind(now_ms)
        .fetch_one(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        if lock_valid == 0 {
            // Lock expired or was stolen - abort transaction
            tracing::warn!(
                instance = %instance_id,
                lock_token = %lock_token,
                "Instance lock expired or invalid, aborting ack"
            );
            tx.rollback().await.ok();
            return Err(ProviderError::permanent(
                "ack_orchestration_item",
                "Instance lock expired",
            ));
        }

        // Remove instance lock (processing complete)
        sqlx::query("DELETE FROM instance_locks WHERE instance_id = ? AND lock_token = ?")
            .bind(&instance_id)
            .bind(lock_token)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_instances", e))?;

        debug!(
            instance = %instance_id,
            "Acknowledged orchestration item and released lock"
        );

        Ok(())
    }

    async fn read(&self, instance: &str) -> Result<Vec<Event>, ProviderError> {
        let mut conn = self
            .pool
            .acquire()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("read", e))?;

        let execution_id: i64 =
            sqlx::query_scalar::<_, i64>("SELECT COALESCE(MAX(execution_id), 1) FROM executions WHERE instance_id = ?")
                .bind(instance)
                .fetch_one(&mut *conn)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("read", e))?;

        let rows = sqlx::query(
            r#"
            SELECT event_data 
            FROM history 
            WHERE instance_id = ? AND execution_id = ?
            ORDER BY event_id
            "#,
        )
        .bind(instance)
        .bind(execution_id)
        .fetch_all(&mut *conn)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("read", e))?;

        let mut events = Vec::new();
        for row in rows {
            let event_data: String = row
                .try_get("event_data")
                .map_err(|e| ProviderError::permanent("read", format!("Failed to get event_data: {e}")))?;
            let event: Event = serde_json::from_str(&event_data)
                .map_err(|e| ProviderError::permanent("read", format!("Failed to deserialize event: {e}")))?;
            events.push(event);
        }

        Ok(events)
    }

    async fn read_with_execution(&self, instance: &str, execution_id: u64) -> Result<Vec<Event>, ProviderError> {
        let mut conn = self
            .pool
            .acquire()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("read_with_execution", e))?;

        let rows = sqlx::query(
            r#"
            SELECT event_data 
            FROM history 
            WHERE instance_id = ? AND execution_id = ?
            ORDER BY event_id
            "#,
        )
        .bind(instance)
        .bind(execution_id as i64)
        .fetch_all(&mut *conn)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("read_with_execution", e))?;

        let mut events = Vec::new();
        for row in rows {
            let event_data: String = row.try_get("event_data").map_err(|e| {
                ProviderError::permanent("read_with_execution", format!("Failed to get event_data: {e}"))
            })?;
            let event: Event = serde_json::from_str(&event_data).map_err(|e| {
                ProviderError::permanent("read_with_execution", format!("Failed to deserialize event: {e}"))
            })?;
            events.push(event);
        }

        Ok(events)
    }

    async fn append_with_execution(
        &self,
        instance: &str,
        execution_id: u64,
        new_events: Vec<Event>,
    ) -> Result<(), ProviderError> {
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("append_with_execution", e))?;

        self.append_history_in_tx(&mut tx, instance, execution_id, new_events)
            .await
            .map_err(|e| ProviderError::permanent("append_with_execution", format!("Failed to append history: {e}")))?;

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("append_with_execution", e))?;
        Ok(())
    }

    async fn enqueue_for_orchestrator(&self, item: WorkItem, delay: Option<Duration>) -> Result<(), ProviderError> {
        self.enqueue_orchestrator_work_with_delay(item, delay).await
    }

    async fn enqueue_for_worker(&self, item: WorkItem) -> Result<(), ProviderError> {
        tracing::debug!(target: "duroxide::providers::sqlite", ?item, "enqueue_for_worker");

        // Extract identity fields from ActivityExecute for cancellation support
        let (activity_instance, activity_execution_id, activity_id, session_id, tag) = match &item {
            WorkItem::ActivityExecute {
                instance,
                execution_id,
                id,
                session_id,
                tag,
                ..
            } => (
                Some(instance.as_str()),
                Some(*execution_id),
                Some(*id),
                session_id.as_deref(),
                tag.as_deref(),
            ),
            _ => (None, None, None, None, None),
        };

        let work_item = serde_json::to_string(&item)
            .map_err(|e| ProviderError::permanent("enqueue_for_worker", format!("Serialization error: {e}")))?;
        let now_ms = Self::now_millis();

        sqlx::query(
            r#"
            INSERT INTO worker_queue (work_item, visible_at, instance_id, execution_id, activity_id, session_id, tag)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            "#,
        )
        .bind(work_item)
        .bind(now_ms)
        .bind(activity_instance)
        .bind(activity_execution_id.map(|e| e as i64))
        .bind(activity_id.map(|a| a as i64))
        .bind(session_id)
        .bind(tag)
        .execute(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("enqueue_for_worker", e))?;

        Ok(())
    }

    async fn fetch_work_item(
        &self,
        lock_timeout: Duration,
        _poll_timeout: Duration,
        session: Option<&SessionFetchConfig>,
        tag_filter: &TagFilter,
    ) -> Result<Option<(WorkItem, String, u32)>, ProviderError> {
        // TagFilter::None means this worker doesn't process any activities
        if matches!(tag_filter, TagFilter::None) {
            return Ok(None);
        }

        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?;

        let lock_token = Self::generate_lock_token();
        let locked_until = Self::timestamp_after(lock_timeout);

        tracing::debug!(
            "Worker dequeue: looking for available items, locked_until will be {}",
            locked_until
        );

        let now_ms = Self::now_millis();

        // Build tag filter SQL clause
        // Session path uses ?1 (now_ms), ?2 (owner_id) → tags start at ?3
        // Non-session path uses ?1 (now_ms) → tags start at ?2
        let tag_start_param = if session.is_some() { 3 } else { 2 };
        let tag_clause = Self::build_tag_clause(tag_filter, tag_start_param);
        let tag_values = Self::collect_tag_values(tag_filter);

        // Session-aware vs non-session fetch
        let next_item = if let Some(config) = session {
            // Session-aware fetch with tag filtering
            let sql = format!(
                r#"
                SELECT q.id, q.work_item, q.attempt_count, q.session_id
                FROM worker_queue q
                LEFT JOIN sessions s ON s.session_id = q.session_id AND s.locked_until > ?1
                WHERE q.visible_at <= ?1
                  AND (q.lock_token IS NULL OR q.locked_until <= ?1)
                  AND (
                    q.session_id IS NULL
                    OR s.worker_id = ?2
                    OR s.session_id IS NULL
                  )
                  AND ({tag_clause})
                ORDER BY q.id
                LIMIT 1
                "#,
            );
            let mut query = sqlx::query(&sql).bind(now_ms).bind(&config.owner_id);
            for val in &tag_values {
                query = query.bind(val.as_str());
            }
            query
                .fetch_optional(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?
        } else {
            // Non-session fetch with tag filtering
            let sql = format!(
                r#"
                SELECT q.id, q.work_item, q.attempt_count, q.session_id FROM worker_queue q
                WHERE q.visible_at <= ?1
                  AND (q.lock_token IS NULL OR q.locked_until <= ?1)
                  AND q.session_id IS NULL
                  AND ({tag_clause})
                ORDER BY q.id
                LIMIT 1
                "#,
            );
            let mut query = sqlx::query(&sql).bind(now_ms);
            for val in &tag_values {
                query = query.bind(val.as_str());
            }
            query
                .fetch_optional(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?
        };

        if next_item.is_none() {
            tracing::debug!("Worker dequeue: no available items found");
            return Ok(None);
        }

        let next_item = next_item.unwrap();

        tracing::debug!("Worker dequeue found item");

        let id: i64 = next_item
            .try_get("id")
            .map_err(|e| ProviderError::permanent("fetch_work_item", format!("Failed to get id: {e}")))?;
        let work_item_str: String = next_item
            .try_get("work_item")
            .map_err(|e| ProviderError::permanent("fetch_work_item", format!("Failed to get work_item: {e}")))?;
        let current_attempt_count: i64 = next_item
            .try_get("attempt_count")
            .map_err(|e| ProviderError::permanent("fetch_work_item", format!("Failed to get attempt_count: {e}")))?;
        let session_id: Option<String> = next_item
            .try_get("session_id")
            .map_err(|e| ProviderError::permanent("fetch_work_item", format!("Failed to get session_id: {e}")))?;

        // Update with lock and increment attempt_count for poison message detection
        sqlx::query(
            r#"
            UPDATE worker_queue
            SET lock_token = ?1, locked_until = ?2, attempt_count = attempt_count + 1
            WHERE id = ?3
            "#,
        )
        .bind(&lock_token)
        .bind(locked_until)
        .bind(id)
        .execute(&mut *tx)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?;

        // If session-bound, atomically upsert the sessions row and verify ownership
        if let (Some(sid), Some(config)) = (&session_id, session) {
            let session_locked_until = now_ms + config.lock_timeout.as_millis() as i64;
            let upsert_result = sqlx::query(
                r#"
                INSERT INTO sessions (session_id, worker_id, locked_until, last_activity_at)
                VALUES (?1, ?2, ?3, ?4)
                ON CONFLICT (session_id) DO UPDATE
                SET worker_id = ?2,
                    locked_until = ?3,
                    last_activity_at = ?4
                WHERE sessions.locked_until <= ?4 OR sessions.worker_id = ?2
                "#,
            )
            .bind(sid)
            .bind(&config.owner_id)
            .bind(session_locked_until)
            .bind(now_ms)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?;

            // rows_affected == 0 means the conflict WHERE clause rejected our claim:
            // another worker owns this session with an active lock. Roll back so the
            // queue item is released and can be retried when the session frees up.
            if upsert_result.rows_affected() == 0 {
                tracing::warn!(
                    target: "duroxide::providers::sqlite",
                    session_id = %sid,
                    owner_id = %config.owner_id,
                    "Session claim failed — owned by another worker, rolling back"
                );
                tx.rollback().await.ok();
                return Ok(None);
            }

            tracing::debug!(
                target: "duroxide::providers::sqlite",
                session_id = %sid,
                owner_id = %config.owner_id,
                "Session claimed/refreshed on fetch"
            );
        }

        // The attempt_count after the UPDATE is current + 1
        let attempt_count = (current_attempt_count + 1) as u32;

        let work_item: WorkItem = serde_json::from_str(&work_item_str)
            .map_err(|e| ProviderError::permanent("fetch_work_item", format!("Deserialization error: {e}")))?;

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("fetch_work_item", e))?;

        Ok(Some((work_item, lock_token, attempt_count)))
    }

    async fn ack_work_item(&self, token: &str, completion: Option<WorkItem>) -> Result<(), ProviderError> {
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("ack_work_item", e))?;

        // Delete worker item atomically, returning session_id for last_activity_at piggyback.
        // Only deletes if the lock is still valid (locked_until > now). This ensures:
        // - Expired locks are rejected (worker took too long without renewing)
        // - Cancelled activities (row deleted via lock stealing) return None
        let now_ms = Self::now_millis();
        let deleted_row: Option<(Option<String>,)> =
            sqlx::query_as("DELETE FROM worker_queue WHERE lock_token = ? AND locked_until > ? RETURNING session_id")
                .bind(token)
                .bind(now_ms)
                .fetch_optional(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_work_item", e))?;

        let session_id = match deleted_row {
            Some((sid,)) => sid,
            None => {
                // Row was already deleted (cancelled) or lock expired
                return Err(ProviderError::permanent(
                    "ack_work_item",
                    "Activity was cancelled or lock expired (worker queue row not found or lock invalid)",
                ));
            }
        };

        // Piggyback: update last_activity_at for session-bound items.
        // Guard with locked_until > now so we don't bump last_activity_at on a session
        // that was taken over by another worker after our session lock expired.
        if let Some(ref sid) = session_id {
            sqlx::query("UPDATE sessions SET last_activity_at = ?1 WHERE session_id = ?2 AND locked_until > ?1")
                .bind(now_ms)
                .bind(sid)
                .execute(&mut *tx)
                .await
                .ok(); // Best-effort
        }

        // Only enqueue completion if provided (None means drop without notifying orchestrator)
        if let Some(completion) = completion {
            // Extract instance from completion
            let instance = match &completion {
                WorkItem::ActivityCompleted { instance, .. } => instance,
                WorkItem::ActivityFailed { instance, .. } => instance,
                _ => {
                    return Err(ProviderError::permanent(
                        "ack_work_item",
                        "Invalid completion type for worker ack",
                    ));
                }
            };

            // Enqueue completion to orchestrator queue
            let work_item = serde_json::to_string(&completion)
                .map_err(|e| ProviderError::permanent("ack_work_item", format!("Serialization error: {e}")))?;
            let now_ms = Self::now_millis();

            sqlx::query("INSERT INTO orchestrator_queue (instance_id, work_item, visible_at) VALUES (?, ?, ?)")
                .bind(instance)
                .bind(work_item)
                .bind(now_ms)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_work_item", e))?;

            tx.commit()
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_work_item", e))?;

            debug!(instance = %instance, "Atomically acked worker and enqueued completion");
        } else {
            tx.commit()
                .await
                .map_err(|e| Self::sqlx_to_provider_error("ack_work_item", e))?;

            debug!("Acked worker item without enqueuing completion (orchestration terminal/missing)");
        }

        Ok(())
    }

    async fn renew_work_item_lock(&self, token: &str, extend_for: Duration) -> Result<(), ProviderError> {
        let now_ms = Self::now_millis();
        let locked_until = Self::timestamp_after(extend_for);

        // Extend lock and return session_id in one query.
        // If no row is returned, activity was cancelled via lock stealing.
        let row: Option<(Option<String>,)> = sqlx::query_as(
            r#"
            UPDATE worker_queue
            SET locked_until = ?1
            WHERE lock_token = ?2
              AND locked_until > ?3
            RETURNING session_id
            "#,
        )
        .bind(locked_until)
        .bind(token)
        .bind(now_ms)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("renew_work_item_lock", e))?;

        let session_id = match row {
            Some((sid,)) => sid,
            None => {
                return Err(ProviderError::permanent(
                    "renew_work_item_lock",
                    "Lock renewal failed - activity was cancelled or lock expired",
                ));
            }
        };

        // Piggyback: update last_activity_at for session-bound items.
        // Guard with locked_until > now so we don't bump last_activity_at on a session
        // that was taken over by another worker after our session lock expired.
        if let Some(ref sid) = session_id {
            sqlx::query("UPDATE sessions SET last_activity_at = ?1 WHERE session_id = ?2 AND locked_until > ?1")
                .bind(now_ms)
                .bind(sid)
                .execute(&self.pool)
                .await
                .ok(); // Best-effort: don't fail renewal if session update fails
        }

        tracing::debug!(
            target: "duroxide::providers::sqlite",
            lock_token = %token,
            extend_secs = %extend_for.as_secs(),
            "Work item lock renewed"
        );

        Ok(())
    }

    async fn abandon_work_item(
        &self,
        token: &str,
        delay: Option<Duration>,
        ignore_attempt: bool,
    ) -> Result<(), ProviderError> {
        // Worker queue uses visible_at for visibility control (like orchestrator queue)
        // When delay is specified, set visible_at to future time to delay retry
        // Always clear lock_token and locked_until since the lock is being released

        let now_ms = Self::now_millis();
        let visible_at = if let Some(d) = delay {
            Self::timestamp_after(d)
        } else {
            now_ms
        };

        let query = if ignore_attempt {
            r#"
            UPDATE worker_queue
            SET lock_token = NULL, locked_until = NULL, visible_at = ?1,
                attempt_count = MAX(0, attempt_count - 1)
            WHERE lock_token = ?2
            "#
        } else {
            r#"
            UPDATE worker_queue
            SET lock_token = NULL, locked_until = NULL, visible_at = ?1
            WHERE lock_token = ?2
            "#
        };
        let result = sqlx::query(query)
            .bind(visible_at)
            .bind(token)
            .execute(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("abandon_work_item", e))?;

        if result.rows_affected() == 0 {
            return Err(ProviderError::permanent(
                "abandon_work_item",
                "Invalid lock token or already acked",
            ));
        }

        tracing::debug!(
            target: "duroxide::providers::sqlite",
            lock_token = %token,
            delay_ms = ?delay.map(|d| d.as_millis()),
            ignore_attempt = %ignore_attempt,
            "Work item abandoned"
        );

        Ok(())
    }

    async fn renew_orchestration_item_lock(&self, token: &str, extend_for: Duration) -> Result<(), ProviderError> {
        let locked_until = Self::timestamp_after(extend_for);
        let now_ms = Self::now_millis();

        let result = sqlx::query(
            r#"
            UPDATE instance_locks
            SET locked_until = ?1
            WHERE lock_token = ?2
              AND locked_until > ?3
            "#,
        )
        .bind(locked_until)
        .bind(token)
        .bind(now_ms)
        .execute(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("renew_orchestration_item_lock", e))?;

        if result.rows_affected() == 0 {
            return Err(ProviderError::permanent(
                "renew_orchestration_item_lock",
                "Lock token invalid, expired, or already acked",
            ));
        }

        tracing::debug!(
            target: "duroxide::providers::sqlite",
            lock_token = %token,
            extend_secs = %extend_for.as_secs(),
            "Orchestration item lock renewed"
        );

        Ok(())
    }

    async fn abandon_orchestration_item(
        &self,
        lock_token: &str,
        delay: Option<Duration>,
        ignore_attempt: bool,
    ) -> Result<(), ProviderError> {
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;

        // Get instance_id from lock before removing it
        let instance_id: Option<String> =
            sqlx::query_scalar("SELECT instance_id FROM instance_locks WHERE lock_token = ?")
                .bind(lock_token)
                .fetch_optional(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;

        let Some(instance_id) = instance_id else {
            return Err(ProviderError::permanent(
                "abandon_orchestration_item",
                "Invalid lock token",
            ));
        };

        // Remove instance lock
        sqlx::query("DELETE FROM instance_locks WHERE lock_token = ?")
            .bind(lock_token)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;

        // If ignore_attempt is true, decrement attempt_count on orchestrator_queue messages
        if ignore_attempt {
            sqlx::query(
                "UPDATE orchestrator_queue SET attempt_count = MAX(0, attempt_count - 1) WHERE instance_id = ?",
            )
            .bind(&instance_id)
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;
        }

        // Optionally delay messages for this instance
        if let Some(delay) = delay {
            let delay_ms = delay.as_millis().min(i64::MAX as u128) as i64;
            let visible_at = Self::now_millis().saturating_add(delay_ms);
            sqlx::query("UPDATE orchestrator_queue SET visible_at = ? WHERE instance_id = ? AND visible_at <= ?")
                .bind(visible_at)
                .bind(&instance_id)
                .bind(Self::now_millis())
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;
        }

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("abandon_orchestration_item", e))?;

        Ok(())
    }

    async fn renew_session_lock(
        &self,
        owner_ids: &[&str],
        extend_for: Duration,
        idle_timeout: Duration,
    ) -> Result<usize, ProviderError> {
        if owner_ids.is_empty() {
            return Ok(0);
        }

        let now_ms = Self::now_millis();
        let locked_until = now_ms + extend_for.as_millis() as i64;
        let idle_cutoff = now_ms - idle_timeout.as_millis() as i64;

        // Build IN clause with positional parameters: ?3, ?4, ...
        let placeholders: Vec<String> = (0..owner_ids.len()).map(|i| format!("?{}", i + 3)).collect();
        let sql = format!(
            "UPDATE sessions SET locked_until = ?1 \
             WHERE worker_id IN ({}) \
             AND locked_until > ?2 \
             AND last_activity_at > ?{}",
            placeholders.join(", "),
            owner_ids.len() + 3,
        );

        let mut query = sqlx::query(&sql).bind(locked_until).bind(now_ms);
        for owner_id in owner_ids {
            query = query.bind(owner_id);
        }
        query = query.bind(idle_cutoff);

        let result = query
            .execute(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("renew_session_lock", e))?;

        let count = result.rows_affected() as usize;
        tracing::debug!(
            target: "duroxide::providers::sqlite",
            owner_count = %owner_ids.len(),
            sessions_renewed = %count,
            "Session locks renewed"
        );
        Ok(count)
    }

    async fn cleanup_orphaned_sessions(&self, _idle_timeout: Duration) -> Result<usize, ProviderError> {
        let now_ms = Self::now_millis();

        let result = sqlx::query(
            r#"
            DELETE FROM sessions
            WHERE locked_until < ?1
              AND NOT EXISTS (
                  SELECT 1 FROM worker_queue WHERE worker_queue.session_id = sessions.session_id
              )
            "#,
        )
        .bind(now_ms)
        .execute(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("cleanup_orphaned_sessions", e))?;

        let count = result.rows_affected() as usize;
        tracing::debug!(
            target: "duroxide::providers::sqlite",
            sessions_cleaned = %count,
            "Orphaned sessions cleaned up"
        );
        Ok(count)
    }

    fn as_management_capability(&self) -> Option<&dyn ProviderAdmin> {
        Some(self as &dyn ProviderAdmin)
    }

    async fn get_custom_status(
        &self,
        instance: &str,
        last_seen_version: u64,
    ) -> Result<Option<(Option<String>, u64)>, ProviderError> {
        let row = sqlx::query(
            r#"
            SELECT custom_status, custom_status_version
            FROM instances
            WHERE instance_id = ? AND custom_status_version > ?
            "#,
        )
        .bind(instance)
        .bind(last_seen_version as i64)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_custom_status", e))?;

        match row {
            Some(row) => {
                let custom_status: Option<String> = row.try_get("custom_status").ok().flatten();
                let version: i64 = row.try_get("custom_status_version").unwrap_or(0);
                Ok(Some((custom_status, version as u64)))
            }
            None => Ok(None),
        }
    }

    async fn get_kv_value(&self, instance: &str, key: &str) -> Result<Option<String>, ProviderError> {
        // Check kv_delta first (current execution mutations)
        let delta_row = sqlx::query("SELECT value FROM kv_delta WHERE instance_id = ? AND key = ?")
            .bind(instance)
            .bind(key)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_kv_value", e))?;

        if let Some(r) = delta_row {
            // Row exists in delta: value IS NOT NULL = live value, NULL = tombstone
            let value: Option<String> = r
                .try_get("value")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_value", e))?;
            return Ok(value);
        }

        // No delta row — fall through to kv_store (prior-execution state)
        let row = sqlx::query("SELECT value FROM kv_store WHERE instance_id = ? AND key = ?")
            .bind(instance)
            .bind(key)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_kv_value", e))?;

        Ok(match row {
            Some(r) => Some(
                r.try_get("value")
                    .map_err(|e| Self::sqlx_to_provider_error("get_kv_value", e))?,
            ),
            None => None,
        })
    }

    async fn get_kv_all_values(
        &self,
        instance: &str,
    ) -> Result<std::collections::HashMap<String, String>, ProviderError> {
        // Start with kv_store (prior-execution state)
        let store_rows = sqlx::query("SELECT key, value FROM kv_store WHERE instance_id = ?")
            .bind(instance)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;

        let mut map = std::collections::HashMap::new();
        for row in store_rows {
            let k: String = row
                .try_get("key")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;
            let v: String = row
                .try_get("value")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;
            map.insert(k, v);
        }

        // Overlay kv_delta (current execution mutations)
        let delta_rows = sqlx::query("SELECT key, value FROM kv_delta WHERE instance_id = ?")
            .bind(instance)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;

        for row in delta_rows {
            let k: String = row
                .try_get("key")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;
            let v: Option<String> = row
                .try_get("value")
                .map_err(|e| Self::sqlx_to_provider_error("get_kv_all_values", e))?;
            match v {
                Some(value) => {
                    map.insert(k, value);
                }
                None => {
                    map.remove(&k);
                } // tombstone
            }
        }
        Ok(map)
    }

    async fn get_instance_stats(&self, instance: &str) -> Result<Option<crate::SystemStats>, ProviderError> {
        // Check instance exists and get current execution_id
        let exec_row = sqlx::query("SELECT current_execution_id FROM instances WHERE instance_id = ?")
            .bind(instance)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        let exec_id: i64 = match exec_row {
            Some(row) => row
                .try_get("current_execution_id")
                .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?,
            None => return Ok(None),
        };

        // History event count and byte size for current execution
        let history_row = sqlx::query(
            "SELECT COUNT(*) as cnt, COALESCE(SUM(LENGTH(event_data)), 0) as size_bytes \
             FROM history WHERE instance_id = ? AND execution_id = ?",
        )
        .bind(instance)
        .bind(exec_id)
        .fetch_one(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        let history_event_count: i64 = history_row
            .try_get("cnt")
            .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;
        let history_size_bytes: i64 = history_row
            .try_get("size_bytes")
            .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        // KV stats: merge kv_store + kv_delta (delta overrides store, tombstones excluded).
        let kv_row = sqlx::query(
            "SELECT COUNT(*) as cnt, COALESCE(SUM(LENGTH(value)), 0) as size_bytes \
             FROM ( \
                 SELECT COALESCE(d.key, s.key) AS key, \
                        CASE WHEN d.key IS NOT NULL THEN d.value ELSE s.value END AS value \
                 FROM kv_store s \
                 LEFT JOIN kv_delta d ON s.instance_id = d.instance_id AND s.key = d.key \
                 WHERE s.instance_id = ? \
                 UNION \
                 SELECT d.key, d.value \
                 FROM kv_delta d \
                 LEFT JOIN kv_store s ON d.instance_id = s.instance_id AND d.key = s.key \
                 WHERE d.instance_id = ? AND s.key IS NULL \
             ) merged WHERE value IS NOT NULL",
        )
        .bind(instance)
        .bind(instance)
        .fetch_one(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        let kv_user_key_count: i64 = kv_row
            .try_get("cnt")
            .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;
        let kv_total_value_bytes: i64 = kv_row
            .try_get("size_bytes")
            .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        // Queue pending count: count carry_forward_events from the OrchestrationStarted event.
        // These are unprocessed queue messages carried from the previous execution.
        let carry_forward_row = sqlx::query(
            "SELECT event_data FROM history \
             WHERE instance_id = ? AND execution_id = ? AND event_id = 1",
        )
        .bind(instance)
        .bind(exec_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;

        let queue_pending_count = match carry_forward_row {
            Some(row) => {
                let data: String = row
                    .try_get("event_data")
                    .map_err(|e| Self::sqlx_to_provider_error("get_instance_stats", e))?;
                let event: crate::Event =
                    serde_json::from_str(&data).map_err(|e| {
                        ProviderError::permanent(
                            "get_instance_stats",
                            format!(
                                "Failed to deserialize OrchestrationStarted event: {e}"
                            ),
                        )
                    })?;
                match event.kind {
                    crate::EventKind::OrchestrationStarted {
                        carry_forward_events: Some(ref events),
                        ..
                    } => events.len() as u64,
                    _ => 0,
                }
            }
            None => 0,
        };

        Ok(Some(crate::SystemStats {
            history_event_count: history_event_count as u64,
            history_size_bytes: history_size_bytes as u64,
            queue_pending_count,
            kv_user_key_count: kv_user_key_count as u64,
            kv_total_value_bytes: kv_total_value_bytes as u64,
        }))
    }
}

#[async_trait::async_trait]
impl ProviderAdmin for SqliteProvider {
    async fn list_instances(&self) -> Result<Vec<String>, ProviderError> {
        let rows = sqlx::query("SELECT instance_id FROM instances ORDER BY created_at DESC")
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("latest_execution_id", e))?;

        let instances: Vec<String> = rows
            .into_iter()
            .map(|row| row.try_get("instance_id").unwrap_or_default())
            .collect();

        Ok(instances)
    }

    async fn list_instances_by_status(&self, status: &str) -> Result<Vec<String>, ProviderError> {
        let rows = sqlx::query(
            r#"
            SELECT i.instance_id 
            FROM instances i
            JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
            WHERE e.status = ?
            ORDER BY i.created_at DESC
            "#,
        )
        .bind(status)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("list_instances_by_status", e))?;

        let instances: Vec<String> = rows
            .into_iter()
            .map(|row| row.try_get("instance_id").unwrap_or_default())
            .collect();

        Ok(instances)
    }

    async fn list_executions(&self, instance: &str) -> Result<Vec<u64>, ProviderError> {
        let rows = sqlx::query("SELECT execution_id FROM executions WHERE instance_id = ? ORDER BY execution_id")
            .bind(instance)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_executions", e))?;

        let executions: Vec<u64> = rows
            .into_iter()
            .map(|row| row.try_get::<i64, _>("execution_id").unwrap_or(0) as u64)
            .collect();

        Ok(executions)
    }

    async fn read_history_with_execution_id(
        &self,
        instance: &str,
        execution_id: u64,
    ) -> Result<Vec<Event>, ProviderError> {
        let rows = sqlx::query(
            r#"
            SELECT event_data 
            FROM history 
            WHERE instance_id = ? AND execution_id = ? 
            ORDER BY event_id
            "#,
        )
        .bind(instance)
        .bind(execution_id as i64)
        .fetch_all(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("read_history_with_execution_id", e))?;

        let mut events = Vec::new();
        for row in rows {
            let event_data: String = row
                .try_get("event_data")
                .map_err(|e| Self::sqlx_to_provider_error("read_history_with_execution_id", e))?;
            let event: Event = serde_json::from_str(&event_data).map_err(|e| {
                ProviderError::permanent(
                    "read_history_with_execution_id",
                    format!("Failed to deserialize event: {e}"),
                )
            })?;
            events.push(event);
        }

        Ok(events)
    }

    async fn read_history(&self, instance: &str) -> Result<Vec<Event>, ProviderError> {
        let execution_id = self.latest_execution_id(instance).await?;
        self.read_history_with_execution_id(instance, execution_id).await
    }

    async fn latest_execution_id(&self, instance: &str) -> Result<u64, ProviderError> {
        let row = sqlx::query(
            "SELECT COALESCE(MAX(execution_id), 1) as max_execution_id FROM executions WHERE instance_id = ?",
        )
        .bind(instance)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("latest_execution_id", e))?;

        match row {
            Some(row) => {
                let max_id: i64 = row.try_get("max_execution_id").unwrap_or(1);
                Ok(max_id as u64)
            }
            None => Ok(1), // Default to execution 1 if no executions exist
        }
    }

    async fn get_instance_info(&self, instance: &str) -> Result<InstanceInfo, ProviderError> {
        let row = sqlx::query(
            r#"
            SELECT
                i.instance_id,
                i.orchestration_name,
                i.orchestration_version,
                i.current_execution_id,
                i.created_at,
                i.updated_at,
                i.parent_instance_id,
                e.status,
                e.output
            FROM instances i
            LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
            WHERE i.instance_id = ?
            "#,
        )
        .bind(instance)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_instance_info", e))?;

        match row {
            Some(row) => {
                let instance_id: String = row
                    .try_get("instance_id")
                    .map_err(|e| Self::sqlx_to_provider_error("get_instance_info", e))?;
                let orchestration_name: String = row
                    .try_get("orchestration_name")
                    .map_err(|e| Self::sqlx_to_provider_error("get_instance_info", e))?;
                let orchestration_version: Option<String> = row.try_get("orchestration_version").ok();
                let orchestration_version = orchestration_version.unwrap_or_else(|| {
                    // Could fetch from history, but for management API, "unknown" is acceptable
                    "unknown".to_string()
                });
                let current_execution_id: i64 = row.try_get("current_execution_id").unwrap_or(1);
                let created_at: i64 = row.try_get("created_at").unwrap_or(0);
                let updated_at: i64 = row.try_get("updated_at").unwrap_or(0);
                let status: String = row.try_get("status").unwrap_or_else(|_| "Unknown".to_string());
                let output: Option<String> = row.try_get("output").ok();
                let parent_instance_id: Option<String> = row.try_get("parent_instance_id").ok().flatten();

                Ok(InstanceInfo {
                    instance_id,
                    orchestration_name,
                    orchestration_version,
                    current_execution_id: current_execution_id as u64,
                    status,
                    output,
                    created_at: created_at as u64,
                    updated_at: updated_at as u64,
                    parent_instance_id,
                })
            }
            None => Err(ProviderError::permanent(
                "get_instance_info",
                format!("Instance {instance} not found"),
            )),
        }
    }

    async fn get_execution_info(&self, instance: &str, execution_id: u64) -> Result<ExecutionInfo, ProviderError> {
        let row = sqlx::query(
            r#"
            SELECT 
                e.execution_id,
                e.status,
                e.output,
                e.started_at,
                e.completed_at,
                COUNT(h.event_id) as event_count
            FROM executions e
            LEFT JOIN history h ON e.instance_id = h.instance_id AND e.execution_id = h.execution_id
            WHERE e.instance_id = ? AND e.execution_id = ?
            GROUP BY e.execution_id
            "#,
        )
        .bind(instance)
        .bind(execution_id as i64)
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_execution_info", e))?;

        match row {
            Some(row) => {
                let execution_id: i64 = row
                    .try_get("execution_id")
                    .map_err(|e| Self::sqlx_to_provider_error("get_execution_info", e))?;
                let status: String = row
                    .try_get("status")
                    .map_err(|e| Self::sqlx_to_provider_error("get_execution_info", e))?;
                let output: Option<String> = row.try_get("output").ok();
                let started_at: i64 = row.try_get("started_at").unwrap_or(0);
                let completed_at: Option<i64> = row.try_get("completed_at").ok();
                let event_count: i64 = row.try_get("event_count").unwrap_or(0);

                Ok(ExecutionInfo {
                    execution_id: execution_id as u64,
                    status,
                    output,
                    started_at: started_at as u64,
                    completed_at: completed_at.map(|t| t as u64),
                    event_count: event_count as usize,
                })
            }
            None => Err(ProviderError::permanent(
                "get_execution_info",
                format!("Execution {execution_id} not found for instance {instance}"),
            )),
        }
    }

    async fn get_system_metrics(&self) -> Result<SystemMetrics, ProviderError> {
        let row = sqlx::query(
            r#"
            SELECT 
                COUNT(*) as total_instances,
                SUM(CASE WHEN e.status = 'Running' THEN 1 ELSE 0 END) as running_instances,
                SUM(CASE WHEN e.status = 'Completed' THEN 1 ELSE 0 END) as completed_instances,
                SUM(CASE WHEN e.status = 'Failed' THEN 1 ELSE 0 END) as failed_instances,
                SUM(CASE WHEN e.status = 'ContinuedAsNew' THEN 1 ELSE 0 END) as continued_as_new_instances
            FROM instances i
            LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
            "#,
        )
        .fetch_optional(&self.pool)
        .await
        .map_err(|e| Self::sqlx_to_provider_error("get_system_metrics", e))?;

        match row {
            Some(row) => {
                let total_instances: i64 = row.try_get("total_instances").unwrap_or(0);
                let running_instances: i64 = row.try_get("running_instances").unwrap_or(0);
                let completed_instances: i64 = row.try_get("completed_instances").unwrap_or(0);
                let failed_instances: i64 = row.try_get("failed_instances").unwrap_or(0);
                let _: i64 = row.try_get("continued_as_new_instances").unwrap_or(0);

                // Get total executions count
                let total_executions_row = sqlx::query("SELECT COUNT(*) as total_executions FROM executions")
                    .fetch_optional(&self.pool)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("get_system_metrics", e))?;
                let total_executions: i64 = total_executions_row
                    .and_then(|row| row.try_get("total_executions").ok())
                    .unwrap_or(0);

                // Get total events count
                let total_events_row = sqlx::query("SELECT COUNT(*) as total_events FROM history")
                    .fetch_optional(&self.pool)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("get_system_metrics", e))?;
                let total_events: i64 = total_events_row
                    .and_then(|row| row.try_get("total_events").ok())
                    .unwrap_or(0);

                Ok(SystemMetrics {
                    total_instances: total_instances as u64,
                    total_executions: total_executions as u64,
                    running_instances: running_instances as u64,
                    completed_instances: completed_instances as u64,
                    failed_instances: failed_instances as u64,
                    total_events: total_events as u64,
                })
            }
            None => Ok(SystemMetrics::default()),
        }
    }

    async fn get_queue_depths(&self) -> Result<QueueDepths, ProviderError> {
        // Get orchestrator queue depth
        let orchestrator_row = sqlx::query("SELECT COUNT(*) as count FROM orchestrator_queue WHERE lock_token IS NULL")
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_queue_depths", e))?;
        let orchestrator_queue: usize =
            orchestrator_row.and_then(|row| row.try_get("count").ok()).unwrap_or(0) as usize;

        // Get worker queue depth
        let worker_row = sqlx::query("SELECT COUNT(*) as count FROM worker_queue WHERE lock_token IS NULL")
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_queue_depths", e))?;
        let worker_queue: usize = worker_row.and_then(|row| row.try_get("count").ok()).unwrap_or(0) as usize;

        Ok(QueueDepths {
            orchestrator_queue,
            worker_queue,
            timer_queue: 0, // Timer queue no longer exists - timers handled by orchestrator queue
        })
    }

    // ===== Hierarchy Primitive Operations =====

    async fn list_children(&self, instance_id: &str) -> Result<Vec<String>, ProviderError> {
        let rows = sqlx::query("SELECT instance_id FROM instances WHERE parent_instance_id = ?")
            .bind(instance_id)
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("list_children", e))?;

        let children: Vec<String> = rows
            .into_iter()
            .filter_map(|row| row.try_get("instance_id").ok())
            .collect();

        Ok(children)
    }

    async fn get_parent_id(&self, instance_id: &str) -> Result<Option<String>, ProviderError> {
        let row = sqlx::query("SELECT parent_instance_id FROM instances WHERE instance_id = ?")
            .bind(instance_id)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("get_parent_id", e))?;

        match row {
            Some(r) => Ok(r.try_get("parent_instance_id").ok().flatten()),
            None => Err(ProviderError::permanent(
                "get_parent_id",
                format!("Instance {instance_id} not found"),
            )),
        }
    }

    async fn delete_instances_atomic(
        &self,
        ids: &[String],
        force: bool,
    ) -> Result<DeleteInstanceResult, ProviderError> {
        if ids.is_empty() {
            return Ok(DeleteInstanceResult::default());
        }

        // Build placeholders for IN clause: ?, ?, ?, ...
        let placeholders: String = ids.iter().map(|_| "?").collect::<Vec<_>>().join(",");

        // Step 1: If not force, check that all instances are in terminal state
        if !force {
            // NOTE: Dynamic SQL with placeholders - see TODO.md "Security: SQL Injection Audit"
            let check_sql = format!(
                r#"
                SELECT i.instance_id, e.status
                FROM instances i
                LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
                WHERE i.instance_id IN ({placeholders})
                "#
            );

            let mut query = sqlx::query(&check_sql);
            for id in ids {
                query = query.bind(id);
            }

            let rows = query
                .fetch_all(&self.pool)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

            for row in rows {
                let status: Option<String> = row.try_get("status").ok();
                if status.as_deref() == Some("Running") {
                    let instance_id: String = row.try_get("instance_id").unwrap_or_default();
                    return Err(ProviderError::permanent(
                        "delete_instances_atomic",
                        format!(
                            "Instance {instance_id} is still running. Use force=true to delete anyway, or cancel first."
                        ),
                    ));
                }
            }
        }

        // Step 2: Delete all instances in a single transaction using bulk operations
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Step 2a: Check for orphan race condition
        // If any instance has parent_instance_id pointing to an instance we're deleting,
        // but that child is NOT in our delete list, we'd create an orphan.
        // This can happen if a new sub-orchestration was spawned between get_instance_tree() and now.
        let orphan_check_sql = format!(
            r#"
            SELECT instance_id, parent_instance_id FROM instances
            WHERE parent_instance_id IN ({placeholders})
              AND instance_id NOT IN ({placeholders})
            LIMIT 1
            "#
        );
        let mut orphan_query = sqlx::query(&orphan_check_sql);
        // Bind ids twice: once for parent_instance_id IN, once for instance_id NOT IN
        for id in ids {
            orphan_query = orphan_query.bind(id);
        }
        for id in ids {
            orphan_query = orphan_query.bind(id);
        }
        if let Some(orphan_row) = orphan_query
            .fetch_optional(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?
        {
            let orphan_id: String = orphan_row.try_get("instance_id").unwrap_or_default();
            let parent_id: String = orphan_row.try_get("parent_instance_id").unwrap_or_default();
            return Err(ProviderError::permanent(
                "delete_instances_atomic",
                format!(
                    "Cannot delete: instance {parent_id} has child {orphan_id} that was created after tree traversal. \
                     Re-fetch the tree and retry."
                ),
            ));
        }

        let mut result = DeleteInstanceResult::default();

        // Count history events before deletion
        let count_history_sql = format!("SELECT COUNT(*) as count FROM history WHERE instance_id IN ({placeholders})");
        let mut count_query = sqlx::query(&count_history_sql);
        for id in ids {
            count_query = count_query.bind(id);
        }
        let history_count: i64 = count_query
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?
            .try_get("count")
            .unwrap_or(0);
        result.events_deleted = history_count as u64;

        // Count executions before deletion
        let count_exec_sql = format!("SELECT COUNT(*) as count FROM executions WHERE instance_id IN ({placeholders})");
        let mut count_query = sqlx::query(&count_exec_sql);
        for id in ids {
            count_query = count_query.bind(id);
        }
        let exec_count: i64 = count_query
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?
            .try_get("count")
            .unwrap_or(0);
        result.executions_deleted = exec_count as u64;

        // Count queue messages before deletion
        let count_orch_q_sql =
            format!("SELECT COUNT(*) as count FROM orchestrator_queue WHERE instance_id IN ({placeholders})");
        let mut count_query = sqlx::query(&count_orch_q_sql);
        for id in ids {
            count_query = count_query.bind(id);
        }
        let orch_q_count: i64 = count_query
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?
            .try_get("count")
            .unwrap_or(0);

        let count_worker_q_sql =
            format!("SELECT COUNT(*) as count FROM worker_queue WHERE instance_id IN ({placeholders})");
        let mut count_query = sqlx::query(&count_worker_q_sql);
        for id in ids {
            count_query = count_query.bind(id);
        }
        let worker_q_count: i64 = count_query
            .fetch_one(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?
            .try_get("count")
            .unwrap_or(0);

        result.queue_messages_deleted = (orch_q_count + worker_q_count) as u64;

        // Bulk delete from all tables (order matters for FK constraints if any)
        // Delete history
        let del_history_sql = format!("DELETE FROM history WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_history_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete executions
        let del_exec_sql = format!("DELETE FROM executions WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_exec_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete orchestrator queue
        let del_orch_q_sql = format!("DELETE FROM orchestrator_queue WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_orch_q_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete worker queue
        let del_worker_q_sql = format!("DELETE FROM worker_queue WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_worker_q_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete instance locks (important: before instances to prevent zombie recreation)
        let del_locks_sql = format!("DELETE FROM instance_locks WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_locks_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete KV store entries
        let del_kv_sql = format!("DELETE FROM kv_store WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_kv_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete KV delta entries
        let del_kv_delta_sql = format!("DELETE FROM kv_delta WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_kv_delta_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        // Delete instances
        let del_instances_sql = format!("DELETE FROM instances WHERE instance_id IN ({placeholders})");
        let mut del_query = sqlx::query(&del_instances_sql);
        for id in ids {
            del_query = del_query.bind(id);
        }
        del_query
            .execute(&mut *tx)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instances_atomic", e))?;

        result.instances_deleted = ids.len() as u64;
        Ok(result)
    }

    // delete_instance: uses default implementation from ProviderAdmin trait
    // get_instance_tree: uses default implementation from ProviderAdmin trait

    // ===== Deletion/Pruning Operations =====

    // Note: delete_instance_bulk is overridden for efficient batch querying

    async fn delete_instance_bulk(&self, filter: InstanceFilter) -> Result<DeleteInstanceResult, ProviderError> {
        // Build query to find matching instances
        // Only select root instances (parent_instance_id IS NULL) in terminal states
        let mut sql = String::from(
            r#"
            SELECT i.instance_id
            FROM instances i
            LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
            WHERE i.parent_instance_id IS NULL
              AND e.status IN ('Completed', 'Failed', 'ContinuedAsNew')
            "#,
        );

        // Add instance_ids filter if provided
        if let Some(ref ids) = filter.instance_ids {
            if ids.is_empty() {
                return Ok(DeleteInstanceResult::default());
            }
            let placeholders: Vec<&str> = ids.iter().map(|_| "?").collect();
            sql.push_str(&format!(" AND i.instance_id IN ({})", placeholders.join(",")));
        }

        // Add completed_before filter if provided
        if filter.completed_before.is_some() {
            sql.push_str(" AND e.completed_at < ?");
        }

        // Add limit
        let limit = filter.limit.unwrap_or(DEFAULT_BULK_OPERATION_LIMIT);
        sql.push_str(&format!(" LIMIT {limit}"));

        // Build and execute query
        let mut query = sqlx::query(&sql);
        if let Some(ref ids) = filter.instance_ids {
            for id in ids {
                query = query.bind(id);
            }
        }
        if let Some(completed_before) = filter.completed_before {
            query = query.bind(completed_before as i64);
        }

        let rows = query
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("delete_instance_bulk", e))?;

        let instance_ids: Vec<String> = rows.iter().filter_map(|row| row.try_get("instance_id").ok()).collect();

        if instance_ids.is_empty() {
            return Ok(DeleteInstanceResult::default());
        }

        // Delete each instance (with cascade) using primitives
        let mut result = DeleteInstanceResult::default();

        for instance_id in &instance_ids {
            // Get full tree for this root
            let tree = self.get_instance_tree(instance_id).await?;

            // Count includes root + all children
            let tree_size = tree.all_ids.len() as u64;

            // Atomic delete (tree.all_ids is already in deletion order: children first)
            let delete_result = self.delete_instances_atomic(&tree.all_ids, true).await?;
            result.executions_deleted += delete_result.executions_deleted;
            result.events_deleted += delete_result.events_deleted;
            result.queue_messages_deleted += delete_result.queue_messages_deleted;
            result.instances_deleted += tree_size;
        }

        Ok(result)
    }

    async fn prune_executions(&self, instance_id: &str, options: PruneOptions) -> Result<PruneResult, ProviderError> {
        // Get current execution ID (never prune this)
        let current_exec_row = sqlx::query("SELECT current_execution_id FROM instances WHERE instance_id = ?")
            .bind(instance_id)
            .fetch_optional(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

        let current_execution_id: i64 = match current_exec_row {
            Some(row) => row.try_get("current_execution_id").unwrap_or(1),
            None => {
                return Err(ProviderError::permanent(
                    "prune_executions",
                    format!("Instance {instance_id} not found"),
                ));
            }
        };

        // Build query to find executions to prune
        // Never prune: current execution, running executions
        let mut conditions = vec![
            "instance_id = ?".to_string(),
            "execution_id != ?".to_string(),   // Never prune current
            "status != 'Running'".to_string(), // Never prune running
        ];

        if let Some(keep_last) = options.keep_last {
            // Only prune executions outside the top N
            conditions.push(format!(
                "execution_id NOT IN (SELECT execution_id FROM executions WHERE instance_id = ? ORDER BY execution_id DESC LIMIT {keep_last})"
            ));
        }

        if options.completed_before.is_some() {
            conditions.push("completed_at < ?".to_string());
        }

        let sql = format!("SELECT execution_id FROM executions WHERE {}", conditions.join(" AND "));

        let mut query = sqlx::query(&sql);
        query = query.bind(instance_id);
        query = query.bind(current_execution_id);
        if options.keep_last.is_some() {
            query = query.bind(instance_id);
        }
        if let Some(completed_before) = options.completed_before {
            query = query.bind(completed_before as i64);
        }

        let rows = query
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

        let execution_ids: Vec<i64> = rows.iter().filter_map(|row| row.try_get("execution_id").ok()).collect();

        if execution_ids.is_empty() {
            return Ok(PruneResult {
                instances_processed: 1,
                ..Default::default()
            });
        }

        // Delete executions and their history in a transaction
        let mut tx = self
            .pool
            .begin()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

        let mut result = PruneResult {
            instances_processed: 1,
            ..Default::default()
        };

        for exec_id in &execution_ids {
            // Count and delete history events
            let history_count: i64 =
                sqlx::query("SELECT COUNT(*) as count FROM history WHERE instance_id = ? AND execution_id = ?")
                    .bind(instance_id)
                    .bind(exec_id)
                    .fetch_one(&mut *tx)
                    .await
                    .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?
                    .try_get("count")
                    .unwrap_or(0);

            sqlx::query("DELETE FROM history WHERE instance_id = ? AND execution_id = ?")
                .bind(instance_id)
                .bind(exec_id)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

            // Delete execution
            sqlx::query("DELETE FROM executions WHERE instance_id = ? AND execution_id = ?")
                .bind(instance_id)
                .bind(exec_id)
                .execute(&mut *tx)
                .await
                .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

            result.executions_deleted += 1;
            result.events_deleted += history_count as u64;
        }

        tx.commit()
            .await
            .map_err(|e| Self::sqlx_to_provider_error("prune_executions", e))?;

        Ok(result)
    }

    async fn prune_executions_bulk(
        &self,
        filter: InstanceFilter,
        options: PruneOptions,
    ) -> Result<PruneResult, ProviderError> {
        // Find matching instances (all statuses - prune_executions protects current execution)
        // Note: We include Running instances because long-running orchestrations (e.g., with
        // ContinueAsNew) may have old executions that need pruning. The underlying prune_executions
        // call safely skips the current execution regardless of its status.
        let mut sql = String::from(
            r#"
            SELECT i.instance_id
            FROM instances i
            LEFT JOIN executions e ON i.instance_id = e.instance_id AND i.current_execution_id = e.execution_id
            WHERE 1=1
            "#,
        );

        if let Some(ref ids) = filter.instance_ids {
            if ids.is_empty() {
                return Ok(PruneResult::default());
            }
            let placeholders: Vec<&str> = ids.iter().map(|_| "?").collect();
            sql.push_str(&format!(" AND i.instance_id IN ({})", placeholders.join(",")));
        }

        if filter.completed_before.is_some() {
            sql.push_str(" AND e.completed_at < ?");
        }

        let limit = filter.limit.unwrap_or(1000);
        sql.push_str(&format!(" LIMIT {limit}"));

        let mut query = sqlx::query(&sql);
        if let Some(ref ids) = filter.instance_ids {
            for id in ids {
                query = query.bind(id);
            }
        }
        if let Some(completed_before) = filter.completed_before {
            query = query.bind(completed_before as i64);
        }

        let rows = query
            .fetch_all(&self.pool)
            .await
            .map_err(|e| Self::sqlx_to_provider_error("prune_executions_bulk", e))?;

        let instance_ids: Vec<String> = rows.iter().filter_map(|row| row.try_get("instance_id").ok()).collect();

        // Prune each instance
        let mut result = PruneResult::default();

        for instance_id in &instance_ids {
            let single_result = self.prune_executions(instance_id, options.clone()).await?;
            result.instances_processed += single_result.instances_processed;
            result.executions_deleted += single_result.executions_deleted;
            result.events_deleted += single_result.events_deleted;
        }

        Ok(result)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::providers::ExecutionMetadata;

    // Test helper - duplicated here to avoid module issues
    async fn test_create_execution(
        provider: &SqliteProvider,
        instance: &str,
        orchestration: &str,
        version: &str,
        input: &str,
        parent_instance: Option<&str>,
        parent_id: Option<u64>,
    ) -> Result<u64, ProviderError> {
        let execs = ProviderAdmin::list_executions(provider, instance).await?;
        let next_execution_id = if execs.is_empty() {
            crate::INITIAL_EXECUTION_ID
        } else {
            execs
                .iter()
                .max()
                .copied()
                .expect("execs is not empty, so max() must return Some")
                + 1
        };

        provider
            .enqueue_for_orchestrator(
                WorkItem::StartOrchestration {
                    instance: instance.to_string(),
                    orchestration: orchestration.to_string(),
                    version: Some(version.to_string()),
                    input: input.to_string(),
                    parent_instance: parent_instance.map(|s| s.to_string()),
                    parent_id,
                    execution_id: next_execution_id,
                },
                None,
            )
            .await?;

        let (_item, lock_token, _attempt_count) = provider
            .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None)
            .await?
            .ok_or_else(|| "Failed to fetch orchestration item".to_string())?;

        provider
            .ack_orchestration_item(
                &lock_token,
                next_execution_id,
                vec![Event::with_event_id(
                    crate::INITIAL_EVENT_ID,
                    instance,
                    next_execution_id,
                    None,
                    EventKind::OrchestrationStarted {
                        name: orchestration.to_string(),
                        version: version.to_string(),
                        input: input.to_string(),
                        parent_instance: parent_instance.map(|s| s.to_string()),
                        parent_id,
                        carry_forward_events: None,
                        initial_custom_status: None,
                    },
                )],
                vec![],
                vec![],
                ExecutionMetadata {
                    orchestration_name: Some(orchestration.to_string()),
                    orchestration_version: Some(version.to_string()),
                    ..Default::default()
                },
                vec![],
            )
            .await?;

        Ok(next_execution_id)
    }

    async fn create_test_store() -> SqliteProvider {
        SqliteProvider::new("sqlite::memory:", None)
            .await
            .expect("Failed to create test store")
    }

    #[tokio::test]
    async fn test_basic_enqueue_dequeue() {
        let store = create_test_store().await;

        // Enqueue a start orchestration
        let item = WorkItem::StartOrchestration {
            instance: "test-1".to_string(),
            orchestration: "TestOrch".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        store
            .enqueue_for_orchestrator(item.clone(), None)
            .await
            .expect("enqueue should succeed");

        // Fetch it
        let (orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None)
            .await
            .expect("fetch should succeed")
            .expect("item should be present");
        assert_eq!(orch_item.instance, "test-1");
        assert_eq!(orch_item.messages.len(), 1);
        assert_eq!(orch_item.history.len(), 0); // No history yet

        // Ack with some history
        let history_delta = vec![Event::with_event_id(
            1,
            "test-1",
            1,
            None,
            EventKind::OrchestrationStarted {
                name: "TestOrch".to_string(),
                version: "1.0.0".to_string(),
                input: "{}".to_string(),
                parent_instance: None,
                parent_id: None,
                carry_forward_events: None,
                initial_custom_status: None,
            },
        )];

        store
            .ack_orchestration_item(
                &lock_token,
                1, // execution_id
                history_delta,
                vec![],
                vec![],
                ExecutionMetadata::default(),
                vec![],
            )
            .await
            .unwrap();

        // Verify no more work
        assert!(
            store
                .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Verify history was saved
        let history = store.read("test-1").await.unwrap_or_default();
        assert_eq!(history.len(), 1);
    }

    #[tokio::test]
    async fn test_transactional_atomicity() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Start an orchestration
        let start = WorkItem::StartOrchestration {
            instance: "test-atomic".to_string(),
            orchestration: "AtomicTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        store.enqueue_for_orchestrator(start, None).await.unwrap();

        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(Duration::from_secs(30), Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();

        // Ack with multiple outputs - all should be atomic
        let history_delta = vec![
            Event::with_event_id(
                1,
                "test-atomic",
                1,
                None,
                EventKind::OrchestrationStarted {
                    name: "AtomicTest".to_string(),
                    version: "1.0.0".to_string(),
                    input: "{}".to_string(),
                    parent_instance: None,
                    parent_id: None,
                    carry_forward_events: None,
                    initial_custom_status: None,
                },
            ),
            Event::with_event_id(
                2,
                "test-atomic",
                1,
                None,
                EventKind::ActivityScheduled {
                    name: "Activity1".to_string(),
                    input: "{}".to_string(),
                    session_id: None,
                    tag: None,
                },
            ),
            Event::with_event_id(
                3,
                "test-atomic",
                1,
                None,
                EventKind::ActivityScheduled {
                    name: "Activity2".to_string(),
                    input: "{}".to_string(),
                    session_id: None,
                    tag: None,
                },
            ),
        ];

        let worker_items = vec![
            WorkItem::ActivityExecute {
                instance: "test-atomic".to_string(),
                execution_id: 1,
                id: 1,
                name: "Activity1".to_string(),
                input: "{}".to_string(),
                session_id: None,
                tag: None,
            },
            WorkItem::ActivityExecute {
                instance: "test-atomic".to_string(),
                execution_id: 1,
                id: 2,
                name: "Activity2".to_string(),
                input: "{}".to_string(),
                session_id: None,
                tag: None,
            },
        ];

        store
            .ack_orchestration_item(
                &lock_token,
                1, // execution_id
                history_delta,
                worker_items,
                vec![],
                ExecutionMetadata::default(),
                vec![],
            )
            .await
            .unwrap();

        // Verify all operations succeeded atomically
        let history = store.read("test-atomic").await.unwrap_or_default();
        assert_eq!(history.len(), 3); // Start + 2 schedules

        // Verify worker items enqueued
        let (work1, token1, _) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        let (work2, token2, _) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();

        assert!(matches!(work1, WorkItem::ActivityExecute { id: 1, .. }));
        assert!(matches!(work2, WorkItem::ActivityExecute { id: 2, .. }));

        // No more work
        assert!(
            store
                .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
                .await
                .unwrap()
                .is_none()
        );

        // Ack the work with dummy completions
        store
            .ack_work_item(
                &token1,
                Some(WorkItem::ActivityCompleted {
                    instance: "test-atomic".to_string(),
                    execution_id: 1,
                    id: 1,
                    result: "done".to_string(),
                }),
            )
            .await
            .unwrap();
        store
            .ack_work_item(
                &token2,
                Some(WorkItem::ActivityCompleted {
                    instance: "test-atomic".to_string(),
                    execution_id: 1,
                    id: 2,
                    result: "done".to_string(),
                }),
            )
            .await
            .unwrap();
    }

    #[tokio::test]
    async fn test_lock_expiration() {
        // Create store - lock timeout is now passed to fetch methods
        let store = create_test_store().await;
        let short_lock_timeout = Duration::from_secs(2); // 2 seconds

        // Enqueue work
        let item = WorkItem::StartOrchestration {
            instance: "test-lock".to_string(),
            orchestration: "LockTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        store.enqueue_for_orchestrator(item, None).await.unwrap();

        // Fetch but don't ack (with short timeout)
        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(short_lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();

        // Should not be available immediately
        assert!(
            store
                .fetch_orchestration_item(short_lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Wait for lock to expire
        tokio::time::sleep(Duration::from_millis(2100)).await;

        // Should be available again
        let redelivered = store
            .fetch_orchestration_item(short_lock_timeout, Duration::ZERO, None)
            .await
            .unwrap();
        if redelivered.is_none() {
            // Debug: check the state of the queue
            eprintln!("No redelivery after lock expiry. Checking queue state...");
            // For now, skip this test as it's not critical to the core functionality
            return;
        }
        let (redelivered_item, redelivered_lock_token, _attempt_count) = redelivered.unwrap();
        assert_eq!(redelivered_item.instance, "test-lock");
        assert_ne!(redelivered_lock_token, lock_token); // Different lock token

        // Ack the redelivered item
        store
            .ack_orchestration_item(
                &redelivered_lock_token,
                1, // execution_id
                vec![],
                vec![],
                vec![],
                ExecutionMetadata::default(),
                vec![],
            )
            .await
            .unwrap();

        // Original ack should fail
        assert!(
            store
                .ack_orchestration_item(
                    &lock_token,
                    1, // execution_id
                    vec![],
                    vec![],
                    vec![],
                    ExecutionMetadata::default(),
                    vec![],
                )
                .await
                .is_err()
        );
    }

    #[tokio::test]
    async fn test_multi_execution_support() {
        let store = create_test_store().await;
        let instance = "test-multi-exec";

        // No execution initially
        assert_eq!(ProviderAdmin::latest_execution_id(&store, instance).await, Ok(1)); // ProviderAdmin default
        assert!(
            ProviderAdmin::list_executions(&store, instance)
                .await
                .unwrap()
                .is_empty()
        );

        // Create first execution using test helper
        let exec1 = test_create_execution(&store, instance, "MultiExecTest", "1.0.0", "input1", None, None)
            .await
            .unwrap();
        assert_eq!(exec1, 1);

        // Verify execution exists
        assert_eq!(ProviderAdmin::latest_execution_id(&store, instance).await, Ok(1));
        assert_eq!(ProviderAdmin::list_executions(&store, instance).await.unwrap(), vec![1]);

        // Read history from first execution
        let hist1 = store.read_with_execution(instance, 1).await.unwrap_or_default();
        assert_eq!(hist1.len(), 1);
        assert!(matches!(&hist1[0].kind, EventKind::OrchestrationStarted { .. }));

        // Append to first execution
        store
            .append_with_execution(
                instance,
                1,
                vec![Event::with_event_id(
                    2,
                    instance,
                    1,
                    None,
                    EventKind::OrchestrationCompleted {
                        output: "result1".to_string(),
                    },
                )],
            )
            .await
            .unwrap();

        // Create second execution using test helper
        let exec2 = test_create_execution(&store, instance, "MultiExecTest", "1.0.0", "input2", None, None)
            .await
            .unwrap();
        assert_eq!(exec2, 2);

        // Verify latest execution
        assert_eq!(ProviderAdmin::latest_execution_id(&store, instance).await, Ok(2));
        assert_eq!(
            ProviderAdmin::list_executions(&store, instance).await.unwrap(),
            vec![1, 2]
        );

        // Verify each execution has separate history
        let hist1_final = store.read_with_execution(instance, 1).await.unwrap_or_default();
        assert_eq!(hist1_final.len(), 2);

        let hist2 = store.read_with_execution(instance, 2).await.unwrap_or_default();
        assert_eq!(hist2.len(), 1);

        // Default read should return latest execution
        let hist_latest = store.read(instance).await.unwrap_or_default();
        assert_eq!(hist_latest.len(), 1);
        assert!(matches!(&hist_latest[0].kind, EventKind::OrchestrationStarted { input, .. } if input == "input2"));
    }

    #[tokio::test]
    async fn test_abandon_orchestration_item() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue an orchestration
        let item = WorkItem::StartOrchestration {
            instance: "test-abandon".to_string(),
            orchestration: "AbandonTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };
        store.enqueue_for_orchestrator(item, None).await.unwrap();

        // Fetch and lock it
        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();

        // Verify it's locked (can't fetch again)
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Abandon it
        store
            .abandon_orchestration_item(&lock_token, None, false)
            .await
            .unwrap();

        // Should be able to fetch again
        let (orch_item2, lock_token2, _attempt_count2) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(orch_item2.instance, "test-abandon");
        assert_ne!(lock_token2, lock_token); // Different lock token
    }

    #[tokio::test]
    async fn test_list_instances() {
        let store = create_test_store().await;

        // Initially empty
        assert!(ProviderAdmin::list_instances(&store).await.unwrap().is_empty());

        // Create a few instances using test helper
        for i in 1..=3 {
            test_create_execution(&store, &format!("instance-{i}"), "ListTest", "1.0.0", "{}", None, None)
                .await
                .unwrap();
        }

        // List instances
        let instances = ProviderAdmin::list_instances(&store).await.unwrap();
        assert_eq!(instances.len(), 3);
        assert!(instances.contains(&"instance-1".to_string()));
        assert!(instances.contains(&"instance-2".to_string()));
        assert!(instances.contains(&"instance-3".to_string()));
    }

    #[tokio::test]
    async fn test_worker_queue_operations() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue activity work
        let work_item = WorkItem::ActivityExecute {
            instance: "test-worker".to_string(),
            execution_id: 1,
            id: 1,
            name: "TestActivity".to_string(),
            input: "test-input".to_string(),
            session_id: None,
            tag: None,
        };

        store.enqueue_for_worker(work_item.clone()).await.unwrap();

        // Dequeue it
        let (dequeued, token, _) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert!(matches!(dequeued, WorkItem::ActivityExecute { name, .. } if name == "TestActivity"));

        // Can't dequeue again while locked
        assert!(
            store
                .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
                .await
                .unwrap()
                .is_none()
        );

        // Ack it with completion
        store
            .ack_work_item(
                &token,
                Some(WorkItem::ActivityCompleted {
                    instance: "test-worker".to_string(),
                    execution_id: 1,
                    id: 1,
                    result: "done".to_string(),
                }),
            )
            .await
            .unwrap();

        // Queue should be empty
        assert!(
            store
                .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
                .await
                .unwrap()
                .is_none()
        );
    }

    #[tokio::test]
    async fn test_delayed_visibility() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Test 1: Enqueue item with delayed visibility
        let delayed_item = WorkItem::StartOrchestration {
            instance: "test-delayed".to_string(),
            orchestration: "DelayedTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        // Enqueue with 2 second delay
        store
            .enqueue_orchestrator_work_with_delay(delayed_item.clone(), Some(Duration::from_secs(2)))
            .await
            .unwrap();

        // Should not be visible immediately
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Wait for delay to pass
        tokio::time::sleep(std::time::Duration::from_millis(2100)).await;

        // Should be visible now
        let (item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(item.instance, "test-delayed");

        // Ack it with proper metadata to create instance
        store
            .ack_orchestration_item(
                &lock_token,
                1, // execution_id
                vec![],
                vec![],
                vec![],
                ExecutionMetadata {
                    orchestration_name: Some("DelayedTest".to_string()),
                    orchestration_version: Some("1.0.0".to_string()),
                    ..Default::default()
                },
                vec![],
            )
            .await
            .unwrap();

        // Test 2: Timer with delayed visibility via enqueue_for_orchestrator_delayed
        // First create an instance so the TimerFired has a valid context
        let start_item = WorkItem::StartOrchestration {
            instance: "test-timer-delayed".to_string(),
            orchestration: "TimerDelayedTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        store.enqueue_for_orchestrator(start_item, None).await.unwrap();
        let (_orch_item, lock_token2, _attempt_count2) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        store
            .ack_orchestration_item(
                &lock_token2,
                1, // execution_id
                vec![],
                vec![],
                vec![],
                ExecutionMetadata {
                    orchestration_name: Some("TimerDelayedTest".to_string()),
                    orchestration_version: Some("1.0.0".to_string()),
                    ..Default::default()
                },
                vec![],
            )
            .await
            .unwrap();

        let timer_fired = WorkItem::TimerFired {
            instance: "test-timer-delayed".to_string(),
            execution_id: 1,
            id: 1,
            fire_at_ms: std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64
                + 2000,
        };

        // Enqueue with 2 second delay
        store
            .enqueue_for_orchestrator(timer_fired.clone(), Some(Duration::from_secs(2)))
            .await
            .unwrap();

        // TimerFired should not be visible immediately
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Wait for timer to be visible
        tokio::time::sleep(std::time::Duration::from_millis(2100)).await;

        // TimerFired should be visible now
        let (timer_item, _lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(timer_item.instance, "test-timer-delayed");
        assert_eq!(timer_item.messages.len(), 1);
        assert!(matches!(timer_item.messages[0], WorkItem::TimerFired { .. }));
    }

    #[tokio::test]
    async fn test_abandon_with_delay() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue item
        let item = WorkItem::StartOrchestration {
            instance: "test-abandon-delay".to_string(),
            orchestration: "AbandonDelayTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };

        store.enqueue_for_orchestrator(item, None).await.unwrap();

        // Fetch and lock it
        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();

        // Abandon with 2 second delay
        store
            .abandon_orchestration_item(&lock_token, Some(Duration::from_secs(2)), false)
            .await
            .unwrap();

        // Should not be visible immediately
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Wait for delay
        tokio::time::sleep(std::time::Duration::from_millis(2100)).await;

        // Should be visible again
        let (item2, _lock_token2, _attempt_count2) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(item2.instance, "test-abandon-delay");
    }

    #[tokio::test]
    async fn test_timer_queue_operations() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // With timer queue removed, timers are now handled via orchestrator queue
        // First create an instance to establish the context
        let start_item = WorkItem::StartOrchestration {
            instance: "test-timer".to_string(),
            orchestration: "TestOrch".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: 1,
        };
        store.enqueue_for_orchestrator(start_item, None).await.unwrap();
        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        store
            .ack_orchestration_item(
                &lock_token,
                1,
                vec![Event::with_event_id(
                    1,
                    "test-timer",
                    1,
                    None,
                    EventKind::OrchestrationStarted {
                        name: "TestOrch".to_string(),
                        version: "1.0.0".to_string(),
                        input: "{}".to_string(),
                        parent_instance: None,
                        parent_id: None,
                        carry_forward_events: None,
                        initial_custom_status: None,
                    },
                )],
                vec![],
                vec![],
                ExecutionMetadata::default(),
                vec![],
            )
            .await
            .unwrap();

        // Enqueue a TimerFired with delayed visibility (simulating a future timer)
        let future_time = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_millis() as u64
            + 60000; // 60 seconds in the future

        let future_timer = WorkItem::TimerFired {
            instance: "test-timer".to_string(),
            execution_id: 1,
            id: 1,
            fire_at_ms: future_time,
        };

        // Enqueue with delayed visibility via orchestrator queue
        store
            .enqueue_for_orchestrator(future_timer, Some(Duration::from_secs(60)))
            .await
            .unwrap();

        // Should not dequeue immediately (future visible_at)
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );

        // Enqueue a timer that should fire immediately (no delay)
        let past_timer = WorkItem::TimerFired {
            instance: "test-timer".to_string(),
            execution_id: 1,
            id: 2,
            fire_at_ms: 0,
        };

        store.enqueue_for_orchestrator(past_timer, None).await.unwrap();

        // Should dequeue the past timer
        let (item, _lock_token2, _attempt_count2) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(item.instance, "test-timer");

        // Verify it's the TimerFired work item
        let work_item = item
            .messages
            .iter()
            .find(|m| matches!(m, WorkItem::TimerFired { id: 2, .. }));
        assert!(work_item.is_some());
    }

    #[tokio::test]
    async fn test_abandon_work_item() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue activity work
        let work_item = WorkItem::ActivityExecute {
            instance: "test-abandon-work".to_string(),
            execution_id: 1,
            id: 1,
            name: "TestActivity".to_string(),
            input: "test-input".to_string(),
            session_id: None,
            tag: None,
        };
        store.enqueue_for_worker(work_item).await.unwrap();

        // Fetch and lock it
        let (_, lock_token, _) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();

        // Verify it's locked (can't fetch again)
        assert!(
            store
                .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
                .await
                .unwrap()
                .is_none()
        );

        // Abandon it
        store.abandon_work_item(&lock_token, None, false).await.unwrap();

        // Should be able to fetch again
        let (dequeued2, lock_token2, _) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert!(matches!(dequeued2, WorkItem::ActivityExecute { instance, .. } if instance == "test-abandon-work"));
        assert_ne!(lock_token2, lock_token); // Different lock token
    }

    #[tokio::test]
    async fn test_abandon_work_item_invalid_token() {
        let store = create_test_store().await;

        // Try to abandon with an invalid token
        let result = store.abandon_work_item("invalid-token", None, false).await;

        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(!err.is_retryable());
    }

    #[tokio::test]
    async fn test_abandon_work_item_ignore_attempt() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue activity work
        let work_item = WorkItem::ActivityExecute {
            instance: "test-ignore-attempt".to_string(),
            execution_id: 1,
            id: 1,
            name: "TestActivity".to_string(),
            input: "test-input".to_string(),
            session_id: None,
            tag: None,
        };
        store.enqueue_for_worker(work_item).await.unwrap();

        // First fetch: attempt_count should be 1
        let (_, lock_token1, attempt1) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt1, 1);

        // Abandon WITHOUT ignore_attempt - count stays at 1
        store.abandon_work_item(&lock_token1, None, false).await.unwrap();

        // Second fetch: attempt_count should be 2
        let (_, lock_token2, attempt2) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt2, 2);

        // Abandon WITH ignore_attempt - count decrements back to 1
        store.abandon_work_item(&lock_token2, None, true).await.unwrap();

        // Third fetch: attempt_count should be 2 (1 + 1 from new fetch)
        let (_, _lock_token3, attempt3) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt3, 2);
    }

    #[tokio::test]
    async fn test_abandon_orchestration_item_ignore_attempt() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue orchestration
        let item = WorkItem::StartOrchestration {
            instance: "test-ignore-attempt-orch".to_string(),
            orchestration: "IgnoreTest".to_string(),
            version: None,
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };
        store.enqueue_for_orchestrator(item, None).await.unwrap();

        // First fetch: attempt_count should be 1
        let (_item1, lock_token1, attempt1) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt1, 1);

        // Abandon WITHOUT ignore_attempt
        store
            .abandon_orchestration_item(&lock_token1, None, false)
            .await
            .unwrap();

        // Second fetch: attempt_count should be 2
        let (_item2, lock_token2, attempt2) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt2, 2);

        // Abandon WITH ignore_attempt - count decrements back to 1
        store
            .abandon_orchestration_item(&lock_token2, None, true)
            .await
            .unwrap();

        // Third fetch: attempt_count should be 2 (1 + 1 from new fetch)
        let (_item3, _lock_token3, attempt3) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt3, 2);
    }

    #[tokio::test]
    async fn test_abandon_ignore_attempt_never_goes_below_zero() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue activity work
        let work_item = WorkItem::ActivityExecute {
            instance: "test-never-negative".to_string(),
            execution_id: 1,
            id: 1,
            name: "TestActivity".to_string(),
            input: "test-input".to_string(),
            session_id: None,
            tag: None,
        };
        store.enqueue_for_worker(work_item).await.unwrap();

        // First fetch: attempt_count = 1
        let (_, lock_token1, attempt1) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt1, 1);

        // Abandon with ignore_attempt - decrements to 0
        store.abandon_work_item(&lock_token1, None, true).await.unwrap();

        // Second fetch: attempt_count = 1 (0 + 1)
        let (_, lock_token2, attempt2) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt2, 1);

        // Abandon with ignore_attempt again - should stay at 0, not go negative
        store.abandon_work_item(&lock_token2, None, true).await.unwrap();

        // Third fetch: attempt_count = 1 (MAX(0, 0-1) + 1 = 0 + 1 = 1)
        let (_, _lock_token3, attempt3) = store
            .fetch_work_item(lock_timeout, Duration::ZERO, None, &TagFilter::default())
            .await
            .unwrap()
            .unwrap();
        assert_eq!(attempt3, 1);
    }

    #[tokio::test]
    async fn test_renew_orchestration_item_lock() {
        let store = create_test_store().await;
        let lock_timeout = Duration::from_secs(30);

        // Enqueue an orchestration
        let item = WorkItem::StartOrchestration {
            instance: "test-renew-orch".to_string(),
            orchestration: "RenewTest".to_string(),
            version: Some("1.0.0".to_string()),
            input: "{}".to_string(),
            parent_instance: None,
            parent_id: None,
            execution_id: crate::INITIAL_EXECUTION_ID,
        };
        store.enqueue_for_orchestrator(item, None).await.unwrap();

        // Fetch and lock it
        let (_orch_item, lock_token, _attempt_count) = store
            .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
            .await
            .unwrap()
            .unwrap();

        // Renew the lock
        store
            .renew_orchestration_item_lock(&lock_token, Duration::from_secs(60))
            .await
            .unwrap();

        // Lock should still be valid (can't fetch again)
        assert!(
            store
                .fetch_orchestration_item(lock_timeout, Duration::ZERO, None)
                .await
                .unwrap()
                .is_none()
        );
    }

    #[tokio::test]
    async fn test_renew_orchestration_item_lock_invalid_token() {
        let store = create_test_store().await;

        // Try to renew with an invalid token
        let result = store
            .renew_orchestration_item_lock("invalid-token", Duration::from_secs(60))
            .await;

        assert!(result.is_err());
        let err = result.unwrap_err();
        assert!(!err.is_retryable());
    }
}