# Provider Implementation Guide

**For:** LLMs and humans implementing new Duroxide providers  
**Reference:** See `src/providers/sqlite.rs` for complete working implementation

---

## Quick Start

To implement a Provider for Duroxide, you need to:

1. Implement the `Provider` trait from `duroxide::providers`
2. Store orchestration history (append-only event log)
3. Manage 3 work queues with peek-lock semantics
4. Ensure atomic commits for orchestration turns

**Key Principle:** Providers are storage abstractions. The runtime owns orchestration logic.

## ⚠️ CRITICAL: ID Generation Contract

**The provider MUST NOT generate `execution_id` or `event_id` values.**

All IDs are generated by the runtime and passed to the provider:

- **`execution_id`**: Passed explicitly to `ack_orchestration_item()`
- **`event_id`**: Set in each `Event` within the `history_delta`

**The provider's role is to STORE these IDs, not generate them.**

### ✅ Correct Pattern
```rust
async fn ack_orchestration_item(
    execution_id: u64,        // ✅ Runtime provides this
    history_delta: Vec<Event>, // ✅ Each event has event_id set by runtime
) {
    // Validate runtime provided IDs
    for event in &history_delta {
        assert!(event.event_id() > 0, "event_id must be set by runtime");
    }
    
    // Store execution with runtime-provided ID
    db.insert_execution(instance, execution_id, "Running");
    
    // Store events with runtime-provided event_ids
    for event in history_delta {
        db.insert_event(instance, execution_id, event.event_id(), event);
    }
}
```

### ❌ Wrong Pattern (DO NOT DO THIS)
```rust
// WRONG - Provider generating execution_id
let execution_id = db.query("SELECT MAX(execution_id) + 1 FROM executions")?;

// WRONG - Provider generating event_id
let event_id = db.query("SELECT MAX(event_id) + 1 FROM history")?;
for mut event in history_delta {
    event.set_event_id(event_id);  // ❌ Never modify event_id!
    event_id += 1;
}
```

---

## Complete Implementation Template

```rust
use duroxide::providers::{Provider, ProviderError, WorkItem, OrchestrationItem, ExecutionMetadata};
use duroxide::Event;
use std::sync::Arc;

pub struct MyProvider {
    // Your storage client (e.g., connection pool)
    // Lock timeout (recommended: 30 seconds)
    // Any configuration
}

#[async_trait::async_trait]
impl Provider for MyProvider {
    // === REQUIRED: Core Orchestration Methods ===
    
    async fn fetch_orchestration_item(
        &self,
        lock_timeout: Duration,
        poll_timeout: Duration,
    ) -> Result<Option<(OrchestrationItem, String, u32)>, ProviderError> {
        // 1. Find next available message in orchestrator queue
        // 2. Lock ALL messages for that instance
        // 3. Load instance metadata (name, version, execution_id)
        // 4. Load history for current execution_id
        // 5. Increment attempt_count on the locked messages
        // 6. Return (OrchestrationItem, lock_token, attempt_count)
        //
        // poll_timeout: Maximum time to wait for work. Providers that support
        // long polling MAY block up to this duration. Short-polling providers
        // (like SQLite) should ignore this and return immediately.
        //
        // attempt_count: Number of times this item has been fetched (for poison detection)
        
        todo!("See detailed docs below")
    }
    
    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: ExecutionMetadata,
    ) -> Result<(), ProviderError> {
        // ALL operations must be atomic (single transaction):
        // 1. Idempotently create execution row for the explicit execution_id (INSERT OR IGNORE)
        // 2. Update instances.current_execution_id = MAX(current_execution_id, execution_id)
        // 3. Append history_delta to the event log for that execution_id
        // 4. Update execution status/output from metadata (DON'T inspect events!)
        // 5. Enqueue worker_items to worker queue
        // 6. Enqueue orchestrator_items to orchestrator queue (may include TimerFired with visible_at delay)
        // 7. Delete locked messages (release lock)
        
        todo!("See detailed docs below")
    }
    
    async fn abandon_orchestration_item(&self, lock_token: &str, delay: Option<Duration>, ignore_attempt: bool) -> Result<(), ProviderError> {
        // Clear lock_token from messages
        // Optionally delay visibility for backoff
        // If ignore_attempt is true, decrement attempt_count (never below 0)
        
        todo!("See detailed docs below")
    }
    
    // === REQUIRED: History Access ===
    
    async fn read(&self, instance: &str) -> Vec<Event> {
        // Return events for LATEST execution, ordered by event_id
        // Return empty Vec if instance doesn't exist
        
        todo!()
    }
    
    async fn append_with_execution(
        &self,
        instance: &str,
        execution_id: u64,
        new_events: Vec<Event>,
    ) -> Result<(), ProviderError> {
        // Append events to history for specified execution
        // DO NOT modify event_ids (runtime assigns these)
        // Reject or ignore duplicates (same event_id)
        
        todo!()
    }
    
    // === REQUIRED: Worker Queue ===
    
    async fn enqueue_for_worker(&self, item: WorkItem) -> Result<(), ProviderError> {
        // Add item to worker queue
        // New items should have lock_token = NULL
        
        todo!()
    }
    
    async fn fetch_work_item(
        &self,
        lock_timeout: Duration,
        poll_timeout: Duration,
    ) -> Result<Option<(WorkItem, String, u32)>, ProviderError> {
        // Find next unlocked item
        // Lock it with unique token
        // Increment and return attempt_count
        // Return Ok(Some((item, token, attempt_count))) if found, Ok(None) if empty
        // Item stays in queue until acked
        //
        // poll_timeout: Maximum time to wait for work. Providers that support
        // long polling MAY block up to this duration. Short-polling providers
        // should ignore this and return immediately.
        //
        // attempt_count: Number of times this item has been fetched (for poison detection)
        
        todo!()
    }
    
    async fn ack_worker(&self, token: &str, completion: WorkItem) -> Result<(), ProviderError> {
        // Atomically:
        // 1. Delete item from worker queue (WHERE lock_token = token)
        // 2. Enqueue completion (ActivityCompleted or ActivityFailed) to orchestrator queue
        // MUST be atomic - prevents lost completions or duplicate work
        
        todo!()
    }
    
    async fn renew_work_item_lock(&self, token: &str, extend_for: Duration) -> Result<(), ProviderError> {
        // Extend lock timeout for in-flight activity
        // Called automatically by worker dispatcher for long-running activities
        // 
        // UPDATE worker_queue
        // SET locked_until = now() + extend_for
        // WHERE lock_token = token AND locked_until > now()
        //
        // Return error if token invalid/expired
        
        todo!()
    }
    
    async fn abandon_work_item(&self, token: &str, delay: Option<Duration>, ignore_attempt: bool) -> Result<(), ProviderError> {
        // Release work item lock without completing
        // Called when activity execution fails and should be retried
        //
        // 1. Clear lock_token from the work item
        // 2. Set visible_at = now() + delay (if delay provided) for backoff
        // 3. If ignore_attempt is true, decrement attempt_count (never below 0)
        // 4. Item becomes available for re-fetch
        //
        // Silently ignore if token not found (idempotent)
        
        todo!()
    }
    
    async fn renew_orchestration_item_lock(&self, token: &str, extend_for: Duration) -> Result<(), ProviderError> {
        // Extend lock timeout for in-flight orchestration turn
        // Called automatically by orchestration dispatcher during long turns
        //
        // UPDATE instance_locks
        // SET locked_until = now() + extend_for
        // WHERE lock_token = token AND locked_until > now()
        //
        // Return error if token invalid/expired
        
        todo!()
    }
    
    // === REQUIRED: Orchestrator Queue ===
    
    async fn enqueue_orchestrator_work(&self, item: WorkItem, delay: Option<Duration>) -> Result<(), ProviderError> {
        // Extract instance from item (see WorkItem docs)
        // Set visible_at = now() + delay.unwrap_or(Duration::ZERO)
        // Enqueue work item to orchestrator_queue
        // 
        // ⚠️ CRITICAL: DO NOT create instance here - runtime will create it via ack_orchestration_item metadata
        // This applies to ALL work items: StartOrchestration, ContinueAsNew, and sub-orchestrations
        // Instance creation happens ONLY in ack_orchestration_item() when metadata is provided
        
        todo!()
    }
    
    // === OPTIONAL: Multi-Execution Support ===
    
    async fn latest_execution_id(&self, instance: &str) -> Option<u64> {
        // Return MAX(execution_id) for instance. Override for performance.
        None
    }
    
    async fn read_with_execution(&self, instance: &str, _execution_id: u64) -> Vec<Event> {
        // Return events for specific execution_id (not just latest). Used for debugging/testing.
        // Default can delegate to read(); production providers should override.
        self.read(instance).await
    }
    
    // === OPTIONAL: Management APIs ===
    
    async fn list_instances(&self) -> Vec<String> {
        Vec::new()  // Override if you want instance listing
    }
    
    async fn list_executions(&self, instance: &str) -> Vec<u64> {
        let h = self.read(instance).await;
        if h.is_empty() { Vec::new() } else { vec![1] }
    }
}
```

---

## Error Handling with ProviderError

**⚠️ CRITICAL: All Provider methods return `Result<..., ProviderError>`**

Providers must return `ProviderError` instead of `String` to enable smart retry logic in the runtime. The runtime uses `ProviderError::is_retryable()` to decide whether to retry operations automatically.

**Special case: `fetch_work_item()` returns `Result<Option<(WorkItem, String)>, ProviderError>`**

The `fetch_work_item()` method has a special return type that distinguishes between:
- Empty queue: `Ok(None)` - No items available (normal, not an error)
- Success: `Ok(Some((item, token)))` - Successfully fetched and locked an item
- Storage failure: `Err(ProviderError)` - Database error, connection failure, etc.

This allows the runtime to properly handle backoff (empty queue) vs retry (storage errors).

### ProviderError Structure

```rust
use duroxide::providers::ProviderError;

// Retryable (transient) errors - runtime will retry automatically
ProviderError::retryable("ack_orchestration_item", "Database is busy")

// Non-retryable (permanent) errors - runtime will fail immediately
ProviderError::permanent("ack_orchestration_item", "Duplicate event detected")
```

### Error Classification

**Retryable errors (`is_retryable = true`):**
- Database busy/locked (e.g., SQLITE_BUSY)
- Connection timeouts
- Network failures
- Temporary resource exhaustion

**Non-retryable errors (`is_retryable = false`):**
- Data corruption (missing instance, invalid format)
- Duplicate events (indicates bug)
- Invalid input (malformed work item)
- Configuration errors
- Invalid lock tokens (idempotent - already processed)

### Converting Database Errors

```rust
use duroxide::providers::ProviderError;

// Example: Converting sqlx::Error to ProviderError
fn sqlx_to_provider_error(operation: &str, e: sqlx::Error) -> ProviderError {
    match e {
        sqlx::Error::Database(db_err) => {
            if db_err.code() == Some("SQLITE_BUSY") {
                ProviderError::retryable(operation, format!("Database busy: {}", e))
            } else if db_err.message().contains("UNIQUE constraint") {
                ProviderError::permanent(operation, format!("Duplicate detected: {}", e))
            } else {
                ProviderError::permanent(operation, format!("Database error: {}", e))
            }
        }
        sqlx::Error::PoolClosed | sqlx::Error::PoolTimedOut => {
            ProviderError::retryable(operation, format!("Connection pool error: {}", e))
        }
        _ => ProviderError::permanent(operation, format!("Unexpected error: {}", e)),
    }
}
```

### Runtime Retry Behavior

The runtime automatically retries retryable errors with exponential backoff:
- **Retryable errors**: Retried up to 5 times with exponential backoff (10ms, 20ms, 40ms, 80ms, 160ms)
- **Non-retryable errors**: Fail immediately, orchestration marked as failed with infrastructure error

If a retryable error persists after all retries, it is treated as an infrastructure failure and the orchestration is marked as failed.

---

## Detailed Implementation Guide

### 1. fetch_orchestration_item() - The Orchestration Turn Fetcher

**Purpose:** Atomically fetch and lock work for one instance.

**⚠️ CRITICAL: Instance-Level Locking**

Before fetching messages, you MUST acquire an instance-level lock. This prevents concurrent dispatchers from processing the same instance simultaneously, which would cause data corruption and race conditions.

**Instance Lock Schema:**
```sql
CREATE TABLE instance_locks (
    instance_id TEXT PRIMARY KEY,
    lock_token TEXT NOT NULL,
    locked_until INTEGER NOT NULL,  -- Unix timestamp (milliseconds)
    locked_at INTEGER NOT NULL      -- Unix timestamp (milliseconds)
);
```

**Lock Acquisition Logic:**
1. **Find available instance**: Select an instance that has work AND is not currently locked
2. **Atomically acquire lock**: Use `INSERT ... ON CONFLICT` to atomically claim the lock
3. **Verify lock acquired**: Check rows_affected to ensure lock was actually acquired
4. **Fetch messages**: Only after lock is confirmed, fetch messages for that instance

**Why Instance-Level Locking?**

- **Prevents race conditions**: Without instance locks, two dispatchers could fetch the same instance simultaneously
- **Ensures atomicity**: All messages for an instance are processed together in one turn
- **Prevents duplicate processing**: Lock token validates that only the lock holder can ack
- **Handles crashes**: Lock expiration allows automatic recovery when dispatchers crash

**Pseudo-code:**
```
BEGIN TRANSACTION

// Step 1: Find an instance that has work AND is not locked
// Join orchestrator_queue with instance_locks to check lock status
row = SELECT q.instance_id
      FROM orchestrator_queue q
      LEFT JOIN instance_locks il ON q.instance_id = il.instance_id
      WHERE q.visible_at <= current_timestamp()
        AND (il.instance_id IS NULL OR il.locked_until <= current_timestamp())
      ORDER BY q.id ASC
      LIMIT 1

IF row IS NULL:
    ROLLBACK
    RETURN None

instance_id = row.instance_id

// Step 2: Generate lock token and calculate expiration
lock_token = generate_uuid()
now_ms = current_timestamp_millis()
// lock_timeout is a Duration - convert to milliseconds for storage
locked_until = now_ms + lock_timeout.as_millis()  // e.g., now + 30000ms (30s from runtime)

// Step 3: Atomically acquire instance lock
// This is CRITICAL - must be atomic to prevent race conditions
lock_result = INSERT INTO instance_locks (instance_id, lock_token, locked_until, locked_at)
              VALUES (instance_id, lock_token, locked_until, now_ms)
              ON CONFLICT(instance_id) DO UPDATE
              SET lock_token = lock_token, locked_until = locked_until, locked_at = now_ms
              WHERE locked_until <= now_ms  // Only update if lock expired

IF lock_result.rows_affected == 0:
    // Lock acquisition failed - another dispatcher has the lock
    ROLLBACK
    RETURN None

// Step 4: Mark ALL messages for this instance with our lock_token
// This "tags" messages so we know which ones to delete on ack
UPDATE orchestrator_queue
SET lock_token = lock_token, locked_until = locked_until
WHERE instance_id = instance_id
  AND visible_at <= current_timestamp()
  AND (lock_token IS NULL OR locked_until <= current_timestamp())

// Step 5: Fetch all locked messages
messages = SELECT work_item FROM orchestrator_queue
           WHERE lock_token = lock_token
           ORDER BY id ASC

messages = messages.map(|m| deserialize_workitem(m))

// Step 6: Load instance metadata
metadata = SELECT orchestration_name, orchestration_version, current_execution_id
           FROM instances
           WHERE instance_id = instance_id

IF metadata IS NULL:
    // New instance - derive from work items or history
    // Search ALL work items for StartOrchestration or ContinueAsNew (not just first)
    start_item = messages.iter().find(|item| 
        matches!(item, StartOrchestration { .. } | ContinueAsNew { .. })
    )
    
    IF start_item IS StartOrchestration OR ContinueAsNew:
        orchestration_name = start_item.orchestration
        orchestration_version = start_item.version  // May be None - runtime will resolve
        current_execution_id = 1
    ELSE IF history is not empty:
        // Extract from OrchestrationStarted event in history
        orchestration_name = extract_from_history(history)
        orchestration_version = extract_from_history(history)  // May be NULL
        current_execution_id = 1
    ELSE:
        // Shouldn't happen - check history as fallback
        RETURN None or build from history

orchestration_name = metadata.orchestration_name
orchestration_version = metadata.orchestration_version  // May be NULL
current_execution_id = metadata.current_execution_id

// ⚠️ CRITICAL: Handle NULL version
// If version is NULL, use "unknown" - don't try to extract from history
// 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)
IF orchestration_version IS NULL:
    orchestration_version = "unknown"
    // Optional: Add debug assertion that history doesn't contain OrchestrationStarted
    // This helps catch data inconsistencies during development
    ASSERT NOT EXISTS (
        SELECT 1 FROM history 
        WHERE instance_id = instance_id 
          AND event_type = 'OrchestrationStarted'
    )

// Step 7: Load history for current execution
history_rows = SELECT event_data FROM history
               WHERE instance_id = instance_id
                 AND execution_id = current_execution_id
               ORDER BY event_id ASC

history = history_rows.map(|row| deserialize_event(row.event_data))

COMMIT TRANSACTION

RETURN Some(OrchestrationItem {
    instance: instance_id,
    orchestration_name,
    execution_id: current_execution_id,
    version: orchestration_version,
    history,
    messages,
    lock_token,
})
```

**Edge Cases:**
- No work available → Return None (dispatcher will sleep)
- Lock contention → Transaction retry or return None (another dispatcher has lock)
- Lock acquisition failed → Return None (try again later)
- Missing instance metadata → Derive from messages or history
- Empty history → Valid (new instance)

**Lock Expiration and Renewal:**
- Locks must expire after `lock_timeout` (passed from RuntimeOptions: 5s for orchestrator, 30s for worker)
- Expired locks allow automatic recovery from crashed dispatchers
- **Worker lock renewal:** Runtime automatically extends locks for long-running activities
  - For timeouts ≥15s: renewal at `(timeout - buffer_secs)`
  - For timeouts <15s: renewal at `0.5 * timeout`
  - Implemented via `renew_work_item_lock()` method
- Check `locked_until <= current_timestamp()` when acquiring locks
- Always validate lock token on ack/renew to ensure lock hasn't expired

**⚠️ CRITICAL: Successful Ack Releases Lock Immediately**

When `ack_orchestration_item()` succeeds, it MUST immediately release the instance lock (delete from `instance_locks` table). This happens BEFORE the transaction commits, ensuring the lock is released atomically with the ack.

**Why this matters:**
- Allows new messages to be fetched immediately after ack
- Prevents unnecessary lock expiration waits
- Enables efficient processing pipeline

**⚠️ CRITICAL: Lock Expiration During Ack**

If a lock expires while `ack_orchestration_item()` is executing, the ack MUST fail with an error. The lock validation step (Step 1-2) must check `locked_until > current_timestamp()`.

**Why this matters:**
- Prevents acks with expired locks from succeeding
- Detects race conditions where lock expired during processing
- Ensures proper lock ownership semantics

**Multi-Thread Safety:**
- Instance locks MUST be acquired atomically (single SQL statement)
- Use database-level locking (e.g., SQLite row locks, PostgreSQL SELECT FOR UPDATE)
- Never check-then-set - always use atomic operations
- Verify lock acquisition with `rows_affected` check

**⚠️ CRITICAL: Lock Token Uniqueness**

Each call to `fetch_orchestration_item()` MUST generate a unique lock token. Lock tokens must be:
- Unique across all instances and dispatchers
- Cryptographically random (use UUID v4 or similar)
- Never reused, even after lock expiration

**Why this matters:**
- Prevents lock token collisions
- Enables proper lock validation
- Allows tracking which dispatcher owns which lock

**⚠️ CRITICAL: Message Tagging During Lock**

When `fetch_orchestration_item()` locks an instance, it MUST mark ALL visible messages for that instance with the lock_token. This "tags" messages so that only these messages are deleted during `ack_orchestration_item()`.

**Critical behavior:**
- Messages that arrive AFTER fetch are NOT tagged with the lock_token
- Only messages present at fetch time are deleted on ack
- New messages remain in queue for the next fetch after lock is released
- This ensures no messages are lost and no duplicate processing occurs

**Why this matters:**
- Prevents losing messages that arrive during processing
- Ensures each message is processed exactly once
- Allows incremental message processing across multiple turns

**Example:**
```
1. Instance has messages [A, B] visible
2. fetch_orchestration_item() tags [A, B] with lock_token "abc"
3. During processing, message C arrives (not tagged)
4. ack_orchestration_item() deletes only [A, B] (those with lock_token "abc")
5. Next fetch gets message C
```

**⚠️ CRITICAL: Cross-Instance Lock Isolation**

Locks on one instance MUST NOT block fetching work for other instances. Each instance has its own independent lock.

**Why this matters:**
- Enables parallel processing of different instances
- Prevents one slow instance from blocking all work
- Essential for scalability

**⚠️ CRITICAL: Completions Arriving During Lock Are Blocked**

If a completion (e.g., ActivityCompleted) arrives for an instance that is currently locked, it MUST NOT be fetchable by other dispatchers until the lock is released (via ack or expiration).

**Why this matters:**
- Ensures all messages for an instance are processed together
- Prevents race conditions from concurrent dispatchers
- Maintains message ordering guarantees

**Implementation:**
- Instance-level lock prevents ANY fetch for that instance
- New messages are enqueued but not visible until lock released
- Lock release happens via successful ack or expiration

**Example SQLite Implementation:**
```sql
-- Atomic lock acquisition with conflict handling
INSERT INTO instance_locks (instance_id, lock_token, locked_until, locked_at)
VALUES (?, ?, ?, ?)
ON CONFLICT(instance_id) DO UPDATE
SET lock_token = excluded.lock_token,
    locked_until = excluded.locked_until,
    locked_at = excluded.locked_at
WHERE locked_until <= excluded.locked_at  -- Only update if expired
```

**Example PostgreSQL Implementation:**
```sql
-- Use SELECT FOR UPDATE SKIP LOCKED for atomic lock acquisition
SELECT instance_id FROM orchestrator_queue
WHERE visible_at <= NOW()
  AND instance_id NOT IN (
    SELECT instance_id FROM instance_locks WHERE locked_until > NOW()
  )
ORDER BY id
LIMIT 1
FOR UPDATE SKIP LOCKED;

-- Then insert lock
INSERT INTO instance_locks (instance_id, lock_token, locked_until, locked_at)
VALUES (?, ?, ?, ?)
ON CONFLICT (instance_id) DO UPDATE
SET lock_token = excluded.lock_token,
    locked_until = excluded.locked_until,
    locked_at = excluded.locked_at
WHERE instance_locks.locked_until <= excluded.locked_at;
```

---

### 2. ack_orchestration_item() - The Atomic Commit

**Purpose:** Atomically commit all changes from an orchestration turn.

**⚠️ CRITICAL: Instance Creation via Metadata**

**Providers MUST NOT create instances when enqueueing work items.** Instead, instances are created by the runtime through `ack_orchestration_item()` using `ExecutionMetadata`.

**Why this design:**
- **Single source of truth**: Runtime resolves version from registry, provider stores it
- **No premature creation**: Instances created only when runtime acknowledges first turn
- **Version resolution**: Runtime determines correct version, not provider
- **Consistent behavior**: All instance creation follows same pattern (via metadata)

**Instance creation flow:**
1. Client enqueues `StartOrchestration` work item (no instance created)
2. Runtime fetches work item via `fetch_orchestration_item()` (instance doesn't exist yet)
3. Runtime resolves version from registry and creates `OrchestrationStarted` event
4. Runtime computes `ExecutionMetadata` with `orchestration_name` and `orchestration_version`
5. Runtime calls `ack_orchestration_item()` with metadata
6. Provider creates instance from metadata (INSERT OR IGNORE + UPDATE pattern)

**Implementation pattern:**
```rust
// In ack_orchestration_item(), after lock validation:

if let (Some(name), Some(version)) = (&metadata.orchestration_name, &metadata.orchestration_version) {
    // Create instance if it doesn't exist (first ack)
    INSERT OR IGNORE INTO instances 
    (instance_id, orchestration_name, orchestration_version, current_execution_id)
    VALUES (instance_id, name, version, execution_id)
    
    // Update instance with resolved version
    UPDATE instances 
    SET orchestration_name = name, orchestration_version = version
    WHERE instance_id = instance_id
}
```

**Important:**
- `orchestration_version` may be NULL initially (before first ack)
- Runtime always provides version in metadata for new instances
- Sub-orchestrations follow same pattern - instance created on their first ack

---

**⚠️ CRITICAL: Duplicate Event Detection**

The provider **MUST** reject duplicate events. Events with the same `(instance_id, execution_id, event_id)` tuple should cause `ack_orchestration_item()` to return an error.

**Why this matters:**
- Prevents data corruption from retry attempts
- Ensures idempotency guarantees
- Detects runtime bugs or provider implementation issues

**Why duplicates are invalid:**
- The runtime performs duplicate detection before calling `ack_orchestration_item()` by checking the existing history
- If a duplicate event reaches the provider, it indicates a serious problem (runtime bug, retry logic failure, or provider state corruption)
- Providers should treat duplicates as errors, not silently ignore them

**Implementation Pattern:**
- Use a PRIMARY KEY constraint on `(instance_id, execution_id, event_id)` in your history table
- When inserting events, allow the database to enforce uniqueness
- If a duplicate is detected (constraint violation), return an error from `ack_orchestration_item()`

**SQLite Example:**
```sql
CREATE TABLE history (
    instance_id TEXT NOT NULL,
    execution_id INTEGER NOT NULL,
    event_id INTEGER NOT NULL,
    event_data TEXT NOT NULL,
    PRIMARY KEY (instance_id, execution_id, event_id)  -- Enforces uniqueness
);

-- When inserting:
INSERT INTO history (instance_id, execution_id, event_id, event_data)
VALUES (?, ?, ?, ?)
-- If duplicate exists, database returns constraint violation error
```

**Error Handling:**
- Return `Err(ProviderError::permanent("ack_orchestration_item", "duplicate event_id detected"))` when duplicate detected
- Do NOT silently ignore duplicates (`ON CONFLICT DO NOTHING` is wrong for history)
- Do NOT update existing events (history is append-only)

**Pseudo-code:**
```
BEGIN TRANSACTION

// Step 1: Validate lock token and get instance_id
// MUST check instance_locks table to verify lock is still valid
instance_id = SELECT instance_id FROM instance_locks
              WHERE lock_token = lock_token
                AND locked_until > current_timestamp()

IF instance_id IS NULL:
    ROLLBACK
    RETURN Err(ProviderError::permanent("ack_orchestration_item", "Invalid or expired lock token"))

// Step 2: Verify lock token matches instance
// This prevents using a lock token from a different instance
lock_valid = SELECT COUNT(*) FROM instance_locks 
             WHERE instance_id = instance_id 
               AND lock_token = lock_token 
               AND locked_until > current_timestamp()

IF lock_valid == 0:
    ROLLBACK
    RETURN Err(ProviderError::permanent("ack_orchestration_item", "Instance lock expired or invalid"))

// Step 3: Remove instance lock (processing complete)
DELETE FROM instance_locks 
WHERE instance_id = instance_id 
  AND lock_token = lock_token

// Use the explicit execution_id provided by the runtime

// 4. Create or update instance metadata from runtime-provided metadata
// Runtime resolves version from registry and provides it via metadata
// ⚠️ CRITICAL: Instance creation happens HERE, not in enqueue_orchestrator_work()
IF metadata.orchestration_name IS NOT NULL AND metadata.orchestration_version IS NOT NULL:
    // Step 4a: Ensure instance exists (creates if new, ignores if exists)
    // Use INSERT OR IGNORE to handle race conditions where instance might be created concurrently
    INSERT OR IGNORE INTO instances 
    (instance_id, orchestration_name, orchestration_version, current_execution_id)
    VALUES (instance_id, metadata.orchestration_name, metadata.orchestration_version, execution_id)
    
    // Step 4b: Update instance with resolved version (will update if instance exists, no-op if just created)
    // This ensures version is always set correctly, even if instance was created in step 4a
    UPDATE instances 
    SET orchestration_name = metadata.orchestration_name,
        orchestration_version = metadata.orchestration_version
    WHERE instance_id = instance_id
ELSE:
    // If metadata is missing, skip instance creation
    // This can happen for edge cases where runtime doesn't provide metadata
    // Instance will be created on next ack when metadata is available

// 5. Idempotently create execution row and update instance pointer
INSERT INTO executions (instance_id, execution_id, status, started_at)
VALUES (instance_id, execution_id, 'Running', CURRENT_TIMESTAMP)
ON CONFLICT DO NOTHING;

UPDATE instances
SET current_execution_id = MAX(current_execution_id, execution_id)
WHERE instance_id = instance_id;

// 6. Append history_delta (DO NOT inspect events!)
FOR event IN history_delta:
    // Validate event_id was set by runtime
    IF event.event_id() == 0:
        RETURN Err(ProviderError::permanent("ack_orchestration_item", "event_id must be set by runtime"))
    
    event_json = serialize(event)
    event_type = extract_discriminant_name(event)  // For indexing only
    
    INSERT INTO history (instance_id, execution_id, event_id, event_type, event_data, created_at)
    VALUES (instance_id, execution_id, event.event_id(), event_type, event_json, NOW())
    // If duplicate exists, PRIMARY KEY constraint will cause INSERT to fail
    // This is CORRECT - providers should error on duplicate events
    // DO NOT use ON CONFLICT DO NOTHING for history table

// 7. Update execution metadata (NO event inspection!)
IF metadata.status IS NOT NULL:
    UPDATE executions
    SET status = metadata.status,
        output = metadata.output,
        completed_at = CURRENT_TIMESTAMP
    WHERE instance_id = instance_id
      AND execution_id = execution_id

// (No implicit next-execution creation here; runtime controls execution_id explicitly)

// 8. Enqueue worker items
FOR item IN worker_items:
    work_json = serialize(item)
    INSERT INTO worker_queue (work_item, lock_token, locked_until, created_at)
    VALUES (work_json, NULL, NULL, NOW())

// 9. Enqueue orchestrator items (may include TimerFired with delayed visibility)
FOR item IN orchestrator_items:
    work_json = serialize(item)
    target_instance = extract_instance(item)  // See WorkItem docs
    
    // Set visible_at based on item type
    visible_at = IF item IS TimerFired:
                     item.fire_at_ms  // Delayed visibility for timers
                 ELSE:
                     NOW()  // Immediate visibility for other items
    
    // ⚠️ CRITICAL: DO NOT create instance here - even for StartOrchestration or ContinueAsNew
    // Instance creation happens ONLY via ack_orchestration_item metadata (step 4 above)
    // This applies to:
    //   - StartOrchestration work items (main orchestrations)
    //   - ContinueAsNew work items (execution restarts)
    //   - StartOrchestration work items in orchestrator_items (sub-orchestrations)
    // All instance creation is deferred to runtime via ack_orchestration_item metadata
    
    INSERT INTO orchestrator_queue (instance_id, work_item, visible_at, lock_token, locked_until, created_at)
    VALUES (target_instance, work_json, visible_at, NULL, NULL, NOW())

// 10. Release lock (delete acknowledged messages)
// Delete messages that were tagged with our lock_token
DELETE FROM orchestrator_queue WHERE lock_token = lock_token

COMMIT TRANSACTION
RETURN Ok(())
```

**Critical:**
- All 10 steps must be in ONE transaction
- Lock validation MUST occur before any other operations
- If lock expired or invalid, ROLLBACK immediately
- Instance lock MUST be deleted before committing (prevents stale locks)
- If ANY step fails, ROLLBACK everything
- Never partially commit

**⚠️ CRITICAL: Lock Released Only on Successful Ack**

The instance lock MUST only be released if the entire ack operation succeeds. If any step fails (e.g., duplicate event, constraint violation, database error), the transaction must rollback AND the lock must remain held.

**Why this matters:**
- Prevents partial state if ack fails mid-way
- Ensures failed acks don't silently lose work
- Allows retry of failed operations with same lock token
- Prevents other dispatchers from fetching work while ack is in progress

**Implementation:**
- Instance lock deletion (Step 3) must be INSIDE the transaction
- If transaction rolls back, lock remains held
- Lock expiration allows automatic recovery if dispatcher crashes during ack
- Subsequent fetches will block until lock expires or is released via successful ack

**⚠️ CRITICAL: Concurrent Ack Prevention**

If multiple dispatchers attempt to ack with the same lock token simultaneously, only ONE should succeed. The others must fail with an error indicating the lock token is invalid or expired.

**Why this matters:**
- Prevents duplicate processing if same lock token is reused
- Ensures idempotency guarantees
- Detects programming errors (same lock token used twice)

**Implementation:**
- Lock validation (Step 1-2) must check lock_token AND locked_until
- After lock validation, immediately delete the instance lock
- This prevents second ack from succeeding (lock no longer exists)
- Use database-level locking to ensure atomicity across concurrent attempts

---

### 3. abandon_orchestration_item() - Release Lock for Retry

**Purpose:** Release instance lock, optionally delay visibility for backoff, and optionally undo the attempt count.

**Parameters:**
- `lock_token` - The lock token from `fetch_orchestration_item`
- `delay` - Optional delay before messages become visible again
- `ignore_attempt` - If true, decrement the `attempt_count` (never below 0)

**⚠️ CRITICAL: Invalid Lock Token Handling**

The provider **MUST** return an error if the lock token is invalid (not found in `instance_locks` table). This is the same behavior as `ack_orchestration_item()`.

**Why errors on invalid tokens:**
- Invalid lock tokens indicate a programming error or state corruption
- Both `ack_orchestration_item()` and `abandon_orchestration_item()` return errors on invalid tokens
- Invalid tokens suggest the lock was already released or never existed, which indicates a bug
- Returning an error helps detect and debug race conditions or state corruption

**Pseudo-code:**
```
BEGIN TRANSACTION

// Step 1: Validate lock token
instance_id = SELECT instance_id FROM instance_locks
              WHERE lock_token = lock_token

IF instance_id IS NULL:
    // Invalid lock token - this is an error condition
    ROLLBACK
    RETURN Err(ProviderError::permanent("abandon_orchestration_item", "Invalid lock token"))

// Step 2: Remove instance lock
DELETE FROM instance_locks WHERE lock_token = lock_token

// Step 3: If ignore_attempt, decrement attempt count (never below 0)
IF ignore_attempt:
    UPDATE orchestrator_queue
    SET attempt_count = MAX(0, attempt_count - 1)
    WHERE instance_id = instance_id

// Step 4: Optionally delay message visibility
IF delay IS NOT NULL:
    visible_at = current_timestamp() + delay
    UPDATE orchestrator_queue
    SET visible_at = visible_at
    WHERE instance_id = instance_id
      AND visible_at <= current_timestamp()

COMMIT TRANSACTION
RETURN Ok(())
```

**Critical:**
- Invalid lock tokens MUST return an error (same as `ack_orchestration_item()`)
- This helps detect programming errors and state corruption
- If lock token is valid, proceed with unlocking messages and removing the instance lock
- Optional delay applies to messages for the instance (useful for backoff strategies)
- `ignore_attempt` is useful when abandoning due to transient infrastructure errors (not actual processing failures)

---

### 4. read() - Load Latest Execution History

**Pseudo-code:**
```
// Get latest execution ID
execution_id = SELECT COALESCE(MAX(execution_id), 1)
               FROM executions
               WHERE instance_id = instance

// Load events for that execution
rows = SELECT event_data FROM history
       WHERE instance_id = instance
         AND execution_id = execution_id
       ORDER BY event_id ASC

events = rows.map(|row| deserialize_event(row.event_data))
         .collect()

RETURN events  // Empty Vec if instance doesn't exist
```

**⚠️ CRITICAL: Missing Instance Handling**

If an instance doesn't exist, `read()` MUST return an empty `Vec<Event>`, not an error. This allows graceful handling of non-existent instances.

**Why this matters:**
- Enables runtime to check if instance exists without error handling
- Consistent behavior across all providers
- Allows creation of instances on first read attempt

**⚠️ CRITICAL: Graceful Handling of Corrupted Data**

If deserialization fails for queue items or history events, the provider MUST handle it gracefully:
- Corrupted queue items should be skipped (return None for fetch/dequeue)
- Corrupted history events should be skipped or logged (don't crash)
- Consider logging corruption for debugging

**Why this matters:**
- Prevents single corrupted item from crashing entire system
- Allows recovery from data corruption
- Maintains system availability

---

### 5. Worker Queue Operations

**enqueue_for_worker():**
```
work_json = serialize(item)
INSERT INTO worker_queue (work_item, lock_token, locked_until)
VALUES (work_json, NULL, NULL)
```

**fetch_work_item(lock_timeout: Duration, poll_timeout: Duration):**
```
// poll_timeout: Maximum time to wait for work. 
// Short-polling providers (like SQLite) ignore this and return immediately.
// Long-polling providers MAY block up to this duration.

BEGIN TRANSACTION

row = SELECT id, work_item FROM worker_queue
      WHERE lock_token IS NULL OR locked_until <= current_timestamp()
      ORDER BY id ASC
      LIMIT 1

IF row IS NULL:
    ROLLBACK
    RETURN Ok(None)  // Empty queue - not an error

lock_token = generate_uuid()
// lock_timeout is a Duration - convert to milliseconds for storage
now_ms = current_timestamp_ms()
locked_until = now_ms + lock_timeout.as_millis()

UPDATE worker_queue
SET lock_token = lock_token, locked_until = locked_until
WHERE id = row.id

COMMIT
item = deserialize_workitem(row.work_item)
RETURN Ok(Some((item, lock_token)))
```

**⚠️ CRITICAL: Error Handling for fetch_work_item**

`fetch_work_item()` MUST return `Result<Option<(WorkItem, String)>, ProviderError>`:
- `Ok(None)` - Queue is empty (normal case, not an error)
- `Ok(Some((item, token)))` - Successfully fetched and locked an item
- `Err(ProviderError)` - Storage operation failed (database error, connection lost, etc.)

**Why this matters:**
- Allows runtime to distinguish between empty queue (back off) and storage failures (retry)
- Enables proper error propagation and retry logic
- Prevents silent failures that could cause worker dispatcher to hang

**Error classification:**
- Database busy/locked → `ProviderError::retryable("fetch_work_item", ...)`
- Connection timeout → `ProviderError::retryable("fetch_work_item", ...)`
- Data corruption (invalid JSON) → `ProviderError::permanent("fetch_work_item", ...)`

**⚠️ CRITICAL: Worker Queue FIFO Ordering**

Worker items MUST be dequeued in FIFO (first-in-first-out) order based on insertion order. This ensures fair processing and maintains execution order.

**Why this matters:**
- Prevents starvation of older work items
- Maintains predictable processing order
- Essential for fairness in multi-worker systems

**⚠️ CRITICAL: Worker Peek-Lock Semantics**

When `fetch_work_item()` returns `Ok(Some((item, token)))`, it MUST:
- Lock the item (set lock_token and locked_until)
- Keep the item in the queue (don't delete it)
- Return item + lock_token
- Only delete the item when `ack_worker()` is called with the lock_token

**Why this matters:**
- Prevents item loss if worker crashes before ack
- Allows lock expiration to recover lost items
- Enables at-least-once processing semantics

**ack_worker():**
```
BEGIN TRANSACTION

// Step 1: Delete item from worker queue
DELETE FROM worker_queue WHERE lock_token = lock_token

// Step 2: Enqueue completion to orchestrator queue
work_json = serialize(completion)
target_instance = extract_instance(completion)

INSERT INTO orchestrator_queue (instance_id, work_item, visible_at, lock_token, locked_until)
VALUES (target_instance, work_json, NOW(), NULL, NULL)

COMMIT TRANSACTION
RETURN Ok(())  // Idempotent
```

**⚠️ CRITICAL: Worker Ack Atomicity**

`ack_worker()` MUST atomically delete the worker item AND enqueue the completion. Both operations must succeed or both must fail.

**Why this matters:**
- Prevents lost completions if enqueue fails
- Prevents duplicate work if delete fails
- Ensures exactly-once completion semantics

**⚠️ CRITICAL: Lost Lock Token Handling**

If a worker dequeues an item but crashes before acking (loses the lock token), the item MUST eventually become available again after lock expiration. This requires:
- Lock timeout on worker queue items
- Expired locks allow item to be dequeued again
- Items are redelivered when lock expires

**Why this matters:**
- Recovers from worker crashes
- Prevents permanent loss of work items
- Enables at-least-once processing semantics

### Method 4: renew_work_item_lock (Lock Renewal)

**Purpose:** Extend lock timeout for in-progress activities to support long-running work.

**Implementation:**
```
async fn renew_work_item_lock(token: &str, extend_for: Duration) -> Result<(), ProviderError> {
    // extend_for is a Duration - convert to milliseconds for storage
    now_ms = current_timestamp_ms()
    locked_until = now_ms + extend_for.as_millis()
    
    result = UPDATE worker_queue
             SET locked_until = locked_until
             WHERE lock_token = token
               AND locked_until > now_ms  // Only renew if still valid
    
    IF result.rows_affected == 0:
        // Token invalid, expired, or already acked
        RETURN Err(ProviderError::permanent("Lock token invalid or expired"))
    
    RETURN Ok(())
}
```

**Called By:**
- Worker dispatcher automatically for all in-flight activities
- Renewal interval based on timeout:
  - timeout ≥15s: renew at `(timeout - buffer_secs)` 
  - timeout <15s: renew at `0.5 * timeout`
- Example: 30s timeout, 5s buffer → renew every 25s

**Why this matters:**
- Enables activities that run longer than initial lock timeout
- Prevents lock expiration for legitimate long-running work
- Stops automatically when activity completes or worker crashes
- Graceful recovery: if worker dies, renewal stops and lock expires naturally

---

### Method 5: abandon_work_item (Release Work Item Lock)

**Purpose:** Release a work item lock without completing it, optionally with a delay for backoff.

**Parameters:**
- `token` - Lock token from `fetch_work_item`
- `delay` - Optional delay before work item becomes visible again
- `ignore_attempt` - If true, decrement the `attempt_count` (never below 0)

**Implementation:**
```
async fn abandon_work_item(
    token: &str,
    delay: Option<Duration>,
    ignore_attempt: bool,
) -> Result<(), ProviderError> {
    locked_until = IF delay IS NOT NULL:
                       current_timestamp() + delay
                   ELSE:
                       NULL

    IF ignore_attempt:
        result = UPDATE worker_queue
                 SET lock_token = NULL,
                     locked_until = locked_until,
                     attempt_count = MAX(0, attempt_count - 1)
                 WHERE lock_token = token
    ELSE:
        result = UPDATE worker_queue
                 SET lock_token = NULL, locked_until = locked_until
                 WHERE lock_token = token

    IF result.rows_affected == 0:
        RETURN Err(ProviderError::permanent("Invalid lock token or already acked"))

    RETURN Ok(())
}
```

**Called By:**
- Worker dispatcher when activity execution fails with a retryable error
- Worker dispatcher when `ack_work_item` fails (infrastructure error)

**ignore_attempt Use Case:**
- When abandoning due to transient infrastructure errors (e.g., database contention)
- The work was not actually processed, so attempt count shouldn't be penalized
- Prevents false poison message detection

**Why this matters:**
- Allows explicit release of work item lock for retry
- Enables backoff strategies via delay parameter
- `ignore_attempt` prevents false poison counts from infrastructure errors

---

## Schema Recommendations

### Minimum Required Tables

1. **history** - Append-only event log
   - PRIMARY KEY: (instance_id, execution_id, event_id) - **Enforces uniqueness, must error on duplicates**
   - Columns: event_data (JSON), event_type (for indexing), created_at
   - **Important:** The PRIMARY KEY constraint ensures duplicate events are rejected. Providers should return an error when duplicate events are detected (database constraint violation), not silently ignore them.

2. **orchestrator_queue** - Orchestration work items
   - PRIMARY KEY: id (auto-increment)
   - Columns: instance_id, work_item (JSON), visible_at, lock_token, locked_until
   - INDEX: (visible_at, lock_token) for fetch performance

3. **worker_queue** - Activity execution requests
   - PRIMARY KEY: id (auto-increment)
   - Columns: work_item (JSON), lock_token, locked_until

4. **instances** - Instance metadata
   - PRIMARY KEY: instance_id
   - Columns: orchestration_name, orchestration_version (NULLable), current_execution_id
   - **Note**: `orchestration_version` is NULLable (TEXT, not TEXT NOT NULL)
   - **Note**: Version starts as NULL, then set by runtime via `ack_orchestration_item` metadata
   - **Note**: If version is NULL and instance exists, use "unknown" (don't extract from history)

5. **executions** - Execution tracking
   - PRIMARY KEY: (instance_id, execution_id)
   - Columns: status, output, started_at, completed_at

6. **instance_locks** - ⚠️ CRITICAL: Instance-level locking table
   - PRIMARY KEY: instance_id
   - Columns: lock_token (unique UUID), locked_until (Unix timestamp milliseconds), locked_at (Unix timestamp milliseconds)
   - **Purpose**: Prevents concurrent dispatchers from processing the same instance
   - **Required**: Must be implemented for correct provider behavior
   - **Index**: Consider adding index on (locked_until) for efficient lock expiration checks

---


## Common Pitfalls

### ❌ DON'T: Inspect Event Contents

```rust
// WRONG - Provider inspecting events
for event in &history_delta {
    match &event.kind {
        EventKind::OrchestrationCompleted { output } => {
            // Provider understanding orchestration semantics
            self.update_status("Completed", output)?;
        }
        _ => {}
    }
}
```

```rust
// CORRECT - Use metadata from runtime
if let Some(status) = &metadata.status {
    self.update_status(status, &metadata.output)?;
}
```

### ❌ DON'T: Modify Event IDs

```rust
// WRONG - Renumbering events
let mut next_id = self.get_next_event_id()?;
for mut event in history_delta {
    event.set_event_id(next_id);  // DON'T DO THIS!
    next_id += 1;
}
```

```rust
// CORRECT - Store as-is
for event in &history_delta {
    assert!(event.event_id() > 0, "Runtime must set event_ids");
    self.insert_event(event)?;  // Use runtime-assigned IDs
}
```

### ❌ DON'T: Silently Ignore Duplicate Events

```rust
// WRONG - Silently ignoring duplicates
INSERT INTO history (...) VALUES (...)
ON CONFLICT DO NOTHING  // DON'T DO THIS!
```

```rust
// CORRECT - Error on duplicates with ProviderError
INSERT INTO history (...) VALUES (...)
// PRIMARY KEY constraint will cause error if duplicate exists
// Provider should return ProviderError when constraint violation occurs
match db.execute(query).await {
    Err(e) if e.is_constraint_violation() => {
        Err(ProviderError::permanent("ack_orchestration_item", "duplicate event_id detected"))
    }
    Err(e) => Err(sqlx_to_provider_error("ack_orchestration_item", e)),
    Ok(_) => Ok(()),
}
```
```

**Why:** Duplicate events indicate a serious problem (retry bug, runtime issue, or provider corruption). Silent ignoring hides the problem and can lead to data inconsistency. The runtime already performs duplicate detection based on history before calling `ack_orchestration_item()`, so duplicates reaching the provider represent an invalid state that must be surfaced as an error.

### ❌ DON'T: Break Atomicity

```rust
// WRONG - Non-atomic
self.append_history(history_delta).await?;  // Committed!
self.enqueue_workers(worker_items).await?;  // If this fails, history already saved!
self.release_lock(lock_token).await?;
```

```rust
// CORRECT - Single transaction
let tx = self.begin_transaction().await?;
self.append_history_in_tx(&tx, history_delta).await?;
self.enqueue_workers_in_tx(&tx, worker_items).await?;
self.release_lock_in_tx(&tx, lock_token).await?;
tx.commit().await?;  // All or nothing
```

---

## Testing Your Provider

After implementing your provider, use the **Provider Validation Test Suite** to validate correctness. The test suite provides granular test functions that you can run individually for easier debugging.

### Quick Start

1. **Create a `ProviderFactory`** that implements `duroxide::provider_validations::ProviderFactory`:

```rust
use duroxide::providers::Provider;
use duroxide::provider_validations::ProviderFactory;
use std::sync::Arc;
use std::time::Duration;

const TEST_LOCK_TIMEOUT: Duration = Duration::from_secs(1);

struct MyProviderFactory;

#[async_trait::async_trait]
impl ProviderFactory for MyProviderFactory {
    async fn create_provider(&self) -> Arc<dyn Provider> {
        // Create a fresh provider instance for each test
        Arc::new(MyProvider::new().await.unwrap())
    }

    fn lock_timeout(&self) -> Duration {
        // Return the lock timeout used in validation tests
        // Note: Actual lock timeout is now configured via RuntimeOptions
        TEST_LOCK_TIMEOUT
    }
}
```

2. **Write individual test functions** for each validation test:

```rust
use duroxide::provider_validations::{
    ProviderFactory,
    test_atomicity_failure_rollback,
    test_exclusive_instance_lock,
    test_worker_queue_fifo_ordering,
    // ... import other tests as needed
};

#[tokio::test]
async fn test_my_provider_atomicity_failure_rollback() {
    let factory = MyProviderFactory;
    test_atomicity_failure_rollback(&factory).await;
}

#[tokio::test]
async fn test_my_provider_exclusive_instance_lock() {
    let factory = MyProviderFactory;
    test_exclusive_instance_lock(&factory).await;
}
```

### Available Test Functions

The validation suite provides **58 individual test functions** organized into 9 categories:

**Atomicity Tests (4 tests):**
- `test_atomicity_failure_rollback` - Verify ack failure rolls back all operations
- `test_multi_operation_atomic_ack` - Verify complex ack succeeds atomically
- `test_lock_released_only_on_successful_ack` - Verify lock only released on success
- `test_concurrent_ack_prevention` - Verify only one ack succeeds with same token

**Error Handling Tests (5 tests):**
- `test_invalid_lock_token_on_ack` - Verify invalid lock tokens are rejected
- `test_duplicate_event_id_rejection` - Verify duplicate events are rejected
- `test_missing_instance_metadata` - Verify missing instances handled gracefully
- `test_corrupted_serialization_data` - Verify corrupted data handled gracefully
- `test_lock_expiration_during_ack` - Verify expired locks are rejected

**Instance Locking Tests (11 tests):**
- `test_exclusive_instance_lock` - Verify only one dispatcher can process an instance
- `test_lock_token_uniqueness` - Verify each fetch generates unique lock token
- `test_invalid_lock_token_rejection` - Verify invalid tokens rejected for ack/abandon
- `test_concurrent_instance_fetching` - Verify concurrent fetches don't duplicate instances
- `test_completions_arriving_during_lock_blocked` - Verify new messages blocked during lock
- `test_cross_instance_lock_isolation` - Verify locks don't block other instances
- `test_message_tagging_during_lock` - Verify only fetched messages deleted on ack
- `test_ack_only_affects_locked_messages` - Verify ack only affects locked messages
- `test_multi_threaded_lock_contention` - Verify locks prevent concurrent processing (multi-threaded)
- `test_multi_threaded_no_duplicate_processing` - Verify no duplicate processing (multi-threaded)
- `test_multi_threaded_lock_expiration_recovery` - Verify lock expiration recovery (multi-threaded)

**Lock Expiration Tests (9 tests):**
- `test_lock_expires_after_timeout` - Verify locks expire after timeout
- `test_abandon_releases_lock_immediately` - Verify abandon releases lock immediately
- `test_lock_renewal_on_ack` - Verify successful ack releases lock immediately
- `test_concurrent_lock_attempts_respect_expiration` - Verify concurrent attempts respect expiration
- `test_worker_lock_renewal_success` - Verify worker lock can be renewed with valid token
- `test_worker_lock_renewal_invalid_token` - Verify renewal fails with invalid token
- `test_worker_lock_renewal_after_expiration` - Verify renewal fails after lock expires
- `test_worker_lock_renewal_extends_timeout` - Verify renewal properly extends lock timeout
- `test_worker_lock_renewal_after_ack` - Verify renewal fails after item has been acked

**Multi-Execution Tests (5 tests):**
- `test_execution_isolation` - Verify each execution has separate history
- `test_latest_execution_detection` - Verify read() returns latest execution
- `test_execution_id_sequencing` - Verify execution IDs increment correctly
- `test_continue_as_new_creates_new_execution` - Verify ContinueAsNew creates new execution
- `test_execution_history_persistence` - Verify all executions' history persists independently

**Queue Semantics Tests (5 tests):**
- `test_worker_queue_fifo_ordering` - Verify worker items dequeued in FIFO order
- `test_worker_peek_lock_semantics` - Verify dequeue doesn't remove item until ack
- `test_worker_ack_atomicity` - Verify ack_worker atomically removes item and enqueues completion
- `test_timer_delayed_visibility` - Verify TimerFired items only dequeued when visible
- `test_lost_lock_token_handling` - Verify locked items become available after expiration

**Instance Creation Tests (4 tests):**
- `test_instance_creation_via_metadata` - Verify instances created via ack metadata, not on enqueue
- `test_no_instance_creation_on_enqueue` - Verify no instance created when enqueueing
- `test_null_version_handling` - Verify NULL version handled correctly
- `test_sub_orchestration_instance_creation` - Verify sub-orchestrations follow same pattern

**Management Tests (7 tests):**
- `test_list_instances` - Verify list_instances returns all instance IDs
- `test_list_instances_by_status` - Verify list_instances_by_status filters correctly
- `test_list_executions` - Verify list_executions returns all execution IDs
- `test_get_instance_info` - Verify get_instance_info returns metadata
- `test_get_execution_info` - Verify get_execution_info returns execution metadata
- `test_get_system_metrics` - Verify get_system_metrics returns accurate counts
- `test_get_queue_depths` - Verify get_queue_depths returns current queue sizes

**Poison Message Tests (12 tests):**
- `orchestration_attempt_count_starts_at_one` - Verify first fetch has attempt_count = 1
- `orchestration_attempt_count_increments_on_refetch` - Verify attempt count increments on abandon/refetch
- `worker_attempt_count_starts_at_one` - Verify worker items start with attempt_count = 1
- `worker_attempt_count_increments_on_lock_expiry` - Verify attempt count increments when lock expires
- `attempt_count_is_per_message` - Verify each message has independent attempt count
- `abandon_work_item_ignore_attempt_decrements` - Verify ignore_attempt=true decrements count
- `abandon_orchestration_item_ignore_attempt_decrements` - Verify ignore_attempt=true decrements count
- `ignore_attempt_never_goes_negative` - Verify attempt count never goes below 0
- `max_attempt_count_across_message_batch` - Verify MAX attempt_count returned for batched messages
- `test_abandon_work_item_releases_lock` - Verify abandon_work_item releases lock immediately
- `test_abandon_work_item_with_delay` - Verify abandon_work_item with delay defers refetch

### Benefits of Granular Tests

**Easier debugging:** When a test fails, you know exactly which behavior is broken. You can run a single test function to quickly isolate the issue.

**Better CI/CD integration:** Run tests in parallel, skip known failures, or focus on specific test categories during development.

**Incremental development:** Implement your provider incrementally, testing one behavior at a time.

### Example: Complete Test File

See `tests/sqlite_provider_validations.rs` for a complete example that runs all 62 validation tests individually.

---

## Performance Considerations

### Indexes (Critical for Performance)

```sql
-- Orchestrator queue (hot path)
CREATE INDEX idx_orch_visible ON orchestrator_queue(visible_at, lock_token);
CREATE INDEX idx_orch_instance ON orchestrator_queue(instance_id);

-- Worker queue
CREATE INDEX idx_worker_lock ON worker_queue(lock_token);

-- History (for read operations)
CREATE INDEX idx_history_lookup ON history(instance_id, execution_id, event_id);
```

### Connection Pooling

- Use connection pools for concurrent dispatcher access
- Recommended pool size: 5-10 connections
- SQLite example: `SqlitePoolOptions::new().max_connections(10)`

### Lock Timeout

- Recommended: 30 seconds
- Too short: False retries under load
- Too long: Slow recovery from crashes

---

## Validation Checklist

Before considering your provider production-ready:

- [ ] All 12 provider tests from sqlite_provider_test.rs pass
- [ ] fetch_orchestration_item returns None when queue empty
- [ ] ack_orchestration_item is fully atomic (single transaction)
- [ ] Lock expiration works (messages redelivered after timeout)
- [ ] Multi-execution support (ContinueAsNew creates execution 2, 3, ...)
- [ ] Execution metadata stored correctly (status, output)
- [ ] History ordering preserved (events returned in event_id order)
- [ ] Concurrent access safe (run with RUST_TEST_THREADS=10)
- [ ] No instance creation on enqueue (instances created via ack metadata)
- [ ] NULL version handling works correctly

---

## Example Providers to Implement

### Easy
- **In-Memory Provider**: Use HashMap + RwLock
- **File-based Provider**: JSON files + file locks

### Medium
- **PostgreSQL Provider**: Similar to SQLite, better concurrency
- **MySQL Provider**: Similar to PostgreSQL

### Advanced
- **Redis Provider**: Lua scripts for atomicity, sorted sets for history
- **DynamoDB Provider**: Conditional writes, GSI for queues
- **Cosmos DB Provider**: Change feed for queues, partition by instance_id

---

## Implementing ProviderAdmin (Optional)

The `ProviderAdmin` trait provides rich management and observability features for your provider. While optional, implementing it enables powerful introspection and monitoring capabilities.

### When to Implement

**Implement ProviderAdmin if:**
- Your provider is for production use
- You need instance/execution discovery
- You want system metrics and monitoring
- You need debugging/troubleshooting capabilities

**Skip if:**
- Building a minimal/test provider
- Storage backend doesn't support efficient queries
- Only need core orchestration execution

### The Trait

```rust
#[async_trait::async_trait]
pub trait ProviderAdmin: Send + Sync {
    // Discovery
    async fn list_instances(&self) -> Result<Vec<String>, ProviderError>;
    async fn list_instances_by_status(&self, status: &str) -> Result<Vec<String>, ProviderError>;
    async fn list_executions(&self, instance: &str) -> Result<Vec<u64>, ProviderError>;
    
    // History access
    async fn read_history_with_execution_id(&self, instance: &str, execution_id: u64) -> Result<Vec<Event>, ProviderError>;
    async fn read_history(&self, instance: &str) -> Result<Vec<Event>, ProviderError>;
    async fn latest_execution_id(&self, instance: &str) -> Result<u64, ProviderError>;
    
    // Metadata
    async fn get_instance_info(&self, instance: &str) -> Result<InstanceInfo, ProviderError>;
    async fn get_execution_info(&self, instance: &str, execution_id: u64) -> Result<ExecutionInfo, ProviderError>;
    
    // System metrics
    async fn get_system_metrics(&self) -> Result<SystemMetrics, ProviderError>;
    async fn get_queue_depths(&self) -> Result<QueueDepths, ProviderError>;
}
```

### Implementing the Trait

```rust
use duroxide::providers::{Provider, ProviderAdmin, InstanceInfo, ExecutionInfo, SystemMetrics, QueueDepths};

#[async_trait::async_trait]
impl Provider for MyProvider {
    // ... core Provider methods ...
    
    fn as_management_capability(&self) -> Option<&dyn ProviderAdmin> {
        Some(self)  // Enable management features
    }
}

#[async_trait::async_trait]
impl ProviderAdmin for MyProvider {
    async fn list_instances(&self) -> Result<Vec<String>, ProviderError> {
        // Return all instance IDs
        let rows = sqlx::query!("SELECT instance_id FROM instances ORDER BY created_at")
            .fetch_all(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("list_instances", e))?;
        
        Ok(rows.into_iter().map(|r| r.instance_id).collect())
    }
    
    async fn list_instances_by_status(&self, status: &str) -> Result<Vec<String>, ProviderError> {
        // Return instances with specific status (Running, Completed, Failed)
        let rows = sqlx::query!(
            "SELECT DISTINCT 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",
            status
        )
        .fetch_all(&self.pool)
        .await
        .map_err(|e| sqlx_to_provider_error("list_instances_by_status", e))?;
        
        Ok(rows.into_iter().map(|r| r.instance_id).collect())
    }
    
    async fn list_executions(&self, instance: &str) -> Result<Vec<u64>, ProviderError> {
        // Return all execution IDs for an instance
        let rows = sqlx::query!(
            "SELECT execution_id FROM executions 
             WHERE instance_id = ? 
             ORDER BY execution_id",
            instance
        )
        .fetch_all(&self.pool)
        .await
            .map_err(|e| sqlx_to_provider_error("list_executions", e))?;
        
        Ok(rows.into_iter().map(|r| r.execution_id as u64).collect())
    }
    
    async fn read_history_with_execution_id(&self, instance: &str, execution_id: u64) -> Result<Vec<Event>, ProviderError> {
        // Return history for specific execution (not just latest)
        let rows = sqlx::query!(
            "SELECT event_data FROM history 
             WHERE instance_id = ? AND execution_id = ? 
             ORDER BY event_id",
            instance,
            execution_id as i64
        )
        .fetch_all(&self.pool)
        .await
            .map_err(|e| sqlx_to_provider_error("read_history_with_execution_id", e))?;
        
        rows.into_iter()
            .filter_map(|row| serde_json::from_str(&row.event_data).ok())
            .collect::<Result<Vec<_>, _>>()
            .map_err(|e| ProviderError::permanent("read_history_with_execution_id", &format!("Failed to deserialize events: {}", e)))
    }
    
    async fn read_history(&self, instance: &str) -> Result<Vec<Event>, ProviderError> {
        // Convenience method: get history for latest execution
        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> {
        // Return highest execution ID
        let row = sqlx::query!(
            "SELECT current_execution_id FROM instances WHERE instance_id = ?",
            instance
        )
        .fetch_optional(&self.pool)
        .await
            .map_err(|e| sqlx_to_provider_error("latest_execution_id", e))?;
        
        row.map(|r| r.current_execution_id as u64)
            .ok_or_else(|| ProviderError::permanent("latest_execution_id", "Instance not found"))
    }
    
    async fn get_instance_info(&self, instance: &str) -> Result<InstanceInfo, ProviderError> {
        // Return instance metadata
        let row = sqlx::query!(
            "SELECT i.orchestration_name, i.orchestration_version, 
                    i.current_execution_id, i.created_at,
                    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 = ?",
            instance
        )
        .fetch_optional(&self.pool)
        .await
            .map_err(|e| sqlx_to_provider_error("get_instance_info", e))?
            .ok_or_else(|| ProviderError::permanent("get_instance_info", "Instance not found"))?;
        
        Ok(InstanceInfo {
            instance_id: instance.to_string(),
            orchestration_name: row.orchestration_name,
            orchestration_version: row.orchestration_version,
            current_execution_id: row.current_execution_id as u64,
            status: row.status.unwrap_or_else(|| "Running".to_string()),
            output: row.output,
            created_at: row.created_at,
        })
    }
    
    async fn get_execution_info(&self, instance: &str, execution_id: u64) -> Result<ExecutionInfo, ProviderError> {
        // Return execution-specific metadata
        let row = sqlx::query!(
            "SELECT status, output, started_at, completed_at 
             FROM executions 
             WHERE instance_id = ? AND execution_id = ?",
            instance,
            execution_id as i64
        )
        .fetch_optional(&self.pool)
        .await
            .map_err(|e| sqlx_to_provider_error("get_execution_info", e))?
            .ok_or_else(|| ProviderError::permanent("get_execution_info", "Execution not found"))?;
        
        Ok(ExecutionInfo {
            instance_id: instance.to_string(),
            execution_id,
            status: row.status,
            output: row.output,
            started_at: row.started_at,
            completed_at: row.completed_at,
        })
    }
    
    async fn get_system_metrics(&self) -> Result<SystemMetrics, ProviderError> {
        // Return system-wide statistics
        let total_instances = sqlx::query!("SELECT COUNT(*) as count FROM instances")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        let total_executions = sqlx::query!("SELECT COUNT(*) as count FROM executions")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        let running = sqlx::query!("SELECT COUNT(*) as count FROM executions WHERE status = 'Running'")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        let completed = sqlx::query!("SELECT COUNT(*) as count FROM executions WHERE status = 'Completed'")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        let failed = sqlx::query!("SELECT COUNT(*) as count FROM executions WHERE status = 'Failed'")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        let total_events = sqlx::query!("SELECT COUNT(*) as count FROM history")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_system_metrics", e))?
            .count as usize;
        
        Ok(SystemMetrics {
            total_instances,
            total_executions,
            running_instances: running,
            completed_instances: completed,
            failed_instances: failed,
            total_events,
        })
    }
    
    async fn get_queue_depths(&self) -> Result<QueueDepths, ProviderError> {
        // Return current queue depths
        let orch = sqlx::query!("SELECT COUNT(*) as count FROM orchestrator_queue")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_queue_depths", e))?
            .count as usize;
        
        let worker = sqlx::query!("SELECT COUNT(*) as count FROM worker_queue")
            .fetch_one(&self.pool)
            .await
            .map_err(|e| sqlx_to_provider_error("get_queue_depths", e))?
            .count as usize;
        
        Ok(QueueDepths {
            orchestrator_queue: orch,
            worker_queue: worker,
            timer_queue: 0,  // Timers go to orchestrator queue with delayed visibility
        })
    }
}
```

### Client Auto-Discovery

When you implement `ProviderAdmin`, the Client automatically discovers it:

```rust
use duroxide::Client;

let client = Client::new(provider);

// Check if management features are available
if client.has_management_capability() {
    // Use management APIs
    let instances = client.list_all_instances().await?;
    let metrics = client.get_system_metrics().await?;
    let info = client.get_instance_info("my-instance").await?;
} else {
    println!("Provider doesn't support management features");
}
```

### Return Types

```rust
pub struct InstanceInfo {
    pub instance_id: String,
    pub orchestration_name: String,
    pub orchestration_version: String,
    pub current_execution_id: u64,
    pub status: String,              // "Running", "Completed", "Failed"
    pub output: Option<String>,      // Final output/error
    pub created_at: String,          // Timestamp
}

pub struct ExecutionInfo {
    pub instance_id: String,
    pub execution_id: u64,
    pub status: String,
    pub output: Option<String>,
    pub started_at: String,
    pub completed_at: Option<String>,
}

pub struct SystemMetrics {
    pub total_instances: usize,
    pub total_executions: usize,
    pub running_instances: usize,
    pub completed_instances: usize,
    pub failed_instances: usize,
    pub total_events: usize,
}

pub struct QueueDepths {
    pub orchestrator_queue: usize,
    pub worker_queue: usize,
    pub timer_queue: usize,  // Usually 0 (timers in orch queue)
}
```

### Performance Tips

**Efficient Status Queries:**
```sql
-- Use JOIN to get current execution status
SELECT i.instance_id, e.status
FROM instances i
JOIN executions e ON i.instance_id = e.instance_id 
  AND i.current_execution_id = e.execution_id
WHERE e.status = 'Running'
```

**Cached Metrics:**
```rust
// For high-traffic systems, consider caching metrics
struct MyProvider {
    pool: SqlitePool,
    metrics_cache: Arc<RwLock<Option<(SystemMetrics, Instant)>>>,
}

async fn get_system_metrics(&self) -> Result<SystemMetrics, String> {
    // Check cache (refresh every 30 seconds)
    if let Some((metrics, cached_at)) = &*self.metrics_cache.read().await {
        if cached_at.elapsed() < Duration::from_secs(30) {
            return Ok(metrics.clone());
        }
    }
    
    // Compute fresh metrics
    let metrics = self.compute_metrics().await?;
    *self.metrics_cache.write().await = Some((metrics.clone(), Instant::now()));
    Ok(metrics)
}
```

### Default Implementations

If you don't implement `ProviderAdmin`, clients will get:
- `list_instances()` → Empty list
- `get_system_metrics()` → Default/zero metrics
- `has_management_capability()` → `false`

Core orchestration execution still works normally.

---

## Testing Your Provider

After implementing your provider, follow the **[Provider Testing Guide](provider-testing-guide.md)** to:
- Run **validation tests** (58 correctness tests) to verify behavior
- Run **stress tests** to measure performance under load
- Compare metrics with built-in providers
- Integrate tests into your CI/CD pipeline

### Quick Testing Setup

Enable the `provider-test` feature to access both validation and stress test infrastructure:

```toml
[dev-dependencies]
duroxide = { version = "0.1", features = ["provider-test"] }
```

**Validation Tests (Correctness):**
```rust
use duroxide::provider_validations::{ProviderFactory, test_atomicity_failure_rollback};

#[tokio::test]
async fn test_my_provider_atomicity() {
    test_atomicity_failure_rollback(&MyProviderFactory).await;
}
```

**Stress Tests (Performance):**
```rust
use duroxide::provider_stress_tests::parallel_orchestrations::{
    ProviderStressFactory, run_parallel_orchestrations_test
};

struct MyProviderStressFactory;

#[async_trait::async_trait]
impl ProviderStressFactory for MyProviderStressFactory {
    async fn create_provider(&self) -> Arc<dyn Provider> {
        Arc::new(MyProvider::new().await.unwrap())
    }
}

#[tokio::test]
async fn stress_test_my_provider() {
    let factory = MyProviderStressFactory;
    let result = run_parallel_orchestrations_test(&factory)
        .await
        .expect("Stress test failed");
    assert!(result.success_rate() > 99.0);
}
```

See the [Provider Testing Guide](provider-testing-guide.md) for comprehensive documentation.

---

## Getting Help

- **Reference Implementation**: `src/providers/sqlite.rs`
- **Interface Definition**: `src/providers/mod.rs` (this file, with full docs)
- **Validation Tests**: Enable `provider-test` feature, see `tests/sqlite_provider_validations.rs`
- **Stress Tests**: Enable `provider-test` feature, see `tests/provider_stress_test_validation.rs`
- **Testing Guide**: `docs/provider-testing-guide.md` (comprehensive testing documentation)
- **Schema**: `migrations/20240101000000_initial_schema.sql`
- **Migration Guide**: `docs/provider-migration-guide-instance-creation.md` (for updating existing providers)

---

**With these annotations, an LLM can implement a fully functional Provider!** 🎉

