# 💾 _inference_context Persistence
**Issue:** [P3.3](#)
**Prioridad:** Media
**Estimación:** 10-12 horas
---
## 🤔 ¿Qué hago? ¿Cómo lo hago? ¿Y para qué lo hago?
La clase `_InferenceContext` almacena estado de request en RAM. Si el proxy crashea, se pierden detalles críticos para debugging.
**What:** Persistir `_InferenceContext` en SQLite durante el ciclo de vida del request.
**How:**
1. Al crear context: guardar en SQLite con estado inicial
2. Actualizar en SQLite después de cada hito (before API call, after response, etc.)
3. Exponer query endpoint `/contexts/{context_id}` para inspection
4. Auto-cleanup de contexts >30 días
**Why:** Debugging de fallos distribuidos requiere trazabilidad completa; actualmente imposible si proxy reinicia.
---
## 🏗️ Schema — Tabla `inference_contexts`
```sql
CREATE TABLE inference_contexts (
id TEXT PRIMARY KEY, -- uuid v4
timestamp_created DATETIME,
timestamp_updated DATETIME,
status TEXT, -- "active", "completed", "failed"
-- Request metadata
model TEXT,
temperature REAL,
max_tokens INTEGER,
-- Lifecycle state
cache_status TEXT, -- "CACHE-HIT", "MISS", "KNOWLEDGE-RAG", etc.
cache_latency_ms FLOAT,
qdrant_cache_state TEXT, -- "open", "closed", "half_open"
qdrant_knowledge_state TEXT,
anthropic_request_sent DATETIME,
anthropic_latency_ms FLOAT,
anthropic_status_code INTEGER,
total_input_tokens INTEGER,
total_output_tokens INTEGER,
estimated_cost_usd REAL,
-- Tracing
trace_log TEXT, -- JSON array de eventos
-- Cleanup
ttl_days INTEGER DEFAULT 30,
deleted_at DATETIME
);
CREATE INDEX idx_timestamp_created ON inference_contexts(timestamp_created);
CREATE INDEX idx_status ON inference_contexts(status);
```
---
## 📝 Event Trace Log (JSON)
```json
{
"context_id": "550e8400-e29b-41d4-a716-446655440000",
"events": [
{
"timestamp": "2026-09-09T12:34:56.001Z",
"type": "CONTEXT_CREATED",
"data": {
"model": "claude-haiku-4-5-20251001",
"temperature": 0.7,
"max_tokens": 100
}
},
{
"timestamp": "2026-09-09T12:34:56.050Z",
"type": "CACHE_CHECK_START",
"data": {}
},
{
"timestamp": "2026-09-09T12:34:56.080Z",
"type": "CACHE_CHECK_COMPLETE",
"data": {
"status": "HIT",
"latency_ms": 30,
"score": 0.98
}
},
{
"timestamp": "2026-09-09T12:34:56.090Z",
"type": "RESPONSE_CACHED",
"data": {
"reason": "semantic cache above threshold"
}
},
{
"timestamp": "2026-09-09T12:34:56.095Z",
"type": "CONTEXT_COMPLETED",
"data": {
"total_latency_ms": 95,
"cost_usd": 0.0
}
}
]
}
```
---
## 🔌 API Endpoints
### GET /contexts/{context_id}
```bash
curl http://localhost:8000/contexts/550e8400-e29b-41d4-a716-446655440000
```
**Respuesta:**
```json
{
"id": "550e8400-e29b-41d4-a716-446655440000",
"status": "completed",
"model": "claude-haiku-4-5-20251001",
"cache_status": "CACHE-HIT",
"cache_latency_ms": 30,
"total_latency_ms": 95,
"estimated_cost_usd": 0.0,
"trace": [
{...}
]
}
```
### GET /contexts?since=2h&limit=100
```bash
curl "http://localhost:8000/contexts?since=2h&limit=100"
```
**Respuesta:**
```json
{
"total": 1500,
"returned": 100,
"contexts": [
{...},
{...}
]
}
```
---
## 🧪 Test Cases
| Caso | Escenario | Expected | Status |
| --- | --- | --- | --- |
| TC1 | Context creado | En SQLite con status=active | ✅ |
| TC2 | Cache HIT registrado | trace.type=CACHE_CHECK_COMPLETE | ✅ |
| TC3 | Anthropic call registrada | timestamp_sent, latency, tokens | ✅ |
| TC4 | Context completado | status=completed, timestamp_updated | ✅ |
| TC5 | GET /contexts/{id} | Recupera traza completa | ✅ |
| TC6 | GET /contexts?since=1h | Filtra por fecha | ✅ |
| TC7 | Auto-cleanup >30 días | Elimina contextos viejos | ✅ |
| TC8 | Concurrencia: 100 requests | Todos registrados sin race | ✅ |
| TC9 | Crash/restart | Contextos incompletos aún queryables | ✅ |
| TC10 | Context con error | status=failed, traza con error | ✅ |
| TC11 | Tamaño DB <500MB | Limpieza automática efectiva | ✅ |
| TC12 | Exportar contextos | CSV/JSON export endpoint | ✅ |
| TC13 | Large trace (10kb) | Compresión o truncation | ✅ |
| TC14 | PII en context | Pseudonymization respetada | ✅ |
| TC15 | Query by model | GET /contexts?model=claude-opus | ✅ |
| TC16 | Query by cost | GET /contexts?cost_gt=0.1 | ✅ |
| TC17 | Pagination | offset/limit | ✅ |
| TC18 | TTL configurable | ENV var CONTEXT_TTL_DAYS | ✅ |
| TC19 | Indexing performance | Queries <100ms | ✅ |
| TC20 | Concurrent exports | Múltiples exports simultáneos | ✅ |
---
## 📊 Implementación
### 1. Tabla de persistencia
```python
# proxy/inference_context.py
class InferenceContextStore:
"""Persistencia de contextos en SQLite."""
async def save_context(self, context: _InferenceContext):
"""Guardar context en DB."""
async with self.db.connection() as conn:
await conn.execute("""
INSERT INTO inference_contexts (id, timestamp_created, status, ...)
VALUES (?, ?, ?, ...)
""", (context.id, datetime.utcnow(), "active", ...))
async def update_event(self, context_id: str, event: dict):
"""Agregar evento a trace."""
trace = await self.get_trace(context_id)
trace["events"].append(event)
await conn.execute("""
UPDATE inference_contexts SET trace_log = ?
WHERE id = ?
""", (json.dumps(trace), context_id))
async def get_context(self, context_id: str) -> dict:
"""Recuperar context completo."""
row = await conn.fetch_one(...)
return {
"id": row["id"],
"status": row["status"],
"trace": json.loads(row["trace_log"]),
...
}
```
### 2. Integración en main.py
```python
@app.post("/v1/messages")
async def messages(request: Request) -> StreamingResponse:
ctx = _InferenceContext(...)
# Guardar contexto inicial
await context_store.save_context(ctx)
try:
# Cache check
await context_store.update_event(ctx.id, {
"type": "CACHE_CHECK_START",
"timestamp": datetime.utcnow().isoformat()
})
cache_result = await cache.search(...)
# ... más eventos
await context_store.update_event(ctx.id, {
"type": "CONTEXT_COMPLETED",
"timestamp": datetime.utcnow().isoformat(),
"total_latency_ms": ctx.latency_ms
})
except Exception as e:
await context_store.update_status(ctx.id, "failed")
raise
```
---
## 🧹 Auto-Cleanup
```python
async def cleanup_old_contexts():
"""Limpiar contextos >TTL_DAYS."""
ttl_days = int(os.getenv("CONTEXT_TTL_DAYS", "30"))
cutoff = datetime.utcnow() - timedelta(days=ttl_days)
async with self.db.connection() as conn:
await conn.execute("""
UPDATE inference_contexts SET deleted_at = ?
WHERE timestamp_created < ? AND deleted_at IS NULL
""", (datetime.utcnow(), cutoff))
# Vacuum DB si >500MB
db_size = os.path.getsize("analytics.db") / (1024**2)
if db_size > 500:
await conn.execute("VACUUM")
```
---
## ✅ Criterios de Aceptación
- ✅ Context guardado en SQLite durante request
- ✅ Queryable via `/contexts/{id}`
- ✅ Traza completa con timestamps y eventos
- ✅ Auto-cleanup >30 días (configurable)
- ✅ Queries <100ms
- ✅ Tests: 20/20 pass
- ✅ DB size <500MB con auto-cleanup
---
## 📝 Próximos Pasos
1. Crear rama: `GH-XXX-inference-context-persistence`
2. Crear tabla SQLite
3. Implementar InferenceContextStore
4. Integrar en main.py
5. Escribir 20 tests
6. Crear endpoints `/contexts/{id}` y `/contexts`
7. PR → merge
8. Deploy y monitorear DB size en staging