**Engineering Response – Duplicate Benchmark Rows in FastAPI + SQLite Service**

Below is a full‑blown engineering investigation, evidence‑gathering plan, patch strategy, database schema changes, streaming‑specific edge‑case handling, a comprehensive test suite, rollback strategy, and a concise decision summary.  
The goal is to eliminate duplicate rows while keeping the solution lightweight enough for a private home‑lab environment, yet robust enough for production‑grade reliability.

---

## # Root Cause Hypotheses

| # | Hypothesis | Why it’s plausible | Typical symptoms |
|---|-------------|--------------------|------------------|
| 1 | **Race condition on `create_run`** | Multiple concurrent requests hit `create_run` before the first commit, each inserting a row with the same `run_id` or auto‑incremented `id`. | Duplicate rows with identical `run_id` or `metadata` but different timestamps. |
| 2 | **Missing unique constraint on `run_id`** | `run_id` is generated client‑side or via a non‑unique sequence; SQLite allows duplicates if no constraint. | Two rows with same `run_id` but different `id`. |
| 3 | **SQLite WAL mode mis‑configured** | Without WAL, concurrent writes may serialize poorly, leading to phantom reads or duplicate inserts. | Intermittent duplicates that disappear when the DB is restarted. |
| 4 | **Improper transaction boundaries** | `create_run` and `finish_llm_request` are called in separate transactions; a crash between them leaves an incomplete run that is later retried. | A “half‑finished” run that is later duplicated. |
| 5 | **Proxy‑level request duplication** | The reverse‑proxy (e.g., Nginx) forwards the same request twice due to retry logic or keep‑alive mis‑config. | Duplicate rows with identical request payloads. |
| 6 | **Streaming requests not committing until stream ends** | Streaming responses may keep the DB connection open; if the client disconnects early, the transaction may be rolled back or left open, causing a retry that inserts again. | Duplicate rows where one has `status='completed'` and another `status='aborted'`. |
| 7 | **Concurrent `add_event` calls** | Multiple events for the same run are inserted concurrently without a lock, leading to duplicate event rows that are mistakenly interpreted as duplicate runs. | Duplicate event rows with same `run_id` and `event_type`. |
| 8 | **Python-level idempotency key missing** | The service does not enforce idempotency on the API layer; identical POSTs from the client create new runs each time. | Duplicate runs with identical `metadata`. |
| 9 | **SQLite autoincrement bug** | SQLite’s `INTEGER PRIMARY KEY AUTOINCREMENT` can reuse IDs if the DB is truncated or if a transaction is rolled back. | Duplicate `id` values across runs. |
| 10 | **Agent client retry logic** | The agent client retries on network errors, causing duplicate `create_run` calls. | Duplicate rows with the same `agent_id`. |

> **Top 3 likely culprits**: 1, 2, and 4.  
> These involve race conditions, missing constraints, and transaction mis‑management—common sources of duplicate rows in a concurrent environment.

---

## # Evidence To Collect

| Item | How to collect | What to look for |
|------|----------------|------------------|
| **Request logs** | Enable FastAPI’s `logging` middleware to capture `request_id`, `user_id`, `endpoint`, `method`, `timestamp`. | Duplicate `request_id`s or identical payloads arriving within milliseconds. |
| **Database transaction logs** | Use SQLite’s `PRAGMA journal_mode; PRAGMA wal_autocheckpoint;` and enable `sqlite3` `detect_types` to log each `BEGIN`, `COMMIT`, `ROLLBACK`. | Interleaved `BEGIN`/`COMMIT` pairs that overlap. |
| **Run table snapshots** | Periodically `SELECT * FROM runs ORDER BY id DESC LIMIT 10;` | Duplicate `run_id` or `metadata` entries. |
| **Event table snapshots** | `SELECT * FROM events WHERE run_id = ?;` | Duplicate events for the same run. |
| **Proxy logs** | Nginx/Traefik logs with `proxy_pass` and `proxy_redirect` disabled. | Same `client_ip` sending identical requests twice. |
| **Agent client logs** | Capture retry attempts, back‑off intervals, and request IDs. | Multiple `create_run` calls with same payload. |
| **SQLite WAL files** | Inspect `*.wal` size and content. | Large WAL files indicating many concurrent writes. |
| **Python stack traces** | Wrap `create_run` and `finish_llm_request` in try/except and log stack traces. | Exceptions during commit or rollback. |
| **Concurrency metrics** | Use `asyncio` `Semaphore` counts or `uvicorn` worker counts. | High concurrency spikes correlating with duplicates. |

> **Action**: Add a temporary `/debug/metrics` endpoint that returns the above snapshots for quick inspection.

---

## # Patch Plan

The patch is split into **database‑level** and **Python‑level** changes. The goal is to enforce uniqueness, serialize writes, and provide idempotency.

### 1. Database Schema Enhancements

```sql
-- 1.1 Enable WAL mode (recommended for concurrent writes)
PRAGMA journal_mode = WAL;
PRAGMA synchronous = NORMAL;   -- trade‑off: faster writes, still safe

-- 1.2 Add a unique constraint on run_id (client‑generated or server‑generated)
ALTER TABLE runs ADD COLUMN run_id TEXT NOT NULL;
CREATE UNIQUE INDEX idx_runs_run_id ON runs(run_id);

-- 1.3 Add a composite unique constraint on (user_id, run_name, created_at)
-- to guard against accidental duplicates from the same user
CREATE UNIQUE INDEX idx_runs_user_name_time ON runs(user_id, run_name, created_at);

-- 1.4 Add a status column to track completion
ALTER TABLE runs ADD COLUMN status TEXT NOT NULL DEFAULT 'pending';
-- status values: pending, completed, aborted, error

-- 1.5 Add foreign key constraints for events
ALTER TABLE events ADD CONSTRAINT fk_events_run_id
  FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE;

-- 1.6 Add indexes for fast lookups
CREATE INDEX idx_events_run_id ON events(run_id);
CREATE INDEX idx_events_timestamp ON events(timestamp);
```

> **Why**:  
> - WAL mode allows concurrent readers/writers.  
> - Unique constraints prevent duplicates at the DB level.  
> - Status column helps detect incomplete runs.  
> - Foreign keys enforce referential integrity.

### 2. Python‑Level Guardrails

#### 2.1 Idempotency Key Middleware

```python
from fastapi import Request, Response, HTTPException
from starlette.middleware.base import BaseHTTPMiddleware
import uuid

class IdempotencyMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        # Extract idempotency key from header or generate one
        idem_key = request.headers.get("Idempotency-Key")
        if not idem_key:
            idem_key = str(uuid.uuid4())
        request.state.idempotency_key = idem_key

        # Check if a run with this key already exists
        async with request.app.state.db.acquire() as conn:
            row = await conn.fetchrow(
                "SELECT * FROM runs WHERE idempotency_key = $1",
                idem_key
            )
            if row:
                # Return the existing run without creating a new one
                return Response(
                    content=JSONResponse(content=row).body,
                    status_code=200,
                    media_type="application/json"
                )
        response = await call_next(request)
        return response
```

> **Why**: Guarantees that identical POSTs (same `Idempotency-Key`) are idempotent.

#### 2.2 Transaction Wrapper

```python
from contextlib import asynccontextmanager
import asyncpg

@asynccontextmanager
async def transactional(conn: asyncpg.Connection):
    try:
        await conn.execute("BEGIN")
        yield
        await conn.execute("COMMIT")
    except Exception:
        await conn.execute("ROLLBACK")
        raise
```

> **Usage**:

```python
async def create_run(conn, run_data):
    async with transactional(conn):
        await conn.execute(
            """
            INSERT INTO runs (run_id, user_id, run_name, metadata, status, idempotency_key)
            VALUES ($1, $2, $3, $4, 'pending', $5)
            """,
            run_data.run_id,
            run_data.user_id,
            run_data.run_name,
            json.dumps(run_data.metadata),
            request.state.idempotency_key
        )
```

#### 2.3 Streaming Safe Finish

```python
async def finish_llm_request(conn, run_id, result, status='completed'):
    async with transactional(conn):
        await conn.execute(
            """
            UPDATE runs
            SET result = $1, status = $2
            WHERE run_id = $3
            """,
            json.dumps(result),
            status,
            run_id
        )
        # Optionally delete orphaned events
        await conn.execute(
            "DELETE FROM events WHERE run_id = $1 AND event_type = 'orphaned'",
            run_id
        )
```

#### 2.4 Concurrency Guard

```python
# Use a simple in‑memory lock per run_id
run_locks = {}

def get_run_lock(run_id):
    if run_id not in run_locks:
        run_locks[run_id] = asyncio.Lock()
    return run_locks[run_id]
```

```python
async def add_event(conn, run_id, event_type, payload):
    lock = get_run_lock(run_id)
    async with lock:
        await conn.execute(
            """
            INSERT INTO events (run_id, event_type, payload, timestamp)
            VALUES ($1, $2, $3, $4)
            """,
            run_id,
            event_type,
            json.dumps(payload),
            datetime.utcnow().isoformat()
        )
```

> **Why**: Prevents concurrent `add_event` calls from interleaving and causing duplicate rows.

#### 2.5 Proxy‑Level Retry Prevention

- Configure the reverse‑proxy to **disable** automatic retries on 5xx responses for the `/runs` endpoint.
- Add a `Retry-After` header on 429 responses to throttle.

#### 2.6 Agent Client Idempotency

- Update the agent client to send an `Idempotency-Key` header derived from the run’s unique identifier (e.g., a hash of the request payload + timestamp).

---

## # SQLite Constraints And Indexes

Below is a consolidated SQL script that can be run once to upgrade the schema. It includes all constraints, indexes, and WAL configuration.

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

-- 1. Add run_id column and unique constraint
ALTER TABLE runs ADD COLUMN run_id TEXT NOT NULL;
CREATE UNIQUE INDEX idx_runs_run_id ON runs(run_id);

-- 2. Add idempotency_key column and unique constraint
ALTER TABLE runs ADD COLUMN idempotency_key TEXT NOT NULL;
CREATE UNIQUE INDEX idx_runs_idempotency ON runs(idempotency_key);

-- 3. Add status column
ALTER TABLE runs ADD COLUMN status TEXT NOT NULL DEFAULT 'pending';

-- 4. Add composite unique constraint to guard against accidental duplicates
CREATE UNIQUE INDEX idx_runs_user_name_time ON runs(user_id, run_name, created_at);

-- 5. Add foreign key to events
ALTER TABLE events ADD CONSTRAINT fk_events_run_id
  FOREIGN KEY (run_id) REFERENCES runs(run_id) ON DELETE CASCADE;

-- 6. Add indexes for performance
CREATE INDEX idx_events_run_id ON events(run_id);
CREATE INDEX idx_events_timestamp ON events(timestamp);
```

> **Tip**: Run `PRAGMA foreign_keys = ON;` at application startup to enforce FK constraints.

---

## # Streaming Edge Cases

Streaming requests introduce several pitfalls that can lead to duplicate or incomplete rows.

| Edge Case | Problem | Mitigation |
|-----------|---------|------------|
| **Client disconnects mid‑stream** | The DB transaction may be left open; the client may retry, causing duplicate `create_run`. | Use `async with transactional(conn):` and set a timeout. On disconnect, the transaction rolls back automatically. |
| **Partial writes to `events`** | If `add_event` is called after the stream ends but before `finish_llm_request`, a duplicate event may be inserted. | Acquire a per‑run lock around `add_event` and `finish_llm_request`. |
| **Streaming with multiple chunks** | Each chunk may trigger a separate `add_event`. If the same chunk is retried, duplicate events occur. | Include a `chunk_id` in the event payload and enforce uniqueness via a composite index: `CREATE UNIQUE INDEX idx_events_chunk ON events(run_id, chunk_id);` |
| **Back‑pressure** | If the event queue is full, the stream may block, leading to duplicate `create_run` on retry. | Use an async queue with bounded size; on overflow, return a 429 and let the client back‑off. |
| **Race between `finish_llm_request` and `add_event`** | `finish_llm_request` may commit before the last event is inserted, leaving the run incomplete. | Wrap both calls in the same transaction or use a `status` flag that is only set to `completed` after all events are flushed. |
| **Duplicate `finish_llm_request`** | If the client retries the finish endpoint, the run may be updated twice. | Idempotent finish: check if `status='completed'` before updating. |

### Streaming‑Safe Flow

```python
async def handle_stream(request):
    run_id = generate_run_id()
    idem_key = request.headers.get("Idempotency-Key", str(uuid.uuid4()))

    async with request.app.state.db.acquire() as conn:
        # 1. Create run (idempotent)
        await create_run(conn, run_id, request.user_id, request.body, idem_key)

        # 2. Stream response
        async for chunk in stream_generator():
            # 3. Add event per chunk
            await add_event(conn, run_id, "chunk", {"data": chunk, "chunk_id": uuid.uuid4()})

        # 4. Finish run
        await finish_llm_request(conn, run_id, result, status='completed')
```

All DB operations are wrapped in `transactional(conn)` to guarantee atomicity.

---

## # Test Plan

Below is a **12‑test** acceptance suite covering concurrency, idempotency, streaming, and rollback scenarios. Each test is written in pseudo‑Python using `pytest` and `httpx` async client.

| # | Test Name | Purpose | Key Assertions |
|---|-----------|---------|----------------|
| 1 | `test_single_run_creation` | Basic run creation | `runs` count increases by 1; `run_id` unique |
| 2 | `test_duplicate_run_id_prevention` | Unique constraint on `run_id` | Duplicate insert raises `IntegrityError` |
| 3 | `test_idempotency_key_prevention` | Idempotent POST | Same `Idempotency-Key` returns same run, no new row |
| 4 | `test_concurrent_run_creation` | Race condition on `create_run` | 10 concurrent POSTs with same `Idempotency-Key` → 1 row |
| 5 | `test_concurrent_event_insertion` | Race condition on `add_event` | 20 concurrent events for same run → 20 rows, no duplicates |
| 6 | `test_streaming_completion` | Streaming flow completes correctly | After stream, `status='completed'` and all events present |
| 7 | `test_streaming_disconnect` | Client disconnect mid‑stream | No duplicate run; transaction rolled back |
| 8 | `test_streaming_retry` | Client retries after disconnect | Idempotent finish; no duplicate run |
| 9 | `test_proxy_retry_prevention` | Proxy does not retry 5xx | Simulate 5xx; ensure only one run created |
| 10 | `test_agent_client_idempotency` | Agent client uses idempotency key | Duplicate agent requests → same run |
| 11 | `test_wal_mode_concurrency` | WAL mode handles concurrent writes | 50 concurrent writes → all persisted |
| 12 | `test_rollback_on_error` | Transaction rollback on error | Simulate exception during `add_event`; no partial data |

### Sample Test Code

```python
import pytest
import httpx
import asyncio
from fastapi import FastAPI
from app.main import app  # your FastAPI app

@pytest.fixture
async def client():
    async with httpx.AsyncClient(app=app, base_url="http://test") as c:
        yield c

@pytest.mark.asyncio
async def test_idempotency_key_prevention(client):
    payload = {"run_name": "test_run", "metadata": {"foo": "bar"}}
    headers = {"Idempotency-Key": "abc123"}
    resp1 = await client.post("/runs", json=payload, headers=headers)
    assert resp1.status_code == 200
    run_id = resp1.json()["run_id"]

    resp2 = await client.post("/runs", json=payload, headers=headers)
    assert resp2.status_code == 200
    assert resp2.json()["run_id"] == run_id

    # Verify only one row in DB
    async with app.state.db.acquire() as conn:
        rows = await conn.fetch("SELECT COUNT(*) FROM runs WHERE run_id=$1", run_id)
        assert rows[0]["count"] == 1
```

> **Tip**: Use `pytest-asyncio` and `httpx.AsyncClient` for async integration tests.

---

## # Rollback Plan

1. **Backup the database** before applying schema changes:
   ```bash
   cp production.db production.db.bak
   ```

2. **Revert schema changes**:
   - Drop indexes and constraints added in the patch.
   - Remove `run_id`, `idempotency_key`, `status` columns if they cause issues.
   - Restore original `runs` table definition.

3. **Disable WAL mode** if it caused problems:
   ```sql
   PRAGMA journal_mode = DELETE;
   ```

4. **Restore application code**:
   - Remove `IdempotencyMiddleware`.
   - Revert transaction wrappers to simple `conn.execute`.
   - Remove per‑run locks.

5. **Re‑run integration tests** to confirm the system is back to baseline.

6. **Deploy** the rollback to staging first, then to production.

> **Rollback safety**: All changes are additive; dropping constraints or indexes is safe. The only destructive change is the removal of columns, which can be done with `ALTER TABLE ... DROP COLUMN` (SQLite 3.35+). If using an older SQLite, copy the table without the new columns.

---

## # Decision Summary

| Decision | Rationale | Impact |
|----------|-----------|--------|
| **Enable WAL mode** | Improves concurrent write performance and reduces write conflicts. | Minimal overhead; safe for home lab. |
| **Add unique constraints on `run_id` and `idempotency_key`** | Prevents duplicate rows at DB level. | Zero risk of duplicates; small schema change. |
| **Add `status` column** | Tracks run lifecycle, helps detect incomplete runs. | Enables graceful recovery. |
| **Implement Idempotency middleware** | Guarantees that identical POSTs are idempotent. | Eliminates duplicate runs from client retries. |
| **Wrap DB ops in explicit transactions** | Guarantees atomicity across create/finish. | Prevents half‑written runs. |
| **Add per‑run async locks** | Serializes concurrent `add_event` calls. | Avoids duplicate events. |
| **Add composite index on `events(run_id, chunk_id)`** | Prevents duplicate chunk events. | Improves event lookup. |
| **Configure proxy to disable retries** | Stops duplicate requests from the reverse‑proxy. | Requires minimal config change. |
| **Update agent client to send idempotency key** | Aligns client behavior with server expectations. | Requires client update but trivial. |
| **Add comprehensive test suite** | Validates all edge cases. | Ensures regression safety. |
| **Rollback plan** | Provides safety net. | Low effort to revert. |

**Conclusion**:  
The combination of database constraints, idempotency enforcement, transaction safety, and concurrency control will eliminate duplicate benchmark rows while keeping the system lightweight for a private home‑lab environment. The changes are additive, reversible, and fully testable. Deploy the patch to a staging environment first, run the 12‑test suite, and then roll out to production.