# Root Cause Hypotheses  

| # | Hypothesis | Why it could happen | Typical symptoms |
|---|-------------|---------------------|------------------|
| 1 | **Race condition in `create_run`** | Multiple concurrent requests hit `create_run` before the first commit. SQLite’s default isolation level is *DEFERRED*, so the first transaction may not lock the table until the first write. | Duplicate rows with identical `run_id` or `run_uuid`. |
| 2 | **Missing unique constraint on `run_id`** | The schema defines `run_id` as a normal column, not `UNIQUE`. | Two rows with the same `run_id` but different `run_uuid`. |
| 3 | **SQLite WAL mode mis‑configured** | The app uses `PRAGMA journal_mode=WAL` but the connection pool re‑uses the same connection across threads. WAL allows readers to run concurrently, but writers still serialize. If the same connection is used, the writer may not see its own uncommitted changes. | Duplicate rows when two requests use the same connection. |
| 4 | **Connection pooling across threads** | FastAPI’s `Depends` injects a single `sqlite3.Connection` object that is shared across requests. SQLite connections are *not* thread‑safe unless `check_same_thread=False`. | `sqlite3.ProgrammingError: SQLite objects created in a thread can only be used in that same thread`. |
| 5 | **`INSERT OR IGNORE` used incorrectly** | The code uses `INSERT OR IGNORE` to avoid duplicates, but the ignore clause is triggered by a *different* constraint (e.g., a foreign key violation) rather than the `run_id`. | Rows are silently dropped, leading to missing data and duplicate `run_id` from a later request. |
| 6 | **`create_llm_request` called twice** | The request handler may call `create_llm_request` twice for the same LLM request (e.g., once for the initial request and once for a retry). | Two `llm_request` rows with the same `request_id`. |
| 7 | **Streaming responses cause early termination** | When the client aborts a streaming request, the connection may close before the transaction commits. The next request may then insert a new row with the same `run_id`. | Duplicate rows with different timestamps. |
| 8 | **`finish_llm_request` called before `create_run`** | The code may finish the LLM request before the run is fully created, causing a foreign key violation that is ignored. | Rows with `run_id` but missing `llm_request_id`. |
| 9 | **`add_event` called after commit** | `add_event` may be called after the transaction has been committed, but the event is inserted into a *different* connection that is not part of the same transaction. | Events that appear out of order or duplicated. |
| 10 | **SQLite autocommit mode** | The default autocommit mode may cause each `INSERT` to be its own transaction, allowing interleaving of writes. | Duplicate rows when two requests interleave. |
| 11 | **Missing `PRAGMA foreign_keys=ON`** | Without foreign key enforcement, the code may insert orphaned rows that later get duplicated. | Duplicate `llm_request` rows with no parent `run`. |
| 12 | **Inconsistent use of `run_uuid`** | Some code paths generate a new UUID for each request, others reuse the same one. | Duplicate rows with same `run_uuid` but different `run_id`. |

---

# Evidence To Collect  

| # | Evidence | How to collect | Why it matters |
|---|----------|----------------|----------------|
| 1 | **Full request logs** | Add a middleware that logs `request_id`, `user_id`, `endpoint`, `timestamp`, and `thread_id`. | Helps correlate duplicate rows with concurrent requests. |
| 2 | **SQLite PRAGMA settings** | Run `PRAGMA journal_mode; PRAGMA synchronous; PRAGMA foreign_keys; PRAGMA busy_timeout;` before each request. | Confirms the database configuration. |
| 3 | **Connection pool size** | Inspect `app.state.db_pool` or `sqlalchemy.create_engine(..., pool_size=...)`. | Determines if connections are reused. |
| 4 | **Transaction boundaries** | Wrap `create_run`, `create_llm_request`, and `add_event` in a single `BEGIN IMMEDIATE` block. Log `BEGIN` and `COMMIT`. | Shows if transactions are nested or missing. |
| 5 | **Duplicate `run_id` detection** | After each insert, query `SELECT COUNT(*) FROM runs WHERE run_id=?`. | Detects duplicates early. |
| 6 | **SQLite WAL page count** | Query `PRAGMA wal_checkpoint;` and `PRAGMA wal_autocheckpoint;`. | Indicates if WAL pages accumulate. |
| 7 | **Thread IDs in logs** | Log `threading.get_ident()` for each request. | Confirms concurrency. |
| 8 | **Event ordering** | Query `SELECT * FROM events ORDER BY event_timestamp;`. | Detects out‑of‑order events. |
| 9 | **Schema introspection** | `PRAGMA table_info('runs'); PRAGMA index_list('runs');`. | Confirms constraints. |
| 10 | **Index usage** | `EXPLAIN QUERY PLAN SELECT * FROM runs WHERE run_id=?;`. | Confirms if indexes are used. |
| 11 | **Busy timeout logs** | Log when SQLite returns `SQLITE_BUSY`. | Indicates contention. |
| 12 | **Connection lifetime** | Log `connection.isolation_level` and `connection.in_transaction`. | Confirms transaction state. |
| 13 | **Error logs** | Capture `sqlite3.OperationalError`, `sqlite3.IntegrityError`. | Detects constraint violations. |
| 14 | **Retry logic** | Log when a request is retried. | Detects duplicate inserts from retries. |
| 15 | **Streaming request aborts** | Log `client_disconnect` events. | Detects early termination. |
| 16 | **Database backup** | Take a `sqlite3` dump before patch. | Baseline for rollback. |
| 17 | **Concurrency test harness** | Run a stress test with 100 concurrent requests and capture logs. | Simulates production load. |
| 18 | **Transaction isolation level** | Query `PRAGMA read_uncommitted;`. | Confirms isolation. |
| 19 | **Foreign key enforcement** | Query `PRAGMA foreign_keys;`. | Confirms enforcement. |
| 20 | **Connection pool metrics** | Log `pool.checkout()` and `pool.checkin()`. | Confirms pool usage. |

---

# Patch Plan  

## 1. Schema Changes  

```sql
-- 1.1 Add a unique constraint on run_id
ALTER TABLE runs ADD CONSTRAINT uq_runs_run_id UNIQUE (run_id);

-- 1.2 Add a unique constraint on llm_request_id
ALTER TABLE llm_requests ADD CONSTRAINT uq_llm_requests_request_id UNIQUE (request_id);

-- 1.3 Add a unique constraint on event_id
ALTER TABLE events ADD CONSTRAINT uq_events_event_id UNIQUE (event_id);

-- 1.4 Add foreign key constraints (if not already present)
ALTER TABLE llm_requests
  ADD CONSTRAINT fk_llm_requests_run_id
  FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE;

ALTER TABLE events
  ADD CONSTRAINT fk_events_llm_request_id
  FOREIGN KEY (llm_request_id) REFERENCES llm_requests(request_id) ON DELETE CASCADE;
```

## 2. Indexes  

```sql
-- 2.1 Index on run_id (already unique)
CREATE INDEX IF NOT EXISTS idx_runs_run_id ON runs(run_id);

-- 2.2 Composite index for fast lookups
CREATE INDEX IF NOT EXISTS idx_llm_requests_run_id_request_id
  ON llm_requests(run_id, request_id);

-- 2.3 Index on event_timestamp for ordering
CREATE INDEX IF NOT EXISTS idx_events_timestamp
  ON events(event_timestamp);

-- 2.4 Composite index for event queries
CREATE INDEX IF NOT EXISTS idx_events_type_timestamp
  ON events(event_type, event_timestamp);
```

## 3. Transaction Handling  

```python
# db.py
import sqlite3
from contextlib import contextmanager

@contextmanager
def transaction(conn: sqlite3.Connection):
    """Context manager that starts an IMMEDIATE transaction and commits/rolls back."""
    try:
        conn.execute("BEGIN IMMEDIATE;")
        yield
        conn.execute("COMMIT;")
    except Exception:
        conn.execute("ROLLBACK;")
        raise
```

Wrap all database writes in `transaction(conn)`:

```python
def create_run(conn, run_id, run_uuid, user_id, ...):
    with transaction(conn):
        conn.execute(
            "INSERT INTO runs (run_id, run_uuid, user_id, created_at) VALUES (?, ?, ?, ?)",
            (run_id, run_uuid, user_id, datetime.utcnow()),
        )
```

## 4. Connection Management  

```python
# main.py
from fastapi import FastAPI
from fastapi import Depends
import sqlite3

app = FastAPI()

def get_db() -> sqlite3.Connection:
    conn = sqlite3.connect("app.db", check_same_thread=False)
    conn.execute("PRAGMA foreign_keys=ON;")
    conn.execute("PRAGMA journal_mode=WAL;")
    conn.execute("PRAGMA synchronous=NORMAL;")
    conn.execute("PRAGMA busy_timeout=5000;")
    return conn

@app.middleware("http")
async def db_middleware(request, call_next):
    request.state.db = get_db()
    try:
        response = await call_next(request)
    finally:
        request.state.db.close()
    return response
```

Each request now gets its own connection; no pooling across threads.

## 5. Guardrails in Python Code  

```python
# utils.py
def safe_insert(conn, sql, params):
    """Insert and return the lastrowid, raising if duplicate."""
    try:
        cur = conn.execute(sql, params)
        return cur.lastrowid
    except sqlite3.IntegrityError as e:
        # Log duplicate and raise a custom exception
        logger.warning(f"Duplicate insert: {sql} {params} -> {e}")
        raise DuplicateInsertError from e
```

Use `safe_insert` for all inserts.

## 6. Streaming Edge‑Case Handling  

- **Abort detection**: Use `request.is_disconnected()` (FastAPI) to detect client disconnects.  
- **Transaction rollback**: If disconnect occurs before commit, the transaction is rolled back automatically by the context manager.  
- **Event finalization**: Ensure `finish_llm_request` is called only after the run is fully committed.  

```python
async def stream_handler(request: Request):
    db = request.state.db
    try:
        run_id = create_run(db, ...)
        # ... streaming logic ...
    except Exception:
        # On any exception, the transaction will rollback
        raise
```

## 7. Retry Logic  

- **Idempotency key**: Use `run_uuid` as the idempotency key.  
- **Retry guard**: If a request with the same `run_uuid` is received, fetch the existing run instead of inserting a new one.  

```python
def get_or_create_run(conn, run_uuid):
    cur = conn.execute("SELECT run_id FROM runs WHERE run_uuid=?", (run_uuid,))
    row = cur.fetchone()
    if row:
        return row[0]
    else:
        return create_run(conn, uuid4(), run_uuid, ...)
```

---

# SQLite Constraints And Indexes  

## 1. Unique Constraints  

```sql
ALTER TABLE runs ADD CONSTRAINT uq_runs_run_id UNIQUE (run_id);
ALTER TABLE llm_requests ADD CONSTRAINT uq_llm_requests_request_id UNIQUE (request_id);
ALTER TABLE events ADD CONSTRAINT uq_events_event_id UNIQUE (event_id);
```

## 2. Foreign Key Constraints  

```sql
ALTER TABLE llm_requests
  ADD CONSTRAINT fk_llm_requests_run_id
  FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE;

ALTER TABLE events
  ADD CONSTRAINT fk_events_llm_request_id
  FOREIGN KEY (llm_request_id) REFERENCES llm_requests(request_id) ON DELETE CASCADE;
```

## 3. Indexes  

| Table | Index | Purpose |
|-------|-------|---------|
| runs | `idx_runs_run_id` | Fast lookup by `run_id`. |
| llm_requests | `idx_llm_requests_run_id_request_id` | Query by run and request. |
| events | `idx_events_timestamp` | Order events chronologically. |
| events | `idx_events_type_timestamp` | Query by type and time. |

## 4. PRAGMA Settings  

```sql
PRAGMA journal_mode=WAL;
PRAGMA synchronous=NORMAL;
PRAGMA foreign_keys=ON;
PRAGMA busy_timeout=5000;
PRAGMA temp_store=MEMORY;
PRAGMA cache_size=10000;
```

These settings give a good balance between performance and consistency for a home‑lab environment.

---

# Streaming Edge Cases  

| # | Edge Case | Why it matters | Mitigation |
|---|-----------|----------------|------------|
| 1 | **Client aborts mid‑stream** | Connection closed before commit → transaction rolled back. | Detect disconnect via `request.is_disconnected()`; if true, abort streaming and let transaction rollback. |
| 2 | **Partial writes to `events`** | Streaming may write events in chunks; if a chunk fails, earlier events may persist. | Buffer events in memory until stream completes, then bulk insert in one transaction. |
| 3 | **Duplicate stream start** | Client may retry the stream due to network glitch. | Use `run_uuid` as idempotency key; if run exists, skip insert. |
| 4 | **Backpressure** | FastAPI may send data faster than DB can handle. | Use `asyncio.sleep` to throttle writes; or batch writes. |
| 5 | **Large payloads** | Streaming large LLM responses may exceed memory. | Stream to disk or use generator. |
| 6 | **Event ordering** | Events may arrive out of order if streaming is interrupted. | Store `event_timestamp` and sort on retrieval. |
| 7 | **Connection reuse** | Same DB connection reused across streams may cause lock contention. | Use per‑request connection. |
| 8 | **WAL checkpoint** | WAL pages may accumulate if not checkpointed. | Call `PRAGMA wal_checkpoint;` after each commit. |
| 9 | **SQLite busy** | Concurrent writes may hit `SQLITE_BUSY`. | Set `busy_timeout=5000` and retry. |
|10 | **Transaction isolation** | `BEGIN IMMEDIATE` ensures writer locks early. | Use `transaction(conn)` context manager. |
|11 | **Event duplication** | Streaming may re‑emit events on retry. | Use `event_id` as unique key; ignore duplicates. |
|12 | **Error handling** | Streaming errors may not propagate to DB layer. | Wrap streaming in try/except and rollback on error. |

---

# Test Plan  

## 1. Acceptance Tests (12+)  

| # | Test | Description | Expected Result |
|---|------|-------------|-----------------|
| 1 | **CreateRunUnique** | Insert a run with a given `run_id`. Attempt to insert again with same `run_id`. | Second insert raises `DuplicateInsertError`. |
| 2 | **CreateLLMRequestUnique** | Insert an LLM request with a given `request_id`. Attempt to insert again. | Duplicate raises `DuplicateInsertError`. |
| 3 | **CreateEventUnique** | Insert an event with a given `event_id`. Attempt to insert again. | Duplicate raises `DuplicateInsertError`. |
| 4 | **ConcurrentRunInsert** | Spin up 10 concurrent requests each inserting a unique run. | All 10 rows exist, no duplicates. |
| 5 | **ConcurrentRunInsertSameUUID** | Spin up 10 concurrent requests each using the same `run_uuid`. | Only one row inserted; others fetch existing. |
| 6 | **StreamingAbort** | Start a streaming request, then abort client. | Transaction rolled back; no run row. |
| 7 | **StreamingComplete** | Start a streaming request, let it finish. | Run, LLM request, and events inserted correctly. |
| 8 | **RetryStream** | Simulate a network glitch causing a retry of a streaming request. | No duplicate rows. |
| 9 | **EventOrdering** | Insert events with timestamps out of order. | Retrieval sorted by timestamp. |
|10 | **WALCheckpoint** | Insert many rows, then call `PRAGMA wal_checkpoint;`. | WAL file size reduced. |
|11 | **ForeignKeyEnforcement** | Attempt to insert an LLM request with non‑existent `run_id`. | Raises `sqlite3.IntegrityError`. |
|12 | **TransactionRollback** | Force an exception during a transaction. | All changes rolled back. |
|13 | **ConnectionIsolation** | Verify that each request gets its own connection. | No `check_same_thread` errors. |
|14 | **PRAGMACheck** | Verify that `PRAGMA foreign_keys=ON` is set. | Query returns 1. |
|15 | **IndexUsage** | Run `EXPLAIN QUERY PLAN` on a query. | Index used. |
|16 | **DuplicateEventOnRetry** | Retry a streaming request that already inserted events. | No duplicate events. |
|17 | **LargePayloadStreaming** | Stream a large LLM response (~10 MB). | No memory overflow; DB writes succeed. |
|18 | **BackpressureHandling** | Simulate high throughput; ensure DB writes keep up. | No `SQLITE_BUSY` errors. |
|19 | **EventBatchInsert** | Insert 100 events in a single transaction. | All events present. |
|20 | **RollbackOnDisconnect** | Disconnect client mid‑stream; ensure rollback. | No partial data. |

## 2. Integration Tests  

- **API Endpoints**: Test `/runs`, `/llm_requests`, `/events` endpoints for CRUD.  
- **Concurrency**: Use `pytest-asyncio` to fire 100 concurrent requests.  
- **Streaming**: Use `httpx` or `httpx.AsyncClient` to stream responses.  

## 3. Performance Tests  

- **Throughput**: Measure inserts per second under 50 concurrent users.  
- **Latency**: Measure average response time for streaming requests.  

## 4. Stress Tests  

- **Long‑Running**: Run for 2 hours with 200 concurrent users.  
- **Resource Utilization**: Monitor CPU, memory, and disk I/O.  

## 5. Edge‑Case Tests  

- **Large Event Payload**: Insert an event with 1 MB JSON.  
- **Invalid JSON**: Insert event with malformed JSON; expect error.  

## 6. Database Consistency Tests  

- **Snapshot**: After each test, dump DB to a file and compare with baseline.  

---

# Rollback Plan  

1. **Backup**  
   - Run `sqlite3 app.db ".backup 'app.db.bak'"`.  
   - Store backup in version control or external storage.  

2. **Migration Script**  
   ```sql
   -- 1. Drop unique constraints
   ALTER TABLE runs DROP CONSTRAINT uq_runs_run_id;
   ALTER TABLE llm_requests DROP CONSTRAINT uq_llm_requests_request_id;
   ALTER TABLE events DROP CONSTRAINT uq_events_event_id;

   -- 2. Drop indexes
   DROP INDEX IF EXISTS idx_runs_run_id;
   DROP INDEX IF EXISTS idx_llm_requests_run_id_request_id;
   DROP INDEX IF EXISTS idx_events_timestamp;
   DROP INDEX IF EXISTS idx_events_type_timestamp;
   ```

3. **Code Revert**  
   - Revert `transaction(conn)` context manager.  
   - Revert `safe_insert` to plain `execute`.  
   - Revert connection pooling to original.  

4. **Testing**  
   - Run the original test suite to confirm baseline behavior.  

5. **Deployment**  
   - Deploy rollback commit to staging.  
   - Monitor logs for duplicate rows.  

---

# Decision Summary  

- **Root Cause**: The combination of missing unique constraints, shared SQLite connections across threads, and lack of proper transaction boundaries caused race conditions that produced duplicate rows.  
- **Fix**: Add unique constraints, enforce foreign keys, use per‑request connections with `check_same_thread=False`, wrap all writes in an `IMMEDIATE` transaction context manager, and guard inserts with `safe_insert`.  
- **SQLite Constraints**: Unique constraints on `run_id`, `request_id`, and `event_id`; foreign keys on `llm_request.run_id` and `event.llm_request_id`.  
- **Index Strategy**: Add indexes on `run_id`, composite `run_id, request_id`, and event timestamps.  
- **Streaming Edge Cases**: Handle client aborts, backpressure, and event ordering by buffering events until stream completion and using idempotency keys.  
- **Test Plan**: 20+ acceptance tests covering concurrency, streaming, rollback, and database consistency.  
- **Rollback**: Backup DB, drop constraints and indexes, revert code, and redeploy.  

This patch ensures that duplicate rows cannot be inserted, that concurrent requests are serialized correctly, and that streaming requests are handled safely even under high load. The solution is fully compatible with a private home‑lab environment while maintaining engineering‑grade robustness.