# Root Cause Hypotheses

Based on the symptoms (intermittent duplicate rows) and the architecture (FastAPI, SQLite, concurrent clients), there are three primary hypotheses for the root cause:

### 1. The "Check-then-Act" Race Condition (Primary Suspect)
The `create_run` helper function likely follows a pattern of:
1. Query the database to see if a run with a specific identifier already exists.
2. If it does not exist, execute an `INSERT` statement.

In a concurrent environment (FastAPI's `async` loop handling multiple requests), two requests from the Agent client or OpenWebUI can enter Step 1 simultaneously. Both see that the record does not exist, and both proceed to Step 2. Because SQLite's default behavior (without specific constraints) allows multiple inserts, two rows are created.

### 2. Client-Side Retries without Idempotency Keys
The proxy or the client (OpenWebUI/Agent) may be experiencing network timeouts or 500 errors. If the server successfully creates a run but the response fails to reach the client, the client retries. Without a unique "Idempotency Key" passed from the client, the server treats the retry as a brand-new request, creating a second row for the same logical operation.

### 3. SQLite Locking and "Database is Locked" Handling
SQLite handles concurrent writes by locking the entire database file. If the `busy_timeout` is not configured or is too low, a write attempt might fail. If the application code catches this exception and attempts a "retry" by calling `create_run` again, it may inadvertently create a new record if the previous attempt partially succeeded or if the logic doesn't distinguish between a "failure to write" and a "failure to find."

---

# Evidence To Collect

To confirm these hypotheses, the following telemetry and data points must be gathered:

1.  **Database Audit:**
    *   Extract all duplicate rows. Compare `created_at` timestamps. If they are within milliseconds, it confirms a race condition.
    *   Check for identical metadata (e.g., same prompt, same model parameters) in the duplicate rows.
2.  **Application Logs:**
    *   Correlate `Request-ID` headers from the proxy with the `run_id` in the database.
    *   Look for "Database is locked" errors in the logs.
    *   Identify if the same `client_request_id` (if any exists) is associated with multiple `run_id`s.
3.  **Network Trace:**
    *   Capture traffic between the Proxy and the FastAPI service. Look for repeated POST requests with identical bodies but different timestamps.
4.  **SQLite Configuration:**
    *   Verify the current `journal_mode` (DELETE vs. WAL).
    *   Verify the `busy_timeout` setting.

---

# Patch Plan

The fix requires a multi-layered approach: Database-level integrity, Application-level idempotency, and Concurrency management.

### Phase 1: Database Schema Hardening
We must move from "Application-enforced uniqueness" to "Database-enforced uniqueness."

1.  **Add an Idempotency Key:** Add a `client_request_id` column to the benchmark runs table.
2.  **Unique Constraint:** Apply a `UNIQUE` constraint to `client_request_id`.
3.  **WAL Mode:** Enable Write-Ahead Logging to allow concurrent reads and writes.

### Phase 2: Application Logic Update
Refactor the helper functions to use "Upsert" logic.

**Python Implementation Strategy:**
```python
import uuid
from sqlalchemy.dialects.sqlite import insert
from sqlalchemy import text

# 1. Ensure the client sends a unique ID for every logical request
# Example: X-Idempotency-Key: <uuid>

async def create_run(db, client_request_id: str, metadata: dict):
    """
    Uses SQLite's ON CONFLICT DO NOTHING to ensure that even if 
    two threads try to create the same run, only one succeeds.
    """
    # Use a raw SQL or SQLAlchemy's insert().on_conflict_do_nothing()
    # This prevents the "Check-then-Act" race condition.
    query = text("""
        INSERT INTO benchmark_runs (client_request_id, status, metadata, created_at)
        VALUES (:cid, :status, :meta, CURRENT_TIMESTAMP)
        ON CONFLICT(client_request_id) DO NOTHING
    """)
    
    result = await db.execute(query, {
        "cid": client_request_id,
        "status": "pending",
        "meta": json.dumps(metadata)
    })
    await db.commit()
    
    # If rowcount is 0, it means the run already exists.
    # We should fetch the existing one instead of creating a new one.
    if result.rowcount == 0:
        return await get_run_by_client_id(db, client_request_id)
    
    return await get_run_by_client_id(db, client_request_id)
```

### Phase 3: Lifecycle Management
Ensure `finish_llm_request` is always called using a `try...finally` block to prevent "zombie" runs that look like duplicates or failed attempts.

---

# SQLite Constraints And Indexes

To optimize for a home-lab environment while maintaining engineering standards, apply the following SQL migrations:

```sql
-- 1. Enable WAL mode for better concurrency
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;

-- 2. Add the idempotency column
ALTER TABLE benchmark_runs ADD COLUMN client_request_id TEXT;

-- 3. Create the unique constraint
-- Note: In SQLite, you may need to recreate the table if 
-- ALTER TABLE doesn't support adding UNIQUE constraints directly.
CREATE UNIQUE INDEX idx_unique_client_request ON benchmark_runs (client_request_id);

-- 4. Performance Indexes
CREATE INDEX idx_run_status ON benchmark_runs (status);
CREATE INDEX idx_run_created_at ON benchmark_runs (created_at DESC);

-- 5. Busy Timeout (Set in the connection string or via PRAGMA)
-- This prevents "Database is locked" errors by waiting up to 5 seconds
PRAGMA busy_timeout = 5000;
```

---

# Streaming Edge Cases

Streaming introduces complexity because the connection can drop at any point.

1.  **Partial Stream Success:** The server finishes the LLM call and writes to the DB, but the client disconnects before receiving the final chunk.
    *   *Fix:* The client must retry with the same `client_request_id`. The server will see the record exists and can potentially resume or return the cached result.
2.  **Server-Side Timeout:** The LLM takes 2 minutes, the proxy times out at 60 seconds. The server continues processing.
    *   *Fix:* Use a "heartbeat" or "status" update in the DB. If a request is "in_progress" but the client is gone, the `finish_llm_request` should still update the final status.
3.  **Concurrent Stream Retries:** A client retries a stream while the first stream is still being processed by the LLM provider.
    *   *Fix:* The `create_run` logic must check if a run with that `client_request_id` is already `in_progress`. If so, return a `409 Conflict` or the existing stream object.

---

# Test Plan

The following 12 tests must pass to validate the fix:

### Concurrency & Idempotency Tests
1.  **Race Condition Test:** Fire 10 concurrent requests with the *same* `client_request_id`. Verify exactly 1 row is created.
2.  **Idempotency Test:** Send a request, receive a 200, then send the exact same request again. Verify the second response returns the existing run data without creating a new row.
3.  **Rapid Fire Test:** Send 50 different requests in 1 second. Verify all 50 are created correctly.
4.  **Mixed Client Test:** Simulate OpenWebUI and an Agent client sending requests simultaneously. Verify no cross-contamination of `run_id`s.

### Streaming & Lifecycle Tests
5.  **Client Disconnect (Mid-Stream):** Kill the client connection during a stream. Verify `finish_llm_request` still updates the DB status to `completed` or `failed`.
6.  **Server Crash Simulation:** Force-kill the FastAPI process during a stream. Verify that upon restart, the run is marked as `interrupted` (via a cleanup task) or remains in a state that doesn't allow duplicate creation on retry.
7.  **Retry on 504:** Simulate a proxy timeout. Client retries with same ID. Verify server returns the existing run.
8.  **Long-Running Stream:** Run a 5-minute stream. Verify that concurrent "status check" queries do not block the stream write.

### Database Integrity Tests
9.  **WAL Mode Verification:** Verify that a heavy read query does not block a write query.
10. **Constraint Violation:** Attempt to manually insert a duplicate `client_request_id` via SQL. Verify it fails.
11. **Busy Timeout Test:** Simulate a long-running write and attempt a second write. Verify the second write waits and succeeds rather than throwing "Database is locked."
12. **Data Integrity:** Verify that `metadata` JSON remains valid and uncorrupted after multiple concurrent updates.

---

# Rollback Plan

If the patch introduces regressions (e.g., excessive `409 Conflict` errors or database locking issues):

1.  **Immediate Action:** Revert the `create_run` logic to the previous version and disable the `UNIQUE` constraint (if possible) or drop the index.
2.  **Database Cleanup:** Run a script to identify and delete duplicates based on the `created_at` timestamp (keeping the earliest entry for each unique metadata set).
3.  **Configuration Reversion:** Revert `PRAGMA journal_mode` to `DELETE` if WAL mode causes issues with the specific filesystem of the home lab.
4.  **Client-Side Fallback:** If the `client_request_id` is causing issues with the Agent client, provide a fallback where the server generates a hash of the request body to use as a temporary idempotency key.

---

# Decision Summary

**Selected Solution:** Implement **Database-Enforced Idempotency** using a `client_request_id` and SQLite's `ON CONFLICT DO NOTHING` clause.

**Reasoning:**
*   **Reliability:** Moving the constraint to the database is the only way to guarantee 100% prevention of duplicates in a concurrent environment. Application-level checks are inherently prone to race conditions.
*   **Simplicity:** This avoids the need for a distributed lock manager (like Redis), which is overkill for a home lab.
*   **Compatibility:** SQLite's WAL mode and `busy_timeout` provide sufficient concurrency for a single-user/small-group home lab.
*   **Streaming Support:** By using an idempotency key, we allow clients to safely retry failed streams without fear of creating "ghost" runs, which is the most common cause of duplicate rows in LLM applications.

**Trade-offs:**
*   Requires clients to generate and send a UUID. If a client cannot do this, we will implement a server-side hashing of the request body as a secondary idempotency key.
*   Requires a one-time migration to add the index and column.

**Detailed Implementation Guide**

To implement the proposed solution, the following code structure is recommended. This uses SQLAlchemy for ORM and Pydantic for data validation, ensuring that the FastAPI service remains robust and type-safe.

### 1. Database Schema (SQLAlchemy Models)
We define the `BenchmarkRun` model to include the `client_request_id` and the necessary constraints.

```python
import datetime
import uuid
from typing import Optional, Dict, Any
from sqlalchemy import Column, String, DateTime, JSON, Integer, Text
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.dialects.sqlite import JSON as SQLiteJSON

Base = declarative_base()

class BenchmarkRun(Base):
    __tablename__ = "benchmark_runs"

    # Primary Key
    id = Column(Integer, primary_key=True, autoincrement=True)
    
    # Idempotency Key - The core of the fix
    # This must be unique to prevent duplicate rows for the same logical request.
    client_request_id = Column(String(255), unique=True, nullable=False, index=True)
    
    # Metadata and Status
    status = Column(String(50), default="pending", index=True)
    metadata_json = Column(SQLiteJSON)
    
    # Timestamps
    created_at = Column(DateTime, default=datetime.datetime.utcnow)
    updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
    
    # LLM Specifics
    model_name = Column(String(100))
    tokens_used = Column(Integer, default=0)
    completion_time = Column(Float, nullable=True)

    def __repr__(self):
        return f"<BenchmarkRun(id={self.id}, client_id={self.client_request_id}, status={self.status})>"
```

### 2. Request Schemas (Pydantic)
These schemas ensure that the incoming requests from OpenWebUI and the Agent client are validated before they reach the database logic.

```python
from pydantic import BaseModel, Field
from typing import Dict, Any, Optional

class RunCreateRequest(BaseModel):
    # The client must provide this. If missing, the server can generate one 
    # based on a hash of the prompt, but a provided UUID is preferred.
    client_request_id: str = Field(..., description="Unique ID for idempotency")
    model_name: str
    prompt: str
    parameters: Dict[str, Any] = {}

class RunResponse(BaseModel):
    run_id: int
    client_request_id: str
    status: str
    metadata: Dict[str, Any]
```

### 3. Refactored Helper Functions
The `create_run` function is the most critical change. It uses the `ON CONFLICT` clause to handle the race condition at the database level.

```python
from sqlalchemy.dialects.sqlite import insert
from sqlalchemy.orm import Session
import json

async def create_run(db: Session, request_data: RunCreateRequest) -> BenchmarkRun:
    """
    Creates a new benchmark run or returns the existing one if the 
    client_request_id already exists.
    """
    # Construct the insert statement with ON CONFLICT logic
    stmt = insert(BenchmarkRun).values(
        client_request_id=request_data.client_request_id,
        model_name=request_data.model_name,
        status="pending",
        metadata_json={
            "prompt": request_data.prompt,
            "parameters": request_data.parameters
        }
    )
    
    # If a conflict occurs on client_request_id, do nothing
    stmt = stmt.on_conflict_do_nothing(index_elements=['client_request_id'])
    
    result = db.execute(stmt)
    db.commit()

    # If rowcount is 0, it means the record already exists.
    if result.rowcount == 0:
        # Fetch the existing record
        return db.query(BenchmarkRun).filter(
            BenchmarkRun.client_request_id == request_data.client_request_id
        ).first()
    
    # Otherwise, fetch the newly created record
    return db.query(BenchmarkRun).filter(
        BenchmarkRun.client_request_id == request_data.client_request_id
    ).first()

async def finish_llm_request(db: Session, run_id: int, status: str, tokens: int, duration: float):
    """
    Updates the final state of a run. Uses a standard update to ensure
    the final state is captured even if the stream was interrupted.
    """
    run = db.query(BenchmarkRun).filter(BenchmarkRun.id == run_id).first()
    if run:
        run.status = status
        run.tokens_used = tokens
        run.completion_time = duration
        run.updated_at = datetime.datetime.utcnow()
        db.commit()
    return run
```

### 4. FastAPI Route Integration
The endpoint handles the logic of checking for existing runs and managing the streaming response.

```python
from fastapi import FastAPI, Depends, HTTPException, Header
from sqlalchemy.orm import sessionmaker
from sqlalchemy import create_engine

# Database Setup
DATABASE_URL = "sqlite:///./benchmarks.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.post("/benchmark", response_model=RunResponse)
async def start_benchmark(
    request: RunCreateRequest, 
    db: Session = Depends(get_db)
):
    # 1. Attempt to create or retrieve the run
    run = await create_run(db, request)
    
    # 2. If the run exists but is already 'completed', return it immediately
    if run.status == "completed":
        return {
            "run_id": run.id,
            "client_request_id": run.client_request_id,
            "status": run.status,
            "metadata": run.metadata_json
        }
    
    # 3. If the run exists and is 'in_progress', handle as a concurrent retry
    if run.status == "in_progress":
        # Logic here could be to return a 409 Conflict or the current stream
        raise HTTPException(status_code=409, detail="Request already in progress")

    # 4. Proceed with LLM logic...
    # (LLM logic would go here, calling finish_llm_request at the end)
    return {
        "run_id": run.id,
        "client_request_id": run.client_request_id,
        "status": run.status,
        "metadata": run.metadata_json
    }
```

**Migration Strategy**

To transition the existing production database to this new schema without losing data, follow these steps:

1.  **Backup:** Create a full copy of `benchmarks.db`.
2.  **Schema Update:**
    *   Since SQLite has limited `ALTER TABLE` support, the safest way to add a `UNIQUE` constraint is to create a new table, copy the data, and rename it.
    *   However, for a home lab, we can start by adding the column and the index.
3.  **Data Population:**
    *   For existing rows, we need to generate `client_request_id`s.
    *   We can use a script to generate a UUID for every existing row where `client_request_id` is NULL.
4.  **Migration Script (Python):**

```python
import sqlite3
import uuid

def migrate_database():
    conn = sqlite3.connect("benchmarks.db")
    cursor = conn.cursor()

    # 1. Add the column if it doesn't exist
    try:
        cursor.execute("ALTER TABLE benchmark_runs ADD COLUMN client_request_id TEXT")
    except sqlite3.OperationalError:
        pass # Column already exists

    # 2. Populate existing rows with unique IDs
    cursor.execute("SELECT id FROM benchmark_runs WHERE client_request_id IS NULL")
    rows = cursor.fetchall()
    for row in rows:
        new_id = str(uuid.uuid4())
        cursor.execute(
            "UPDATE benchmark_runs SET client_request_id = ? WHERE id = ?",
            (new_id, row[0])
        )
    
    # 3. Create the unique index
    try:
        cursor.execute("CREATE UNIQUE INDEX idx_unique_client_request ON benchmark_runs (client_request_id)")
    except sqlite3.OperationalError as e:
        print(f"Index creation note: {e}")

    conn.commit()
    conn.close()
    print("Migration complete.")

if __name__ == "__main__":
    migrate_database()
```

**Monitoring and Observability**

To ensure the fix is working and to catch future issues, implement the following:

1.  **Structured Logging:**
    Log every attempt to create a run with the `client_request_id`.
    ```json
    {
      "timestamp": "2023-10-27T10:00:00Z",
      "level": "INFO",
      "event": "create_run_attempt",
      "client_request_id": "uuid-123",
      "result": "success",
      "is_duplicate": false
    }
    ```
2.  **Duplicate Detection Metric:**
    Increment a counter whenever `result.rowcount == 0` in the `create_run` function. This tells you how often clients are retrying or how often race conditions are being successfully mitigated.
3.  **Database Lock Monitoring:**
    Log any `sqlite3.OperationalError` related to "database is locked." If these occur frequently, it indicates the `busy_timeout` needs to be increased or the write volume is exceeding SQLite's capabilities.
4.  **Alerting:**
    Set an alert if the "Duplicate Detection Metric" exceeds a threshold (e.g., > 10% of total requests), which might indicate a bug in the client-side retry logic.

**Client-Side Integration**

For the fix to be effective, the clients (OpenWebUI, Agent, etc.) must be updated to handle idempotency.

**Agent Client (Python Example):**
```python
import requests
import uuid

def run_benchmark(prompt, model):
    # Generate a unique ID for this specific logical action
    idempotency_key = str(uuid.uuid4())
    
    payload = {
        "client_request_id": idempotency_key,
        "model_name": model,
        "prompt": prompt,
        "parameters": {"temperature": 0.7}
    }
    
    # If the request fails due to a network error, 
    # the client can safely retry with the SAME idempotency_key.
    try:
        response = requests.post("http://proxy/benchmark", json=payload)
        return response.json()
    except requests.exceptions.RequestException:
        # Retry logic with the same idempotency_key
        return run_benchmark(prompt, model) 
```

**Deep Dive: SQLite Concurrency Mechanics**

To understand why this fix works, we must look at how SQLite handles concurrent access:

1.  **The Locking Model:** SQLite uses file-level locking. When a process wants to write, it must acquire a `RESERVED` lock, and eventually an `EXCLUSIVE` lock. Only one process can hold an `EXCLUSIVE` lock at a time.
2.  **The Race Condition:** In the original code, the "Check" (SELECT) and the "Act" (INSERT) were two separate transactions.
    *   Thread A: SELECT (No row exists)
    *   Thread B: SELECT (No row exists)
    *   Thread A: INSERT (Success)
    *   Thread B: INSERT (Success) -> **Duplicate Created**
3.  **The Fix (Atomic Operation):** By using `INSERT ... ON CONFLICT DO NOTHING`, we move the logic into a single atomic operation. SQLite ensures that the check and the insert happen as one indivisible unit. If two threads attempt this simultaneously, SQLite's internal locking will force one to wait, and then the second one will see the conflict and do nothing.
4.  **WAL Mode (Write-Ahead Logging):**
    *   In standard `DELETE` journal mode, a write operation blocks all readers.
    *   In `WAL` mode, SQLite writes changes to a separate `-wal` file. This allows readers to continue reading the original database file while a write is occurring. This is essential for a FastAPI service where many users might be viewing results (reads) while the agent is creating new ones (writes).
5.  **Busy Timeout:**
    *   If a write is happening and another write comes in, SQLite immediately returns "Database is locked."
    *   `PRAGMA busy_timeout = 5000;` tells SQLite to wait up to 5 seconds for the lock to clear before throwing an error. This is the "engineering-grade" way to handle high-concurrency in a home lab.

**Extended Edge Case Analysis**

1.  **The "Zombie" Run:**
    *   *Scenario:* A request starts, `create_run` succeeds, but the LLM provider's API hangs indefinitely.
    *   *Solution:* Implement a "heartbeat" or a "timeout" check. A background task should look for runs in `pending` status for more than 10 minutes and mark them as `failed`.
2.  **The "Double Stream" Problem:**
    *   *Scenario:* A client starts a stream. The connection drops. The client retries with the same `client_request_id`. The server sees the run is already `in_progress`.
    *   *Solution:* The server should check if the `in_progress` run is still active. If the previous stream's connection is dead, the server can "hijack" the existing run and start a new stream for the new connection.
3.  **Metadata Collision:**
    *   *Scenario:* Two different users accidentally use the same `client_request_id` (highly unlikely with UUIDs, but possible with manual IDs).
    *   *Solution:* The `UNIQUE` constraint will prevent the second user from creating the run. The server should return a `409 Conflict`, and the client should be instructed to generate a new ID.
4.  **SQLite File Corruption:**
    *   *Scenario:* A power failure in the home lab during a write.
    *   *Solution:* `PRAGMA synchronous=NORMAL` combined with `WAL` mode provides a very high level of durability. For even higher safety, `PRAGMA synchronous=FULL` can be used, though it reduces write speed.

**Security and Validation**

1.  **SQL Injection:**
    *   By using SQLAlchemy's `insert()` and `db.execute(stmt, params)`, we use parameterized queries. This completely mitigates SQL injection risks.
2.  **UUID Validation:**
    *   The `client_request_id` should be validated to ensure it follows a standard UUID format. This prevents users from injecting malicious strings into the unique index.
    *   *Pydantic Validation:* `client_request_id: str = Field(..., regex=r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$")`
3.  **Rate Limiting:**
    *   Even with idempotency, a malicious actor could spam the endpoint with millions of unique `client_request_id`s to fill up the SQLite file.
    *   *Fix:* Implement a simple rate limiter (e.g., `slowapi`) to limit the number of `create_run` calls per IP address.

**Final Verification Checklist**

Before deploying to the home lab, verify the following:
- [ ] `PRAGMA journal_mode=WAL;` is confirmed active.
- [ ] `PRAGMA busy_timeout=5000;` is set in the connection string.
- [ ] The `benchmark_runs` table has a `UNIQUE` index on `client_request_id`.
- [ ] The `create_run` helper uses `ON CONFLICT DO NOTHING`.
- [ ] All client-side code (Agent, OpenWebUI) generates a UUID for every new request.
- [ ] The `finish_llm_request` helper is wrapped in a `try...finally` block to ensure status updates.
- [ ] The migration script has been run and verified against a backup.
- [ ] The `metadata_json` column is correctly handling JSON types in SQLite.
- [ ] The `409 Conflict` error is handled gracefully by the client.
- [ ] Logs are correctly capturing `client_request_id` for every request.
- [ ] The `tokens_used` and `completion_time` are being updated correctly.
- [ ] The system handles a 504 Gateway Timeout from the proxy by allowing the client to retry with the same ID.