# Root Cause Hypotheses

The intermittent duplication of benchmark rows in a FastAPI/SQLite stack under concurrent proxy traffic typically stems from a combination of missing database-level idempotency, async concurrency race conditions, and streaming lifecycle misalignment. Below are the primary hypotheses, ranked by likelihood in this architecture:

1. **Missing UNIQUE Constraints on Logical Identifiers**: The `runs` table likely lacks a `UNIQUE` constraint on the business key that identifies a benchmark run (e.g., `run_id`, `session_id`, or `trace_id`). Without it, SQLite treats every `INSERT` as a new row, even if the application logically considers them duplicates.

2. **Async Race Condition in `create_run`**: FastAPI handles concurrent requests via an event loop. If multiple async tasks (OpenWebUI, agent client, benchmark runner) invoke `create_run` simultaneously for the same logical run, and the function performs a `SELECT` followed by an `INSERT` without a transaction or lock, interleaving occurs. Both tasks see "not exists" and both insert.

3. **Proxy Retry & Streaming Reconnection Loops**: The proxy may retry failed or timed-out requests. If `create_run` is called at the start of request processing without checking for existing runs, retries spawn duplicates. Streaming endpoints (SSE) are particularly vulnerable: client disconnects, network blips, or proxy buffering can trigger re-connections that re-invoke the run creation logic.

4. **SQLite Connection/Thread Safety Misconfiguration**: Using the synchronous `sqlite3` module directly in an async FastAPI app without `aiosqlite` or proper `check_same_thread=False` + explicit locking leads to undefined behavior. SQLite's default locking mode may block or silently fail under concurrent writes, causing application-level retries that duplicate rows.

5. **Streaming Lifecycle Boundary Mismatch**: If `create_run` is inadvertently called per SSE chunk, per token, or on each `add_event` callback instead of exactly once per logical session, duplicates accumulate rapidly during high-throughput streaming.

6. **Lack of Transaction Boundaries & WAL Mode**: Inserts not wrapped in explicit `BEGIN`/`COMMIT` transactions, combined with SQLite's default `DELETE` journal mode, cause high lock contention. Under load, `SQLITE_BUSY` errors trigger application retries, which again insert duplicates if idempotency isn't enforced.

# Evidence To Collect

Before applying patches, gather the following telemetry and configuration data to validate hypotheses and avoid blind fixes:

1. **Schema & Constraints Audit**: Run `PRAGMA table_info(runs);` and `PRAGMA index_list(runs);` to verify existing keys, indexes, and constraints.
2. **Journal & Locking Mode**: Execute `PRAGMA journal_mode;` and `PRAGMA locking_mode;`. Confirm if WAL is enabled. Check `PRAGMA busy_timeout;`.
3. **Connection Configuration**: Inspect how the DB connection is initialized. Is `aiosqlite` used? Is `check_same_thread=False` set? Is there a connection pool or single shared connection?
4. **Application Logs**: Search for duplicate `run_id` or `trace_id` values within a 500ms window. Look for `SQLITE_BUSY`, `IntegrityError`, or retry stack traces.
5. **Proxy & Gateway Metrics**: Check retry counts, timeout thresholds, and streaming reconnection rates from OpenWebUI, agent client, and benchmark runners.
6. **Code Path Tracing**: Add temporary structured logging with `trace_id` and `coro_id` to `create_run`, `create_llm_request`, `finish_llm_request`, and `add_event`. Track call frequency per logical run.
7. **Concurrency Load Test**: Simulate 50-100 concurrent requests targeting the same benchmark endpoint. Measure duplicate rate and SQLite lock wait times.
8. **Streaming Boundary Analysis**: Capture raw SSE payloads. Verify if `create_run` is invoked on connection open, first chunk, or every chunk.

# Patch Plan

The fix must be defense-in-depth: database constraints for hard guarantees, application-level idempotency for clean semantics, and async-safe transaction management for concurrency.

### 1. Database-Level Idempotency (Primary Guardrail)
Add a `UNIQUE` constraint on the logical run identifier. Use `INSERT OR IGNORE` to safely handle concurrent attempts.

```python
# Migration SQL (run once via alembic or raw PRAGMA)
ALTER TABLE runs ADD COLUMN run_id TEXT UNIQUE NOT NULL;
-- If table must be rebuilt:
CREATE TABLE runs_new (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id TEXT UNIQUE NOT NULL,
    session_id TEXT,
    status TEXT DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO runs_new SELECT *, id AS run_id FROM runs;
DROP TABLE runs;
ALTER TABLE runs_new RENAME TO runs;
```

### 2. Async-Safe `create_run` Implementation
Wrap inserts in explicit transactions. Use `INSERT OR IGNORE` and check `lastrowid` or `changes()` to determine if a new row was created.

```python
import aiosqlite
import asyncio
from contextlib import asynccontextmanager

@asynccontextmanager
async def db_transaction(db: aiosqlite.Connection):
    await db.execute("BEGIN IMMEDIATE")
    try:
        yield db
        await db.commit()
    except Exception:
        await db.rollback()
        raise

async def create_run(db: aiosqlite.Connection, run_id: str, session_id: str):
    async with db_transaction(db) as tx:
        cursor = await tx.execute(
            "INSERT OR IGNORE INTO runs (run_id, session_id, status) VALUES (?, ?, 'pending')",
            (run_id, session_id)
        )
        await tx.execute("SELECT changes()")
        changes = await cursor.fetchone()
        created = changes[0] > 0
        return {"run_id": run_id, "created": created}
```

### 3. Application-Level Idempotency & Locking
For high-concurrency hot paths, add an `asyncio.Lock` keyed by `run_id` to serialize creation attempts before hitting SQLite.

```python
_run_locks: dict[str, asyncio.Lock] = {}

async def get_run_lock(run_id: str) -> asyncio.Lock:
    if run_id not in _run_locks:
        _run_locks[run_id] = asyncio.Lock()
    return _run_locks[run_id]

async def safe_create_run(db: aiosqlite.Connection, run_id: str, session_id: str):
    lock = await get_run_lock(run_id)
    async with lock:
        return await create_run(db, run_id, session_id)
```

### 4. Lifecycle Guardrails for `create_llm_request`, `finish_llm_request`, `add_event`
- Enforce state machine transitions: `pending -> running -> completed/failed`.
- Use `UPDATE ... WHERE status = ?` to prevent overwriting completed runs.
- Add `ON CONFLICT DO UPDATE` for event logs if deduplication is needed.

```python
async def create_llm_request(db: aiosqlite.Connection, run_id: str, request_id: str):
    async with db_transaction(db) as tx:
        await tx.execute(
            "UPDATE runs SET status = 'running' WHERE run_id = ? AND status = 'pending'",
            (run_id,)
        )
        await tx.execute(
            "INSERT OR IGNORE INTO llm_requests (request_id, run_id, status) VALUES (?, ?, 'active')",
            (request_id, run_id)
        )
```

### 5. Proxy & Retry Configuration
- Set `Idempotency-Key` header on client requests.
- Configure proxy to cache `200 OK` for idempotent creation endpoints.
- Disable automatic retries on `200`/`201` responses.

# SQLite Constraints And Indexes

SQLite's performance and correctness under concurrency depend heavily on schema design and pragma configuration.

### Schema Constraints
```sql
CREATE TABLE runs (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id TEXT UNIQUE NOT NULL,
    session_id TEXT NOT NULL,
    status TEXT CHECK(status IN ('pending', 'running', 'completed', 'failed')) DEFAULT 'pending',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE llm_requests (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    request_id TEXT UNIQUE NOT NULL,
    run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE,
    status TEXT CHECK(status IN ('active', 'completed', 'failed')) DEFAULT 'active',
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

CREATE TABLE events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    run_id TEXT NOT NULL REFERENCES runs(run_id) ON DELETE CASCADE,
    event_type TEXT NOT NULL,
    payload TEXT,
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
```

### Indexes
```sql
-- Speed up status transitions and lifecycle queries
CREATE INDEX idx_runs_status ON runs(status);
CREATE INDEX idx_runs_session ON runs(session_id);
CREATE INDEX idx_llm_requests_run ON llm_requests(run_id);
CREATE INDEX idx_events_run_type ON events(run_id, event_type);
```

### Pragma Configuration (Critical for Concurrency)
```sql
-- Enable Write-Ahead Logging for concurrent read/write safety
PRAGMA journal_mode = WAL;
-- Allow concurrent readers while writer holds lock
PRAGMA busy_timeout = 5000; -- 5 seconds
-- Optimize for home-lab SSD/NVMe
PRAGMA synchronous = NORMAL;
PRAGMA cache_size = -64000; -- 64MB
PRAGMA foreign_keys = ON;
```

### Conflict Resolution Strategy
- `INSERT OR IGNORE`: Safely drops duplicates, returns `changes() == 0`.
- `INSERT OR REPLACE`: Dangerous for benchmark rows; can reset timestamps/status. Avoid.
- `ON CONFLICT DO UPDATE`: Use only for upserts where state transitions are explicit.

# Streaming Edge Cases

Streaming endpoints (SSE/NDJSON) introduce unique concurrency and lifecycle challenges. The following edge cases must be handled explicitly:

1. **First-Chunk vs. Connection-Open Trigger**: `create_run` must fire exactly once when the logical run begins, not on every chunk. Use a flag or state check: `if not run_exists: await create_run()`.
2. **Client Disconnect & Reconnect**: If a client drops and reconnects, the proxy may treat it as a new request. Validate `run_id` in the reconnect payload. If it exists, resume streaming from the last event ID instead of creating a new run.
3. **Proxy Timeout & Retry Loops**: Configure the proxy to treat streaming endpoints as non-retryable on `200`. If retries are unavoidable, enforce idempotency via `Idempotency-Key` and `INSERT OR IGNORE`.
4. **Backpressure & Buffer Overflow**: FastAPI's `StreamingResponse` can block if the client is slow. Use `asyncio.Queue` with `maxsize` to prevent memory leaks. Do not call `add_event` synchronously inside the stream generator; use a background task or async queue.
5. **Graceful Shutdown**: On `SIGTERM` or `SIGINT`, flush pending events, mark run as `failed` or `completed`, and close DB connections. Use `asyncio.shield` to prevent cancellation mid-transaction.
6. **Chunk Boundary Misalignment**: Ensure `finish_llm_request` is called only once per run, typically on `[DONE]` or stream termination. Add a guard: `if status != 'running': return`.
7. **Eventual Consistency in Streaming**: If `add_event` fails due to DB lock, queue events in memory and retry with exponential backoff. Never block the stream generator on DB writes.

### Python Guardrails for Streaming
```python
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
import asyncio

async def stream_benchmark(run_id: str, db: aiosqlite.Connection):
    # Ensure run exists exactly once
    result = await safe_create_run(db, run_id, session_id="stream_session")
    if not result["created"]:
        # Resume or reject based on business logic
        pass

    async def event_generator():
        try:
            yield f"data: {{\"type\": \"start\", \"run_id\": \"{run_id}\"}}\n\n"
            # ... streaming logic ...
            yield f"data: [DONE]\n\n"
        finally:
            await finish_llm_request(db, run_id)

    return StreamingResponse(event_generator(), media_type="text/event-stream")
```

# Test Plan

The following 12 acceptance tests validate correctness, concurrency safety, streaming behavior, and rollback readiness. Each test includes setup, execution, and validation criteria.

| # | Test Name | Setup | Execution | Validation |
|---|-----------|-------|-----------|------------|
| 1 | **Concurrent `create_run` Deduplication** | 50 async tasks targeting same `run_id` | Fire all simultaneously | Exactly 1 row in `runs`; `changes() == 0` for 49 tasks |
| 2 | **UNIQUE Constraint Enforcement** | Schema with `run_id UNIQUE` | Attempt direct `INSERT` with duplicate `run_id` | `IntegrityError` raised; no duplicate row |
| 3 | **Streaming First-Chunk Guard** | Mock SSE endpoint | Send 100 chunks, verify `create_run` call count | `create_run` called exactly once |
| 4 | **Proxy Retry Idempotency** | Proxy configured to retry on 500 | Simulate 3 retries for same request | 1 run created; subsequent retries return existing run |
| 5 | **WAL Mode Concurrent Reads/Writes** | `PRAGMA journal_mode=WAL` | 10 writers, 50 readers concurrently | No `SQLITE_BUSY`; reads return consistent state |
| 6 | **Transaction Rollback on Failure** | Mock DB failure mid-transaction | Trigger `create_run` with forced exception | No partial rows; `ROLLBACK` executed; status unchanged |
| 7 | **State Machine Transition Safety** | Run in `completed` state | Call `create_llm_request` again | `UPDATE` affects 0 rows; no status overwrite |
| 8 | **Event Queue Backpressure** | `asyncio.Queue(maxsize=10)` | Flood `add_event` faster than DB can write | Queue blocks producer; no memory leak; events flushed |
| 9 | **Client Disconnect & Resume** | Stream active, kill client | Reconnect with same `run_id` | Stream resumes from last event; no new run created |
| 10 | **Graceful Shutdown Flush** | Send `SIGTERM` mid-stream | Monitor DB state | Run marked `failed`; pending events written; connection closed |
| 11 | **Index Performance Under Load** | 100k rows in `runs` | Query by `status` and `session_id` | Query time < 5ms; `EXPLAIN QUERY PLAN` uses indexes |
| 12 | **Schema Migration Safety** | Backup DB, run migration | Execute `ALTER TABLE` + rebuild | Data preserved; constraints active; zero downtime window |

### Test Execution Framework
- Use `pytest-asyncio` for async tests.
- Use `aiosqlite` with in-memory DB (`:memory:`) for speed, but run subset on disk with WAL.
- Use `locust` or `aiohttp` for concurrency simulation.
- Validate DB state via raw SQL queries post-test.

# Rollback Plan

Engineering-grade fixes require a safe, reversible deployment strategy. The following rollback plan ensures zero data loss and minimal downtime.

### 1. Pre-Deployment Safeguards
- **Full Backup**: `sqlite3 benchmark.db ".backup benchmark_backup_$(date +%Y%m%d_%H%M%S).db"`
- **Version Control**: Tag current codebase `v1.0.0-incident`.
- **Feature Flag**: Wrap new `create_run` logic behind `ENABLE_IDEMPOTENT_RUNS=true` env var.

### 2. Rollback Triggers
- Duplicate rate > 0.1% post-deployment.
- `SQLITE_BUSY` errors spike > 5%.
- Streaming latency increases > 200ms.
- Test suite fails in staging.

### 3. Rollback Procedure
1. **Disable Feature Flag**: Set `ENABLE_IDEMPOTENT_RUNS=false`. Restart service.
2. **Revert Schema (if needed)**:
   ```sql
   -- Only if migration broke compatibility
   ALTER TABLE runs DROP COLUMN run_id; -- SQLite 3.35+
   -- Or restore from backup
   sqlite3 benchmark.db ".restore benchmark_backup_YYYYMMDD_HHMMSS.db"
   ```
3. **Code Revert**: `git revert <commit-hash>`; redeploy.
4. **Data Reconciliation**: Run script to merge duplicate rows if any slipped through:
   ```python
   async def reconcile_duplicates(db):
       async with db.execute("SELECT run_id, COUNT(*) FROM runs GROUP BY run_id HAVING COUNT(*) > 1") as cur:
           duplicates = await cur.fetchall()
       for run_id, count in duplicates:
           # Keep oldest, delete others
           await db.execute("DELETE FROM runs WHERE run_id = ? AND id NOT IN (SELECT MIN(id) FROM runs WHERE run_id = ?)", (run_id, run_id))
   ```

### 4. Post-Rollback Validation
- Verify duplicate rate returns to pre-incident baseline.
- Confirm streaming endpoints function normally.
- Audit logs for `ROLLBACK` events and constraint violations.

# Decision Summary

The intermittent duplicate benchmark rows stem from a combination of missing database-level idempotency, async concurrency race conditions in `create_run`, and streaming lifecycle misalignment. The recommended fix implements a defense-in-depth strategy:

1. **Database Constraints**: Add `UNIQUE` on `run_id`, enforce `INSERT OR IGNORE`, and configure WAL mode with `busy_timeout` for concurrent safety.
2. **Application Guardrails**: Wrap DB operations in explicit async transactions, add `asyncio.Lock` per run for serialization, and enforce state machine transitions for lifecycle helpers.
3. **Streaming Safety**: Ensure `create_run` fires exactly once per logical session, handle disconnects via resume logic, and decouple event logging from the stream generator using async queues.
4. **Testing & Rollback**: Validate with 12+ acceptance tests covering concurrency, streaming, constraints, and failure modes. Maintain a feature-flagged deployment with backup-driven rollback and data reconciliation scripts.

This approach is safe for a private home lab (minimal infrastructure overhead, SQLite-native optimizations) while meeting engineering-grade standards (idempotency, transactional integrity, concurrency safety, and reversible deployment). Implement the schema migration first, deploy the async transaction wrapper, enable the feature flag, and monitor duplicate rates and SQLite lock metrics. If metrics stabilize, promote to permanent configuration.

# Operational Runbook & Observability

To sustain the fix and prevent regression, implement the following observability and operational procedures:

- **Metrics Collection**: Expose Prometheus-compatible metrics via FastAPI middleware. Track `benchmark_runs_created_total`, `benchmark_runs_duplicated_total`, `sqlite_lock_wait_seconds`, and `streaming_latency_ms`. Alert on `benchmark_runs_duplicated_total > 0` over a 5-minute window.
- **Structured Logging**: Replace print/debug statements with `structlog` or `logging` with JSON formatting. Include `trace_id`, `run_id`, `coro_id`, and `db_changes` in every lifecycle event. Route logs to a lightweight aggregator (e.g., Loki or local JSON files) for post-incident forensics.
- **Health Checks**: Implement `/health` (liveness) and `/ready` (readiness) endpoints. The readiness probe should verify SQLite WAL checkpoint status and connection pool availability before accepting traffic.
- **Automated Vacuum & Integrity Checks**: Schedule a cron job or background task to run `PRAGMA integrity_check;` and `VACUUM;` during low-traffic windows. Monitor `PRAGMA page_count;` and `PRAGMA freelist_count;` to detect bloat.

# Advanced SQLite Tuning & Maintenance

For a private home lab running concurrent workloads, SQLite requires explicit tuning to match relational database performance characteristics:

- **WAL Checkpointing**: Configure `PRAGMA wal_autocheckpoint = 1000;` to balance write throughput and reader visibility. Monitor `PRAGMA wal_checkpoint(PASSIVE);` output to ensure checkpoints complete without blocking writers.
- **Memory-Mapped I/O**: Enable `PRAGMA mmap_size = 268435456;` (256MB) to allow SQLite to cache frequently accessed pages in OS memory, reducing disk I/O for metadata queries.
- **Connection Pooling Strategy**: Use `aiosqlite` with a bounded semaphore (`asyncio.Semaphore(10)`) to limit concurrent DB connections. SQLite handles concurrent reads well but serializes writes; a semaphore prevents thread starvation and `SQLITE_BUSY` cascades.
- **Backup Automation**: Implement point-in-time backups using `sqlite3 benchmark.db ".backup /backups/benchmark_$(date +%Y%m%d_%H%M%S).db"`. Store backups on a separate disk or network mount. Test restore procedures monthly.

# CI/CD & Schema Validation Pipeline

Integrate database safety checks into the deployment workflow to catch constraint violations before they reach production:

- **Schema Drift Detection**: Use `sqlalchemy` or `alembic` to generate a canonical schema hash. Compare against the deployed database on startup. Fail fast if constraints are missing.
- **Idempotency Contract Tests**: Run a dedicated test suite that simulates proxy retries, concurrent SSE connections, and forced network partitions. Assert that `INSERT OR IGNORE` behaves predictably and state machines reject invalid transitions.
- **SQLite Version Pinning**: Lock the SQLite library version in `requirements.txt` or `pyproject.toml`. Home-lab OS updates can silently upgrade SQLite, altering locking behavior or pragma defaults. Use `sqlite3.sqlite_version` to log the active version at startup.
- **Pre-Deployment Smoke Test**: Execute a lightweight migration script that verifies `PRAGMA foreign_keys = ON`, `PRAGMA journal_mode = WAL`, and index existence. Block deployment if any pragma returns a non-compliant value.

# Incident Post-Mortem Template

Document the resolution using the following structure for future reference:

- **Timeline**: Request spike → Duplicate detection → Schema audit → Patch deployment → Validation.
- **Impact**: X duplicate rows, Y failed benchmarks, Z minutes of degraded proxy throughput.
- **Root Cause**: Missing UNIQUE constraint + async race condition in `create_run` + streaming lifecycle misalignment.
- **Fix**: Defense-in-depth idempotency, WAL tuning, async transaction wrappers, feature-flagged rollout.
- **Action Items**: Add constraint validation to CI/CD pipeline, enforce idempotency headers in proxy config, schedule quarterly concurrency load tests.

This completes the engineering response. All sections are aligned with home-lab constraints while maintaining production-grade reliability standards.