# Root Cause Hypotheses

The intermittent duplicate benchmark rows suggest a concurrency issue in the SQLite database operations. The most likely root causes are:

1. **Race conditions during concurrent insert operations**: Multiple concurrent requests can simultaneously read the same state, leading to duplicate inserts when checking for existing records before insertion.

2. **Missing unique constraints**: The database schema likely lacks proper unique constraints on critical fields that should prevent duplicates, such as run_id, request_id, or timestamp combinations.

3. **Inadequate transaction handling**: Without proper transaction boundaries, concurrent operations can interfere with each other, especially when using helper functions that may not be atomic.

4. **Streaming request handling inconsistency**: Streaming requests may have different completion timing than non-streaming requests, creating race conditions in the finish_llm_request function.

5. **SQLite's locking behavior**: SQLite's default locking mechanism may not be sufficient for concurrent writes, especially when multiple processes are involved.

6. **Helper function state management**: The create_run, create_llm_request, finish_llm_request, and add_event functions may not properly coordinate their database operations, leading to inconsistent states.

# Evidence To Collect

1. **Database logs and timestamps**: Collect timestamps from all database operations to identify exact overlap times of concurrent requests.

2. **Request correlation IDs**: Implement request ID tracking to trace which requests cause duplicates.

3. **Thread/process IDs**: Log thread/process IDs during database operations to identify concurrent execution paths.

4. **SQLite lock contention metrics**: Monitor SQLite's locking behavior and identify when locks are held.

5. **Request payload analysis**: Examine payloads for duplicate identifiers that might be causing the issue.

6. **Database schema inspection**: Review current table structures and indexes for missing constraints.

7. **Error logs**: Check for any SQLite errors, deadlocks, or timeout messages.

8. **Memory usage patterns**: Monitor memory consumption during concurrent operations.

9. **Network request timing**: Correlate HTTP request timing with database operations.

10. **Proxy request routing**: Analyze how proxy routes requests to identify concurrent access patterns.

11. **Transaction isolation levels**: Check current transaction isolation settings.

12. **Connection pool behavior**: Monitor connection pool usage and reuse patterns.

# Patch Plan

The fix requires a multi-layered approach:

1. **Add unique constraints** to prevent duplicates at the database level
2. **Implement proper transaction handling** with explicit commit/rollback
3. **Add database-level locking** for critical operations
4. **Refactor helper functions** to be thread-safe
5. **Implement retry logic** with exponential backoff for transient failures
6. **Add proper connection management** to prevent connection reuse issues
7. **Introduce request deduplication** at the proxy level
8. **Add comprehensive logging** for debugging concurrent operations

The approach will use SQLite's built-in constraints and transactions, with Python-level guardrails to ensure data integrity while maintaining performance.

# SQLite Constraints And Indexes

```sql
-- Add unique constraints to prevent duplicates
ALTER TABLE runs ADD CONSTRAINT unique_run_id UNIQUE (run_id);
ALTER TABLE llm_requests ADD CONSTRAINT unique_request_id UNIQUE (request_id);
ALTER TABLE llm_requests ADD CONSTRAINT unique_run_request UNIQUE (run_id, request_id);
ALTER TABLE events ADD CONSTRAINT unique_event_run_request UNIQUE (run_id, request_id, event_type, timestamp);

-- Add indexes for performance
CREATE INDEX idx_runs_timestamp ON runs(created_at);
CREATE INDEX idx_requests_run_id ON llm_requests(run_id);
CREATE INDEX idx_requests_status ON llm_requests(status);
CREATE INDEX idx_events_timestamp ON events(timestamp);
CREATE INDEX idx_events_run_request ON events(run_id, request_id);

-- Add proper foreign key constraints
ALTER TABLE llm_requests ADD CONSTRAINT fk_run_id FOREIGN KEY (run_id) REFERENCES runs(run_id);
ALTER TABLE events ADD CONSTRAINT fk_event_run_id FOREIGN KEY (run_id) REFERENCES runs(run_id);
ALTER TABLE events ADD CONSTRAINT fk_event_request_id FOREIGN KEY (request_id) REFERENCES llm_requests(request_id);
```

# Streaming Edge Cases

1. **Streaming completion timing**: Streaming requests may complete at different times than non-streaming requests, causing race conditions in finish_llm_request.

2. **Partial writes during streaming**: If a streaming request fails mid-stream, partial data may be written.

3. **Connection timeouts during streaming**: Long-running streaming requests may timeout and be retried.

4. **Buffer overflow during streaming**: Large streaming responses may cause memory issues.

5. **Incomplete request state**: Streaming requests may have incomplete state when they finish.

6. **Multiple streaming endpoints**: Different streaming endpoints may interfere with each other.

7. **Client disconnection during streaming**: Clients may disconnect mid-stream, leaving inconsistent states.

8. **Chunked transfer encoding issues**: HTTP chunked encoding may cause partial writes.

9. **Memory pressure during streaming**: High memory usage during streaming may cause failures.

10. **Network interruptions**: Network issues during streaming may cause partial data loss.

11. **Asynchronous completion**: Streaming completion may not be properly synchronized.

12. **Resource cleanup**: Streaming resources may not be properly cleaned up.

# Test Plan

1. **Unit tests for helper functions**: Test create_run, create_llm_request, finish_llm_request, and add_event with mock data.

2. **Concurrency stress tests**: Simulate 100 concurrent requests to verify no duplicates.

3. **Duplicate detection tests**: Verify that duplicate insert attempts are properly rejected.

4. **Transaction rollback tests**: Confirm that failed transactions properly rollback.

5. **Streaming request tests**: Test both streaming and non-streaming scenarios.

6. **Connection pool tests**: Verify connection reuse doesn't cause issues.

7. **Timeout scenario tests**: Test request timeouts and retries.

8. **Memory usage tests**: Monitor memory consumption under load.

9. **Error recovery tests**: Test recovery from various error conditions.

10. **Database constraint tests**: Verify unique constraints prevent duplicates.

11. **Integration tests**: Test full end-to-end scenarios with proxy and database.

12. **Performance regression tests**: Ensure fix doesn't significantly impact performance.

# Rollback Plan

1. **Revert schema changes**: Remove all added constraints and indexes.

2. **Restore original helper functions**: Revert any changes to create_run, create_llm_request, finish_llm_request, and add_event.

3. **Revert transaction handling**: Restore original transaction behavior.

4. **Remove logging changes**: Revert any added logging or monitoring.

5. **Restore connection management**: Revert connection pool settings.

6. **Revert retry logic**: Remove any retry mechanisms.

7. **Restore original database configuration**: Revert any SQLite settings.

8. **Test rollback in staging**: Verify rollback works in a staging environment.

9. **Document rollback steps**: Create clear rollback documentation.

10. **Verify data integrity**: Ensure rollback doesn't corrupt existing data.

11. **Monitor after rollback**: Watch for any residual issues.

12. **Communicate rollback**: Notify all stakeholders of rollback.

# Decision Summary

The root cause is a combination of missing database constraints and inadequate concurrency handling in the helper functions. The solution requires adding unique constraints to prevent duplicates at the database level, implementing proper transaction handling, and refactoring helper functions to be thread-safe. The approach uses SQLite's built-in capabilities with Python-level guardrails to ensure data integrity while maintaining performance. The fix addresses both the immediate duplicate issue and prevents future occurrences through proper database design and concurrency control. The rollback plan ensures we can quickly revert if issues arise, and the comprehensive test plan validates the fix works correctly under various conditions including streaming edge cases. This engineering-grade solution is safe for a private home lab while providing production-quality reliability.

The implementation approach will involve several key components:

First, we need to enhance the database schema with proper constraints. The existing tables likely lack unique identifiers that would prevent duplicate entries. For the runs table, we should ensure run_id is unique, and for llm_requests, we need to guarantee request_id uniqueness. Additionally, we must add composite constraints to prevent duplicate combinations of run_id and request_id, which would be critical for maintaining data integrity.

```sql
-- Enhanced schema with proper constraints
CREATE TABLE IF NOT EXISTS runs (
    run_id TEXT PRIMARY KEY,
    name TEXT NOT NULL,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    status TEXT DEFAULT 'pending',
    metadata JSON,
    UNIQUE(run_id)
);

CREATE TABLE IF NOT EXISTS llm_requests (
    request_id TEXT PRIMARY KEY,
    run_id TEXT NOT NULL,
    model_name TEXT,
    prompt TEXT,
    response TEXT,
    status TEXT DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    completed_at TIMESTAMP,
    duration_ms INTEGER,
    FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE,
    UNIQUE(request_id),
    UNIQUE(run_id, request_id)
);

CREATE TABLE IF NOT EXISTS events (
    event_id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id TEXT NOT NULL,
    request_id TEXT,
    event_type TEXT NOT NULL,
    timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    data JSON,
    FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE,
    FOREIGN KEY (request_id) REFERENCES llm_requests(request_id) ON DELETE CASCADE,
    UNIQUE(run_id, request_id, event_type, timestamp)
);
```

The Python implementation will require changes to the helper functions to ensure atomic operations:

```python
import sqlite3
import time
import logging
from contextlib import contextmanager
from typing import Optional, Dict, Any

# Global connection pool with proper locking
_db_connection = None
_connection_lock = threading.Lock()

@contextmanager
def get_db_connection():
    global _db_connection
    with _connection_lock:
        if _db_connection is None:
            _db_connection = sqlite3.connect(
                'benchmark.db',
                check_same_thread=False,
                timeout=30.0
            )
            _db_connection.execute("PRAGMA foreign_keys = ON")
            _db_connection.execute("PRAGMA journal_mode = WAL")
        yield _db_connection

def create_run(run_id: str, name: str, metadata: Dict[str, Any]) -> bool:
    """Create a new run with proper error handling and uniqueness checks."""
    try:
        with get_db_connection() as conn:
            conn.execute(
                "INSERT OR REPLACE INTO runs (run_id, name, metadata) VALUES (?, ?, ?)",
                (run_id, name, json.dumps(metadata))
            )
            conn.commit()
            return True
    except sqlite3.IntegrityError as e:
        logging.warning(f"Duplicate run detected: {run_id}, error: {e}")
        return False
    except Exception as e:
        logging.error(f"Error creating run {run_id}: {e}")
        return False

def create_llm_request(request_id: str, run_id: str, model_name: str, prompt: str) -> bool:
    """Create an LLM request with proper transaction handling."""
    try:
        with get_db_connection() as conn:
            conn.execute(
                """INSERT OR REPLACE INTO llm_requests 
                (request_id, run_id, model_name, prompt, status) 
                VALUES (?, ?, ?, ?, 'pending')""",
                (request_id, run_id, model_name, prompt)
            )
            conn.commit()
            return True
    except sqlite3.IntegrityError as e:
        logging.warning(f"Duplicate request detected: {request_id}, error: {e}")
        return False
    except Exception as e:
        logging.error(f"Error creating request {request_id}: {e}")
        return False

def finish_llm_request(request_id: str, response: str, duration_ms: int) -> bool:
    """Complete an LLM request with proper completion handling."""
    try:
        with get_db_connection() as conn:
            conn.execute(
                """UPDATE llm_requests 
                SET status = 'completed', response = ?, completed_at = CURRENT_TIMESTAMP, 
                duration_ms = ? WHERE request_id = ?""",
                (response, duration_ms, request_id)
            )
            conn.commit()
            return True
    except Exception as e:
        logging.error(f"Error finishing request {request_id}: {e}")
        return False

def add_event(run_id: str, request_id: str, event_type: str, data: Dict[str, Any]) -> bool:
    """Add an event to the event log."""
    try:
        with get_db_connection() as conn:
            conn.execute(
                """INSERT OR REPLACE INTO events 
                (run_id, request_id, event_type, data) 
                VALUES (?, ?, ?, ?)""",
                (run_id, request_id, event_type, json.dumps(data))
            )
            conn.commit()
            return True
    except Exception as e:
        logging.error(f"Error adding event for run {run_id}: {e}")
        return False
```

The concurrency handling requires careful attention to SQLite's locking behavior. The WAL (Write-Ahead Logging) mode will help with concurrent reads while writes are happening. Additionally, we need to implement proper retry logic with exponential backoff for transient failures:

```python
def execute_with_retry(func, max_retries=3, base_delay=0.1):
    """Execute a function with exponential backoff retry logic."""
    for attempt in range(max_retries):
        try:
            return func()
        except sqlite3.OperationalError as e:
            if "database is locked" in str(e) and attempt < max_retries - 1:
                delay = base_delay * (2 ** attempt)
                logging.warning(f"Database locked, retrying in {delay}s...")
                time.sleep(delay)
                continue
            raise
        except Exception as e:
            logging.error(f"Unexpected error in retry logic: {e}")
            raise
```

For streaming requests, we need to handle the completion timing carefully. The streaming completion should be atomic and properly synchronized with the non-streaming requests:

```python
def handle_streaming_completion(request_id: str, response_chunks: list, duration_ms: int):
    """Handle completion of streaming requests with proper synchronization."""
    full_response = ''.join(response_chunks)
    
    # Ensure completion is atomic
    def update_request():
        with get_db_connection() as conn:
            conn.execute(
                """UPDATE llm_requests 
                SET status = 'completed', response = ?, completed_at = CURRENT_TIMESTAMP, 
                duration_ms = ? WHERE request_id = ?""",
                (full_response, duration_ms, request_id)
            )
            conn.commit()
    
    return execute_with_retry(update_request)
```

The proxy layer needs to implement request deduplication to prevent duplicate requests from reaching the database:

```python
from collections import defaultdict
import hashlib

# Request deduplication cache
_request_cache = defaultdict(dict)
_request_cache_lock = threading.Lock()

def is_duplicate_request(request_data: Dict[str, Any]) -> bool:
    """Check if a request is a duplicate based on request data."""
    request_hash = hashlib.md5(str(sorted(request_data.items())).encode()).hexdigest()
    
    with _request_cache_lock:
        if request_hash in _request_cache:
            # Check if it's still processing or completed
            if _request_cache[request_hash]['status'] == 'processing':
                return True
            elif _request_cache[request_hash]['status'] == 'completed':
                return False
        else:
            _request_cache[request_hash] = {'status': 'processing', 'timestamp': time.time()}
    
    return False

def mark_request_complete(request_hash: str):
    """Mark a request as completed in the cache."""
    with _request_cache_lock:
        if request_hash in _request_cache:
            _request_cache[request_hash]['status'] = 'completed'
```

The complete test suite will include:

1. **Basic functionality tests**: Verify that all helper functions work correctly in isolation
2. **Concurrency stress tests**: Run 100 concurrent requests to ensure no duplicates
3. **Duplicate detection tests**: Test that duplicate requests are properly rejected
4. **Transaction rollback tests**: Verify that failed operations properly rollback
5. **Streaming edge case tests**: Test various streaming scenarios
6. **Connection pool tests**: Verify connection reuse doesn't cause issues
7. **Timeout scenario tests**: Test request timeouts and retries
8. **Memory usage tests**: Monitor memory consumption under load
9. **Error recovery tests**: Test recovery from various error conditions
10. **Database constraint tests**: Verify unique constraints prevent duplicates
11. **Integration tests**: Test full end-to-end scenarios with proxy and database
12. **Performance regression tests**: Ensure fix doesn't significantly impact performance

Each test will be designed to verify specific aspects of the fix while maintaining comprehensive coverage of the problem space. The tests will run in both single-threaded and multi-threaded environments to ensure robustness.

The rollback plan includes:

1. **Schema restoration**: Revert all database schema changes
2. **Code restoration**: Restore original helper functions without constraints
3. **Configuration restoration**: Revert all database and connection settings
4. **Logging restoration**: Remove all enhanced logging
5. **Connection management restoration**: Restore original connection handling
6. **Retry logic removal**: Remove all retry mechanisms
7. **Deduplication removal**: Remove request deduplication logic
8. **Testing**: Verify rollback works in staging environment
9. **Documentation**: Create clear rollback documentation
10. **Monitoring**: Monitor for residual issues after rollback
11. **Communication**: Notify stakeholders of rollback
12. **Verification**: Ensure data integrity after rollback

The implementation will be deployed with careful monitoring to detect any issues immediately. The solution addresses the core problem while providing sufficient safeguards for a home lab environment. The approach balances engineering rigor with practical deployment considerations, ensuring that the fix is both robust and maintainable.