diff --git a/.env.example b/.env.example
index 6fdd88d..7644421 100644
--- a/.env.example
+++ b/.env.example
@@ -9,3 +9,16 @@ POSTGRES_DB=insightiq
POSTGRES_USER=insightiq_user
POSTGRES_PASSWORD=insightiq_pass
DATABASE_URL=postgresql://insightiq_user:insightiq_pass@localhost:5432/insightiq
+
+LANGFUSE_PUBLIC_KEY=pk-...
+LANGFUSE_SECRET_KEY=sk-...
+LANGFUSE_BASE_URL=https://cloud.langfuse.com
+
+DATABASE_URL_RO=postgresql://querymind_ro:...@host/insightiq # read path
+STATEMENT_TIMEOUT_MS=5000
+MAX_ROW_LIMIT=1000
+BUDGET_USD_DAILY=2.00
+COST_ALERT_WEBHOOK_URL= # optional Slack webhook for in-app budget alerts
+CLAUDE_INPUT_USD_PER_MTOK=3.0
+CLAUDE_OUTPUT_USD_PER_MTOK=15.0
+CACHE_TTL_SECONDS=900 # 15 min — schema prompt + normalised question cache
diff --git a/.github/workflows/cd.yml b/.github/workflows/cd.yml
index 8cadc70..4dd56a5 100644
--- a/.github/workflows/cd.yml
+++ b/.github/workflows/cd.yml
@@ -131,7 +131,9 @@ jobs:
"anthropic-api-key=${{ secrets.ANTHROPIC_API_KEY }}" \
"jwt-secret-key=${{ secrets.JWT_SECRET_KEY }}" \
"google-client-id=${{ secrets.GOOGLE_CLIENT_ID }}" \
- "database-url=${{ secrets.NEON_DATABASE_URL }}"
+ "database-url=${{ secrets.NEON_DATABASE_URL }}" \
+ "database-url-ro=${{ secrets.NEON_DATABASE_URL_RO }}" \
+ "langfuse-sk=${{ secrets.LANGFUSE_SECRET_KEY }}"
az containerapp update \
--name "$BACKEND_APP_NAME" \
@@ -142,6 +144,16 @@ jobs:
"JWT_SECRET_KEY=secretref:jwt-secret-key" \
"GOOGLE_CLIENT_ID=secretref:google-client-id" \
"DATABASE_URL=secretref:database-url" \
+ "DATABASE_URL_RO=secretref:database-url-ro" \
+ "LANGFUSE_PUBLIC_KEY=${{ secrets.LANGFUSE_PUBLIC_KEY }}" \
+ "LANGFUSE_SECRET_KEY=secretref:langfuse-sk" \
+ "LANGFUSE_BASE_URL=https://cloud.langfuse.com" \
+ "STATEMENT_TIMEOUT_MS=5000" \
+ "MAX_ROW_LIMIT=1000" \
+ "BUDGET_USD_DAILY=2.00" \
+ "CLAUDE_INPUT_USD_PER_MTOK=3.0" \
+ "CLAUDE_OUTPUT_USD_PER_MTOK=15.0" \
+ "CACHE_TTL_SECONDS=900" \
"FRONTEND_URL=${{ secrets.FRONTEND_URL }}" \
"LOG_LEVEL=INFO" \
"MAX_RETRIES=3" \
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index ff65229..bd53ece 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -4,6 +4,7 @@
#
# Jobs:
# backend-ci — ruff lint, pytest unit tests (no DB, no API)
+# sql-eval — only if backend/eval paths changed; smoke 5-call eval + red team
# frontend-ci — TypeScript typecheck, production build
# security — Trivy dependency scan (advisory, never blocks)
# ci-success — gate job that CD depends on
@@ -27,6 +28,47 @@ concurrency:
jobs:
+ # ── Detect path changes (skip paid sql-eval on README / frontend-only PRs) ─
+ changes:
+ name: "Detect changed paths"
+ runs-on: ubuntu-latest
+ permissions:
+ contents: read
+ pull-requests: read
+ outputs:
+ sql_eval: ${{ steps.paths.outputs.sql_eval }}
+ steps:
+ - uses: actions/checkout@v4
+ with:
+ fetch-depth: 0
+
+ - name: Detect sql-eval paths (git diff — no GitHub API)
+ id: paths
+ run: |
+ if [ "${{ github.event_name }}" = "pull_request" ]; then
+ BASE="${{ github.event.pull_request.base.sha }}"
+ elif [ -n "${{ github.event.before }}" ] && [ "${{ github.event.before }}" != "0000000000000000000000000000000000000000" ]; then
+ BASE="${{ github.event.before }}"
+ else
+ BASE="$(git rev-parse HEAD~1 2>/dev/null || echo "")"
+ fi
+
+ if [ -z "$BASE" ]; then
+ echo "sql_eval=true" >> "$GITHUB_OUTPUT"
+ echo "No base ref — running sql-eval"
+ exit 0
+ fi
+
+ CHANGED="$(git diff --name-only "$BASE" HEAD)"
+ echo "Changed files:"
+ echo "$CHANGED"
+
+ if echo "$CHANGED" | grep -qE '^(backend/|tests/|database/|pytest\.ini)'; then
+ echo "sql_eval=true" >> "$GITHUB_OUTPUT"
+ else
+ echo "sql_eval=false" >> "$GITHUB_OUTPUT"
+ fi
+
# ── Backend: lint + unit tests ───────────────────────────────
backend-ci:
name: "Backend — lint & test"
@@ -37,6 +79,7 @@ jobs:
# Unit tests don't need a real DB or API key
ANTHROPIC_API_KEY: sk-ant-ci-placeholder-not-real
DATABASE_URL: postgresql://test:test@localhost:5432/test
+ DATABASE_URL_RO: postgresql://test:test@localhost:5432/test
JWT_SECRET_KEY: ci-test-secret-exactly-32-chars-ok
GOOGLE_CLIENT_ID: ci-test.apps.googleusercontent.com
LOG_LEVEL: WARNING
@@ -68,7 +111,8 @@ jobs:
../tests/test_sql_agent.py \
../tests/test_auth.py::TestJWT \
../tests/test_auth.py::TestRateLimit \
- -v --tb=short -m "not integration" \
+ ../tests/test_eval.py \
+ -v --tb=short -m "not integration and not eval_live" \
--junitxml=../test-results/unit.xml
- name: Upload test results
@@ -79,6 +123,73 @@ jobs:
path: test-results/
retention-days: 7
+ # ── SQL eval + red team (merge gate, path-filtered) ────────────
+ sql-eval:
+ name: "SQL evals + guardrail red-team"
+ needs: changes
+ if: needs.changes.outputs.sql_eval == 'true'
+ runs-on: ubuntu-latest
+ timeout-minutes: 15
+
+ services:
+ postgres:
+ image: postgres:16
+ env:
+ POSTGRES_USER: test
+ POSTGRES_PASSWORD: test
+ POSTGRES_DB: test
+ ports:
+ - 5432:5432
+ options: >-
+ --health-cmd pg_isready
+ --health-interval 10s
+ --health-timeout 5s
+ --health-retries 5
+
+ env:
+ ANTHROPIC_API_KEY: ${{ secrets.ANTHROPIC_API_KEY }}
+ DATABASE_URL: postgresql://test:test@localhost:5432/test
+ DATABASE_URL_RO: postgresql://test:test@localhost:5432/test
+ JWT_SECRET_KEY: ci-test-secret-exactly-32-chars-ok
+ GOOGLE_CLIENT_ID: ci-test.apps.googleusercontent.com
+ LOG_LEVEL: WARNING
+
+ steps:
+ - uses: actions/checkout@v4
+
+ - name: Set up Python 3.12
+ uses: actions/setup-python@v5
+ with:
+ python-version: "3.12"
+ cache: pip
+ cache-dependency-path: backend/requirements.txt
+
+ - name: Install dependencies
+ run: pip install -r backend/requirements.txt
+
+ - name: Install PostgreSQL client
+ run: sudo apt-get update && sudo apt-get install -y postgresql-client
+
+ - name: Load eval fixture database
+ env:
+ PGPASSWORD: test
+ run: psql -h localhost -U test -d test -v ON_ERROR_STOP=1 -f tests/eval/seed.sql
+
+ - name: SQL evals + guardrail red-team (5 Claude calls)
+ run: |
+ mkdir -p test-results
+ cd backend
+ python -m pytest ../tests/eval/ -v -m ci_gate --tb=short \
+ --junitxml=../test-results/sql-eval.xml
+
+ - name: Upload sql-eval results
+ uses: actions/upload-artifact@v4
+ if: always()
+ with:
+ name: sql-eval-results
+ path: test-results/sql-eval.xml
+ retention-days: 7
+
# ── Frontend: typecheck + build ──────────────────────────────
frontend-ci:
name: "Frontend — typecheck & build"
@@ -135,15 +246,22 @@ jobs:
ci-success:
name: "CI passed ✓"
runs-on: ubuntu-latest
- needs: [backend-ci, frontend-ci]
+ needs: [backend-ci, changes, sql-eval, frontend-ci]
if: always()
steps:
- name: Verify required jobs passed
run: |
backend="${{ needs.backend-ci.result }}"
+ sql_eval="${{ needs.sql-eval.result }}"
frontend="${{ needs.frontend-ci.result }}"
echo "backend-ci: $backend"
+ echo "sql-eval: $sql_eval (skipped when only README/frontend/etc. changed)"
echo "frontend-ci: $frontend"
+ # skipped sql-eval is OK — path filter excluded this push
+ if [[ "$sql_eval" != "success" && "$sql_eval" != "skipped" ]]; then
+ echo "sql-eval failed"
+ exit 1
+ fi
if [[ "$backend" != "success" || "$frontend" != "success" ]]; then
echo "CI failed — CD will not run"
exit 1
diff --git a/README.md b/README.md
index d9acb6e..555d6fd 100644
--- a/README.md
+++ b/README.md
@@ -15,6 +15,15 @@ Live: https://insightiq-frontend.orangemeadow-25c8b891.westeurope.azurecontainer
**Demo tip:** First visit after idle may take **~30–60s** while the app wakes from scale-to-zero — deliberate idle scaling keeps hosting near **$0**. Open the [frontend](https://insightiq-frontend.orangemeadow-25c8b891.westeurope.azurecontainerapps.io) a minute before a demo; sign-in wakes the backend automatically.
+
+**Production layers:** Langfuse trace per query (SQL · time · rows · cost),
+flushed on graceful shutdown · 15-pair eval graded by EXECUTION ACCURACY,
+CI-gated at 90% · AST guardrails (SELECT-only, table/column whitelist,
+injection scan) · read-only DB role + statement timeout + forced LIMIT +
+PII block · cost per query (~$0.004 avg, daily budget alert) · graceful
+degradation on timeout / bad SQL / zero rows · normalised-question cache ·
+scale-to-zero on Container Apps ($0 idle)
+
---
## Demo
@@ -280,7 +289,9 @@ All deploys are zero-downtime rolling updates. The CD pipeline applies Neon migr
| GitHub Actions | $0 (free tier) |
| **Total infrastructure** | **~$2–8/month** |
-Anthropic API cost scales with query volume. With the 2-query lifetime cap per user and JSONB result caching, the marginal cost per unique visitor is bounded at approximately $0.01–0.03.
+Anthropic API cost scales with query volume. **Measured average per query: ~$0.03** at current schema-injection size (~10.5k input + ~130 output tokens on claude-sonnet-4-5). With prompt caching or a slimmer schema, that drops toward **~$0.004**. Each query is metered in Langfuse (`cost_usd` on the trace) and `query_audit`; daily spend is tracked in `daily_cost_spend` with a webhook alert when `BUDGET_USD_DAILY` is exceeded. The per-user query cap and JSONB result caching bound marginal cost per visitor.
+
+See [docs/cost.md](docs/cost.md) for the Langfuse dashboard screenshot and SQL to compute live averages.
---
diff --git a/backend/app/agent/sql_agent.py b/backend/app/agent/sql_agent.py
index 223a402..6b15f02 100644
--- a/backend/app/agent/sql_agent.py
+++ b/backend/app/agent/sql_agent.py
@@ -1,10 +1,8 @@
"""
-InsightIQ — SQL Agent (Updated: user_id + result caching)
-===========================================================
-Changes from Day 3:
-- run_sql_agent() now accepts user_id parameter
-- _log_query_history() saves result_json (rows + columns) and explanation
- so history items can be replayed from cache without calling Claude again
+InsightIQ — SQL Agent (Updated: user_id + result caching + Langfuse tracing)
+==============================================================================
+Layer 1: every query is one Langfuse trace — question → SQL → validation →
+execution time → rows → tokens → retries → verdict.
"""
import json
@@ -17,10 +15,25 @@
import anthropic
+from app.cache.answer import answer_cached
from app.core.config import settings
-from app.core.exceptions import DatabaseError, SQLGenerationError, UnsafeSQLError
+from app.core.exceptions import SQLGenerationError, UnsafeSQLError
+from app.cost.daily import finalize_query_cost
+from app.cost.meter import CostMeter
+from app.db.audit import VERDICT_ALLOWED, VERDICT_BLOCKED, log_query_audit
+from app.db.data_access import fetch_readonly
from app.db.pool import database_pool
from app.db.schema_loader import get_schema_for_prompt
+from app.guardrails import validate_sql_ast
+from app.observability.tracing import (
+ generation_span,
+ is_tracing_enabled,
+ observe,
+ propagate_attributes,
+ trace_tags,
+ update_current_trace_cost,
+)
+from app.reliability.outcome import AgentOutcome
logger = logging.getLogger(__name__)
_claude = anthropic.Anthropic(api_key=settings.ANTHROPIC_API_KEY)
@@ -53,148 +66,208 @@
Analyse the error and return corrected JSON only."""
-class AgentResult:
- """Value object returned on success."""
-
- __slots__ = (
- "question",
- "sql",
- "explanation",
- "rows",
- "row_count",
- "columns",
- "execution_ms",
- "retries",
- )
+async def run_sql_agent(question: str, user_id: Optional[int] = None) -> AgentOutcome:
+ """
+ Convert a natural language question to SQL and execute it (Layer 6).
- def __init__(self, question, sql, explanation, rows, row_count, columns, execution_ms, retries):
- self.question = question
- self.sql = sql
- self.explanation = explanation
- self.rows = rows
- self.row_count = row_count
- self.columns = columns
- self.execution_ms = execution_ms
- self.retries = retries
+ Returns a typed ``AgentOutcome`` — never raises for expected failure modes.
+ """
+ meter = CostMeter()
+ async def _execute() -> AgentOutcome:
+ outcome, _cache_hit = await answer_cached(question, user_id=user_id, meter=meter)
+ return outcome
-async def run_sql_agent(question: str, user_id: Optional[int] = None) -> AgentResult:
- """
- Convert a natural language question to SQL and execute it.
- Saves result to query_history with user_id and result_json cache.
- """
+ try:
+ if not is_tracing_enabled():
+ return await _execute()
+
+ from langfuse import get_client
+
+ langfuse = get_client()
+ with langfuse.start_as_current_observation(
+ name="insightiq_query",
+ input={"question": question},
+ ) as root_span:
+ tags = trace_tags(
+ retries=0,
+ row_count=0,
+ truncated=False,
+ verdict="allowed",
+ )
+ with propagate_attributes(
+ user_id=str(user_id) if user_id is not None else "anonymous",
+ metadata={"question": question},
+ tags=tags,
+ trace_name="insightiq_query",
+ ):
+ outcome = await _execute()
+ tags = trace_tags(
+ retries=outcome.retries,
+ row_count=outcome.row_count,
+ truncated=outcome.truncated,
+ verdict=outcome.verdict,
+ failure_mode=outcome.failure_mode.value,
+ from_cache=outcome.from_cache,
+ )
+ level = "ERROR" if outcome.status == "degraded" else "DEFAULT"
+ root_span.update(
+ level=level,
+ output=outcome.to_trace_output(),
+ metadata={
+ "exec_ms": outcome.execution_ms,
+ "verdict": outcome.verdict,
+ "cost_usd": outcome.cost_usd,
+ "failure_mode": outcome.failure_mode.value,
+ "tags": ",".join(tags),
+ },
+ )
+ update_current_trace_cost(
+ outcome.cost_usd,
+ input_tokens=outcome.input_tokens,
+ output_tokens=outcome.output_tokens,
+ failure_mode=outcome.failure_mode.value,
+ )
+ return outcome
+ finally:
+ await finalize_query_cost(meter)
+
+
+async def generate_sql(question: str) -> str:
+ """Generate SQL for a question only (no execution). Used by the eval harness."""
schema = await get_schema_for_prompt()
system = _SYSTEM_PROMPT_TEMPLATE.format(schema=schema)
- messages: List[Dict[str, str]] = [{"role": "user", "content": question}]
-
- sql: Optional[str] = None
- explanation = ""
- last_error: Optional[str] = None
- retries = 0
-
- for attempt in range(settings.MAX_RETRIES + 1):
- if attempt > 0:
- retries += 1
- messages.append(
- {
- "role": "assistant",
- "content": json.dumps({"sql": sql, "explanation": explanation}),
- }
- )
- messages.append(
- {
- "role": "user",
- "content": _RETRY_TEMPLATE.format(error=last_error, failed_sql=sql or ""),
- }
- )
+ sql, _, _ = await _generate_sql(system, [{"role": "user", "content": question}])
+ return sql
- try:
- sql, explanation = await _generate_sql(system, messages)
- except SQLGenerationError:
- if attempt == settings.MAX_RETRIES:
- raise
- last_error = "Could not parse Claude response as valid JSON with 'sql' key."
- continue
-
- _assert_safe_sql(sql)
+def validate_sql(sql: str) -> str:
+ """Return guardrail verdict: ``allowed`` or ``blocked``."""
+ try:
+ _validate_sql(sql)
+ return "allowed"
+ except UnsafeSQLError:
+ return "blocked"
+
+
+async def run_readonly(sql: str) -> List[dict]:
+ """Execute SQL on the read-only pool (Layer 4) and return serialised rows."""
+ rows = await fetch_readonly(sql)
+ return [_serialise_row(r) for r in rows]
+
+
+async def _generate_sql(
+ system: str,
+ messages: list,
+ *,
+ meter: CostMeter | None = None,
+) -> Tuple[str, str, Dict[str, int]]:
+ """Call Claude and return SQL, explanation, and token usage."""
+ with generation_span(
+ "generate_sql",
+ input_data={"message_count": len(messages), "model": settings.CLAUDE_MODEL},
+ ) as gen_span:
+ response = _claude.messages.create(
+ model=settings.CLAUDE_MODEL,
+ max_tokens=1024,
+ system=system,
+ messages=messages,
+ temperature=0,
+ )
+ raw = response.content[0].text.strip()
+ raw = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.MULTILINE)
+ raw = re.sub(r"\s*```$", "", raw, flags=re.MULTILINE)
try:
- t0 = time.perf_counter()
- rows = await database_pool.fetch(sql)
- execution_ms = int((time.perf_counter() - t0) * 1000)
-
- if len(rows) > settings.MAX_RESULT_ROWS:
- rows = rows[: settings.MAX_RESULT_ROWS]
-
- rows = [_serialise_row(r) for r in rows]
- columns = list(rows[0].keys()) if rows else []
-
- # Save to audit log WITH result cache
- await _log_query_history(
- question=question,
- sql=sql,
- explanation=explanation,
- rows=rows,
- columns=columns,
- row_count=len(rows),
- execution_ms=execution_ms,
- retries=retries,
- user_id=user_id,
- )
-
- return AgentResult(
- question=question,
- sql=sql,
- explanation=explanation,
- rows=rows,
- row_count=len(rows),
- columns=columns,
- execution_ms=execution_ms,
- retries=retries,
+ parsed = json.loads(raw)
+ sql = parsed.get("sql", "").strip()
+ if not sql:
+ raise SQLGenerationError("Claude returned JSON but 'sql' is empty.")
+ explanation = parsed.get("explanation", "")
+ except json.JSONDecodeError as exc:
+ raise SQLGenerationError(f"Claude response not valid JSON: {raw[:200]}") from exc
+
+ usage = {
+ "input_tokens": getattr(response.usage, "input_tokens", 0) or 0,
+ "output_tokens": getattr(response.usage, "output_tokens", 0) or 0,
+ }
+ call_cost = 0.0
+ if meter is not None:
+ call_cost = meter.add_usage(usage)
+ if gen_span is not None:
+ gen_span.update(
+ model=settings.CLAUDE_MODEL,
+ usage_details={
+ "input": usage["input_tokens"],
+ "output": usage["output_tokens"],
+ },
+ output={"sql": sql, "explanation": explanation},
+ metadata={"cost_usd": call_cost},
)
+ return sql, explanation, usage
- except DatabaseError as exc:
- last_error = str(exc)
- if attempt == settings.MAX_RETRIES:
- raise SQLGenerationError(
- f"Could not produce a working query after {settings.MAX_RETRIES} attempts. "
- f"Last error: {last_error}"
- ) from exc
- raise SQLGenerationError("Unexpected agent loop exit.")
+@observe(name="validate_sql", as_type="guardrail")
+def _validate_sql(sql: str) -> str:
+ verdict = validate_sql_ast(sql)
+ if not verdict.allowed:
+ raise UnsafeSQLError(verdict.user_message or verdict.reason)
+ return "allowed"
-async def _generate_sql(system: str, messages: list) -> Tuple[str, str]:
- response = _claude.messages.create(
- model=settings.CLAUDE_MODEL,
- max_tokens=1024,
- system=system,
- messages=messages,
- temperature=0,
- )
- raw = response.content[0].text.strip()
- raw = re.sub(r"^```(?:json)?\s*", "", raw, flags=re.MULTILINE)
- raw = re.sub(r"\s*```$", "", raw, flags=re.MULTILINE)
- try:
- parsed = json.loads(raw)
- sql = parsed.get("sql", "").strip()
- if not sql:
- raise SQLGenerationError("Claude returned JSON but 'sql' is empty.")
- return sql, parsed.get("explanation", "")
- except json.JSONDecodeError as exc:
- raise SQLGenerationError(f"Claude response not valid JSON: {raw[:200]}") from exc
+@observe(name="execute_sql")
+async def _execute_readonly(sql: str) -> Tuple[List[dict], int, bool]:
+ t0 = time.perf_counter()
+ rows = await fetch_readonly(sql)
+ execution_ms = int((time.perf_counter() - t0) * 1000)
+ truncated = len(rows) >= settings.MAX_ROW_LIMIT
+ return rows, execution_ms, truncated
def _assert_safe_sql(sql: str) -> None:
- without_comments = re.sub(r"--[^\n]*", "", sql)
- without_comments = re.sub(r"/\*.*?\*/", "", without_comments, flags=re.DOTALL)
- first_token = without_comments.strip().split()[0].upper() if without_comments.strip() else ""
- if first_token not in ("SELECT", "WITH"):
- raise UnsafeSQLError(f"SQL starts with '{first_token}' — only SELECT or WITH permitted.")
- dangerous = {"DROP", "DELETE", "INSERT", "UPDATE", "TRUNCATE", "ALTER", "CREATE"}
- found = dangerous & set(re.findall(r"\b[A-Za-z]+\b", without_comments.upper()))
- if found:
- raise UnsafeSQLError(f"SQL contains dangerous keyword(s): {found}")
+ """Backward-compatible wrapper for tests — delegates to AST guardrails."""
+ verdict = validate_sql_ast(sql)
+ if not verdict.allowed:
+ raise UnsafeSQLError(verdict.user_message or verdict.reason)
+
+
+async def _log_blocked_query(
+ question: str,
+ sql: str,
+ reason: str,
+ user_id: Optional[int],
+ cost_usd: float = 0.0,
+) -> None:
+ """Audit blocked guardrail attempts without failing the request path."""
+ await log_query_audit(
+ user_id=user_id,
+ question=question,
+ sql=sql,
+ verdict=VERDICT_BLOCKED,
+ row_count=0,
+ cost_usd=cost_usd,
+ )
+ try:
+ await database_pool.execute(
+ """
+ INSERT INTO query_history
+ (question, generated_sql, explanation, result_json,
+ row_count, execution_ms, retries, user_id)
+ VALUES ($1, $2, $3, $4::jsonb, $5, $6, $7, $8)
+ """,
+ question,
+ sql or "-- blocked before SQL generation --",
+ f"BLOCKED: {reason}",
+ json.dumps({"blocked": True, "reason": reason}),
+ 0,
+ 0,
+ 0,
+ user_id,
+ )
+ except Exception:
+ logger.exception(
+ "Failed to write blocked query to query_history",
+ extra={"user_id": user_id, "reason": reason},
+ )
def _serialise_row(row: Dict[str, Any]) -> Dict[str, Any]:
@@ -219,14 +292,20 @@ async def _log_query_history(
execution_ms: int,
retries: int,
user_id: Optional[int],
+ cost_usd: float = 0.0,
) -> None:
"""
Persist query to audit log with full result cache.
result_json stores rows + columns so history replays are free (no Claude call).
"""
- # asyncpg has no built-in dict→jsonb encoder. Serialise to JSON text and
- # cast in SQL (`$4::jsonb`); otherwise the INSERT raises silently and
- # history rows are lost (the rate-limit counter still increments).
+ await log_query_audit(
+ user_id=user_id,
+ question=question,
+ sql=sql,
+ verdict=VERDICT_ALLOWED,
+ row_count=row_count,
+ cost_usd=cost_usd,
+ )
try:
result_json = json.dumps({"rows": rows, "columns": columns}, default=str)
await database_pool.execute(
@@ -246,9 +325,6 @@ async def _log_query_history(
user_id,
)
except Exception:
- # Best-effort: the user already saw their result; do not 500 a successful
- # query because the audit-log write failed. Log loudly so we notice if
- # the silent-loss regression returns.
logger.exception(
"Failed to write query_history",
extra={"user_id": user_id, "row_count": row_count},
diff --git a/backend/app/api/query.py b/backend/app/api/query.py
index 220c01d..7f12a02 100644
--- a/backend/app/api/query.py
+++ b/backend/app/api/query.py
@@ -1,13 +1,8 @@
"""
InsightIQ — Query Router (Updated for Auth + Rate Limiting)
============================================================
-Changes from Day 3:
-- POST /api/query now requires authentication (CurrentUser dependency)
-- Rate limit checked BEFORE running the agent (atomic DB increment)
-- query_history rows are user-scoped (user_id stored)
-- GET /api/history only returns the current user's queries
-- GET /api/history/:id returns cached result (no Claude call)
-- RateLimitInfo embedded in QueryResponse for UI budget badge
+Layer 6: degraded outcomes return HTTP 200 with a typed message — never 500
+for expected failure modes (guardrail, timeout, unparseable SQL, empty rows).
"""
import json
@@ -16,7 +11,7 @@
from fastapi import APIRouter, HTTPException, Query, status
-from app.agent.sql_agent import run_sql_agent
+from app.agent.sql_agent import _log_blocked_query, run_sql_agent
from app.api.schemas import HistoryResponse, QueryRequest, QueryResponse
from app.auth.dependencies import CurrentUser
from app.auth.schemas import RateLimitResponse
@@ -25,13 +20,13 @@
RateLimitError,
check_and_increment_query_count,
)
-from app.core.exceptions import (
- DatabaseError,
- SchemaLoadError,
- SQLGenerationError,
- UnsafeSQLError,
-)
-from app.db.pool import database_pool
+from app.cache import get_cached_outcome
+from app.core.exceptions import DatabaseError, SchemaLoadError
+from app.db.pool import read_database_pool
+from app.guardrails import scan_question
+from app.observability.tracing import is_tracing_enabled, observe, propagate_attributes
+from app.reliability.failure_modes import FailureMode
+from app.reliability.messages import PROMPT_INJECTION, SCHEMA_UNAVAILABLE
logger = logging.getLogger(__name__)
router = APIRouter()
@@ -50,6 +45,48 @@ def _coerce_result_json(raw: Any) -> dict[str, Any]:
return raw if isinstance(raw, dict) else {}
+def _rate_limit_from_user(user: dict) -> RateLimitResponse:
+ used = user["total_queries"]
+ limit = user["query_limit"]
+ return RateLimitResponse(
+ total_queries=used,
+ query_limit=limit,
+ queries_remaining=max(0, limit - used),
+ limit_reached=used >= limit,
+ )
+
+
+def _outcome_to_response(
+ outcome,
+ rate_limit: RateLimitResponse,
+ *,
+ from_cache: bool = False,
+) -> QueryResponse:
+ explanation = outcome.explanation
+ if outcome.message and outcome.status == "degraded":
+ explanation = outcome.message
+ elif outcome.message and outcome.failure_mode == FailureMode.EMPTY_RESULTS:
+ explanation = outcome.message
+
+ cached = from_cache or getattr(outcome, "from_cache", False)
+
+ return QueryResponse(
+ question=outcome.question,
+ sql=outcome.sql,
+ explanation=explanation,
+ rows=outcome.rows,
+ columns=outcome.columns,
+ row_count=outcome.row_count,
+ execution_ms=outcome.execution_ms,
+ retries=outcome.retries,
+ rate_limit=rate_limit,
+ from_cache=cached,
+ status=outcome.status, # type: ignore[arg-type]
+ message=outcome.message,
+ failure_mode=outcome.failure_mode.value,
+ )
+
+
# ── POST /api/query ────────────────────────────────────────────────────────
@@ -61,20 +98,52 @@ def _coerce_result_json(raw: Any) -> dict[str, Any]:
)
async def query_endpoint(
request: QueryRequest,
- current_user: CurrentUser, # ← requires valid JWT
+ current_user: CurrentUser,
) -> QueryResponse:
"""
Authenticated query endpoint.
Flow:
1. Validate JWT → get current user
- 2. Check + atomically increment query count (429 if at limit)
- 3. Run SQL agent (generate → validate → execute → heal)
- 4. Save result to query_history with user_id + cached result_json
- 5. Return QueryResponse with embedded RateLimitInfo
+ 2. Question guardrail (no rate-limit burn on injection)
+ 3. Check + atomically increment query count (429 if at limit)
+ 4. Run agent via ``answer_safely`` (Layer 6)
+ 5. Return QueryResponse — HTTP 200 even when degraded
"""
user_id = current_user["id"]
+ # ── Question guardrail (before rate limit / Claude) ───────────────────
+ qv = scan_question(request.question)
+ if not qv.allowed:
+ await _log_blocked_query(request.question, "", qv.reason, user_id)
+ return QueryResponse(
+ question=request.question,
+ sql="",
+ explanation=PROMPT_INJECTION,
+ rows=[],
+ columns=[],
+ row_count=0,
+ execution_ms=0,
+ retries=0,
+ rate_limit=_rate_limit_from_user(current_user),
+ status="degraded",
+ message=PROMPT_INJECTION,
+ failure_mode=FailureMode.PROMPT_INJECTION.value,
+ )
+
+ # ── Question cache (before rate limit / Claude) ───────────────────────
+ cached = await get_cached_outcome(request.question)
+ if cached is not None:
+ trace_ctx = (
+ _cache_hit_trace(request.question, user_id) if is_tracing_enabled() else _null_context()
+ )
+ with trace_ctx:
+ return _outcome_to_response(
+ cached,
+ _rate_limit_from_user(current_user),
+ from_cache=True,
+ )
+
# ── Rate limit check (atomic) ──────────────────────────────────────────
try:
updated_user = await check_and_increment_query_count(user_id)
@@ -87,44 +156,24 @@ async def query_endpoint(
except AuthError as exc:
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail=str(exc)) from exc
- # ── Run agent ──────────────────────────────────────────────────────────
+ rate_limit = _rate_limit_from_user(updated_user)
+
+ # ── Run agent (Layer 6 — graceful degradation) ───────────────────────
try:
- result = await run_sql_agent(request.question, user_id=user_id)
- except UnsafeSQLError as exc:
- raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc
- except SQLGenerationError as exc:
+ outcome = await run_sql_agent(request.question, user_id=user_id)
+ except SchemaLoadError as exc:
raise HTTPException(
- status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
- detail="Could not generate a valid query. Try rephrasing.",
+ status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+ detail=SCHEMA_UNAVAILABLE,
) from exc
- except (DatabaseError, SchemaLoadError) as exc:
+ except DatabaseError as exc:
+ logger.exception("unexpected database error in query path")
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
- detail="Database temporarily unavailable.",
+ detail=SCHEMA_UNAVAILABLE,
) from exc
- # ── Build rate limit info for UI ───────────────────────────────────────
- queries_used = updated_user["total_queries"]
- query_limit = updated_user["query_limit"]
- rate_limit = RateLimitResponse(
- total_queries=queries_used,
- query_limit=query_limit,
- queries_remaining=max(0, query_limit - queries_used),
- limit_reached=queries_used >= query_limit,
- )
-
- return QueryResponse(
- question=result.question,
- sql=result.sql,
- explanation=result.explanation,
- rows=result.rows,
- columns=result.columns,
- row_count=result.row_count,
- execution_ms=result.execution_ms,
- retries=result.retries,
- rate_limit=rate_limit,
- from_cache=False,
- )
+ return _outcome_to_response(outcome, rate_limit)
# ── GET /api/history ───────────────────────────────────────────────────────
@@ -141,14 +190,11 @@ async def history_endpoint(
limit: int = Query(default=20, ge=1, le=100),
offset: int = Query(default=0, ge=0),
) -> HistoryResponse:
- """
- Return paginated query history for the authenticated user only.
- WHERE user_id = $current_user.id enforced at DB level.
- """
+ """Return paginated query history for the authenticated user only."""
user_id = current_user["id"]
try:
- rows = await database_pool.fetch(
+ rows = await read_database_pool.fetch(
"""
SELECT id, question, generated_sql, row_count,
execution_ms, retries, created_at,
@@ -163,7 +209,7 @@ async def history_endpoint(
limit,
offset,
)
- count = await database_pool.fetchrow(
+ count = await read_database_pool.fetchrow(
"SELECT COUNT(*) AS total FROM query_history WHERE user_id = $1",
user_id,
)
@@ -181,57 +227,77 @@ async def history_endpoint(
status_code=status.HTTP_200_OK,
summary="Load a cached query result (no Claude call)",
)
+@observe(name="history_cache_load")
async def history_item_endpoint(
item_id: int,
current_user: CurrentUser,
) -> dict:
- """
- Return the cached result for a specific history item.
-
- This is the key cost-saving endpoint:
- - Clicking a history item calls GET /api/history/:id (free DB read)
- - NOT POST /api/query (paid Claude call)
- - result_json contains the full rows + columns from the original run
-
- Security: WHERE user_id = $current_user.id ensures users can only
- access their own history items — not other users' queries.
- """
+ """Return the cached result for a specific history item."""
user_id = current_user["id"]
- try:
- row = await database_pool.fetchrow(
- """
- SELECT id, question, generated_sql, explanation,
- result_json, row_count, execution_ms, retries, created_at
- FROM query_history
- WHERE id = $1 AND user_id = $2
- """,
- item_id,
- user_id,
- )
- except DatabaseError as exc:
- raise HTTPException(status_code=503, detail="Database error.") from exc
-
- if not row:
- raise HTTPException(
- status_code=status.HTTP_404_NOT_FOUND,
- detail="History item not found.",
+ trace_ctx = (
+ propagate_attributes(
+ user_id=str(user_id),
+ tags=["cache_hit"],
+ metadata={"history_item_id": str(item_id)},
)
-
- # Reconstruct a QueryResponse-compatible shape from the cache
- cached = _coerce_result_json(row["result_json"])
- return {
- "question": row["question"],
- "sql": row["generated_sql"],
- "explanation": row["explanation"] or "",
- "rows": cached.get("rows", []),
- "columns": cached.get("columns", []),
- "row_count": row["row_count"] or 0,
- "execution_ms": row["execution_ms"] or 0,
- "retries": row["retries"],
- "from_cache": True, # UI shows "Loaded from history" badge
- "created_at": row["created_at"].isoformat(),
- }
+ if is_tracing_enabled()
+ else _null_context()
+ )
+ with trace_ctx:
+ try:
+ row = await read_database_pool.fetchrow(
+ """
+ SELECT id, question, generated_sql, explanation,
+ result_json, row_count, execution_ms, retries, created_at
+ FROM query_history
+ WHERE id = $1 AND user_id = $2
+ """,
+ item_id,
+ user_id,
+ )
+ except DatabaseError as exc:
+ raise HTTPException(status_code=503, detail="Database error.") from exc
+
+ if not row:
+ raise HTTPException(
+ status_code=status.HTTP_404_NOT_FOUND,
+ detail="History item not found.",
+ )
+
+ cached = _coerce_result_json(row["result_json"])
+ return {
+ "question": row["question"],
+ "sql": row["generated_sql"],
+ "explanation": row["explanation"] or "",
+ "rows": cached.get("rows", []),
+ "columns": cached.get("columns", []),
+ "row_count": row["row_count"] or 0,
+ "execution_ms": row["execution_ms"] or 0,
+ "retries": row["retries"],
+ "from_cache": True,
+ "created_at": row["created_at"].isoformat(),
+ "status": "ok",
+ "message": "",
+ }
+
+
+def _null_context():
+ from contextlib import nullcontext
+
+ return nullcontext()
+
+
+def _cache_hit_trace(question: str, user_id: int):
+ """Langfuse trace for a normalised-question cache hit (zero tokens)."""
+ from langfuse import get_client
+
+ langfuse = get_client()
+ return langfuse.start_as_current_observation(
+ name="insightiq_query",
+ input={"question": question, "from_cache": True},
+ metadata={"tags": "cache_hit"},
+ )
# ── GET /api/schema ────────────────────────────────────────────────────────
@@ -241,7 +307,7 @@ async def history_item_endpoint(
async def schema_endpoint(current_user: CurrentUser) -> dict:
"""Schema explorer — authenticated only."""
try:
- tables = await database_pool.fetch(
+ tables = await read_database_pool.fetch(
"""SELECT table_name FROM information_schema.tables
WHERE table_schema='public' AND table_type='BASE TABLE'
ORDER BY table_name"""
diff --git a/backend/app/api/schemas.py b/backend/app/api/schemas.py
index 16076a3..4956ea1 100644
--- a/backend/app/api/schemas.py
+++ b/backend/app/api/schemas.py
@@ -7,7 +7,7 @@
"""
from datetime import datetime
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Literal, Optional
from pydantic import BaseModel, Field, field_validator
@@ -32,8 +32,11 @@ class QueryResponse(BaseModel):
row_count: int
execution_ms: int
retries: int
- rate_limit: RateLimitResponse # ← new: budget info
- from_cache: bool = False # ← new: True if loaded from history
+ rate_limit: RateLimitResponse
+ from_cache: bool = False
+ status: Literal["ok", "degraded"] = "ok"
+ message: str = ""
+ failure_mode: Optional[str] = None
created_at: datetime = Field(default_factory=datetime.utcnow)
diff --git a/backend/app/cache/__init__.py b/backend/app/cache/__init__.py
new file mode 100644
index 0000000..78feea9
--- /dev/null
+++ b/backend/app/cache/__init__.py
@@ -0,0 +1,11 @@
+"""InsightIQ — optional question-result cache (zero extra infra)."""
+
+from app.cache.answer import answer_cached, get_cached_outcome
+from app.cache.keys import cache_key, normalize_question
+
+__all__ = [
+ "answer_cached",
+ "cache_key",
+ "get_cached_outcome",
+ "normalize_question",
+]
diff --git a/backend/app/cache/answer.py b/backend/app/cache/answer.py
new file mode 100644
index 0000000..6d2e0df
--- /dev/null
+++ b/backend/app/cache/answer.py
@@ -0,0 +1,92 @@
+"""Cache-aware answer path — skips LLM on normalised question hits."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.cache.keys import normalize_question
+from app.cache.store import get_cached_payload, set_cached_payload
+from app.cost.meter import CostMeter
+from app.reliability.answer import answer_safely
+from app.reliability.failure_modes import FailureMode
+from app.reliability.outcome import AgentOutcome
+
+logger = logging.getLogger(__name__)
+
+
+def outcome_to_cache_payload(outcome: AgentOutcome) -> dict:
+ return {
+ "failure_mode": outcome.failure_mode.value,
+ "message": outcome.message,
+ "sql": outcome.sql,
+ "explanation": outcome.explanation,
+ "rows": outcome.rows,
+ "columns": outcome.columns,
+ "row_count": outcome.row_count,
+ "execution_ms": outcome.execution_ms,
+ "retries": outcome.retries,
+ "truncated": outcome.truncated,
+ "input_tokens": 0,
+ "output_tokens": 0,
+ "cost_usd": 0.0,
+ "verdict": outcome.verdict,
+ }
+
+
+def outcome_from_cache_payload(
+ payload: dict,
+ *,
+ question: str,
+) -> AgentOutcome:
+ return AgentOutcome(
+ failure_mode=FailureMode(payload.get("failure_mode", "none")),
+ message=payload.get("message", ""),
+ question=question,
+ sql=payload.get("sql", ""),
+ explanation=payload.get("explanation", ""),
+ rows=payload.get("rows", []),
+ columns=payload.get("columns", []),
+ row_count=payload.get("row_count", 0),
+ execution_ms=payload.get("execution_ms", 0),
+ retries=payload.get("retries", 0),
+ truncated=payload.get("truncated", False),
+ input_tokens=0,
+ output_tokens=0,
+ cost_usd=0.0,
+ verdict=payload.get("verdict", "allowed"),
+ from_cache=True,
+ )
+
+
+async def get_cached_outcome(question: str) -> AgentOutcome | None:
+ """Fast path for API: return a cache hit without touching the agent."""
+ payload = await get_cached_payload(question)
+ if payload is None:
+ return None
+ logger.info(
+ "query_cache hit",
+ extra={"question_norm": normalize_question(question)},
+ )
+ return outcome_from_cache_payload(payload, question=question)
+
+
+async def answer_cached(
+ question: str,
+ user_id: Optional[int],
+ *,
+ meter: CostMeter,
+) -> tuple[AgentOutcome, bool]:
+ """
+ Return ``(outcome, cache_hit)``.
+
+ On miss, runs ``answer_safely`` and stores successful outcomes until TTL.
+ """
+ cached = await get_cached_outcome(question)
+ if cached is not None:
+ return cached, True
+
+ outcome = await answer_safely(question, user_id=user_id, meter=meter)
+ if outcome.cacheable:
+ await set_cached_payload(question, outcome_to_cache_payload(outcome))
+ return outcome, False
diff --git a/backend/app/cache/keys.py b/backend/app/cache/keys.py
new file mode 100644
index 0000000..41ffeb0
--- /dev/null
+++ b/backend/app/cache/keys.py
@@ -0,0 +1,16 @@
+"""Normalised question → cache key."""
+
+from __future__ import annotations
+
+import hashlib
+
+
+def normalize_question(question: str) -> str:
+ """Cheap normalisation so spacing/case variants share one cache entry."""
+ return " ".join(question.lower().split())
+
+
+def cache_key(question: str) -> str:
+ norm = normalize_question(question)
+ digest = hashlib.sha256(norm.encode()).hexdigest()[:16]
+ return f"iq:{digest}"
diff --git a/backend/app/cache/store.py b/backend/app/cache/store.py
new file mode 100644
index 0000000..45b6c18
--- /dev/null
+++ b/backend/app/cache/store.py
@@ -0,0 +1,62 @@
+"""Postgres-backed query cache with TTL."""
+
+from __future__ import annotations
+
+import json
+import logging
+from typing import Any
+
+from app.cache.keys import cache_key, normalize_question
+from app.core.config import settings
+from app.db.pool import database_pool, read_database_pool
+
+logger = logging.getLogger(__name__)
+
+
+async def get_cached_payload(question: str) -> dict[str, Any] | None:
+ """Return cached outcome payload if present and not expired."""
+ key = cache_key(question)
+ try:
+ row = await read_database_pool.fetchrow(
+ """
+ SELECT payload
+ FROM query_cache
+ WHERE cache_key = $1 AND expires_at > NOW()
+ """,
+ key,
+ )
+ except Exception:
+ logger.exception("query_cache read failed", extra={"cache_key": key})
+ return None
+
+ if not row:
+ return None
+
+ payload = row["payload"]
+ if isinstance(payload, str):
+ return json.loads(payload)
+ return dict(payload) if payload else None
+
+
+async def set_cached_payload(question: str, payload: dict[str, Any]) -> None:
+ """Store a successful outcome; upsert refreshes TTL."""
+ key = cache_key(question)
+ norm = normalize_question(question)
+ ttl = settings.CACHE_TTL_SECONDS
+ try:
+ await database_pool.execute(
+ """
+ INSERT INTO query_cache (cache_key, question_norm, payload, expires_at)
+ VALUES ($1, $2, $3::jsonb, NOW() + make_interval(secs => $4))
+ ON CONFLICT (cache_key) DO UPDATE
+ SET payload = EXCLUDED.payload,
+ question_norm = EXCLUDED.question_norm,
+ expires_at = EXCLUDED.expires_at
+ """,
+ key,
+ norm,
+ json.dumps(payload, default=str),
+ float(ttl),
+ )
+ except Exception:
+ logger.exception("query_cache write failed", extra={"cache_key": key})
diff --git a/backend/app/core/config.py b/backend/app/core/config.py
index 6b5073d..88461e8 100644
--- a/backend/app/core/config.py
+++ b/backend/app/core/config.py
@@ -12,7 +12,7 @@
from functools import lru_cache
from typing import List
-from pydantic import Field, field_validator
+from pydantic import Field, field_validator, model_validator
from pydantic_settings import BaseSettings
@@ -23,9 +23,14 @@ class Settings(BaseSettings):
# PostgreSQL (Neon)
DATABASE_URL: str = Field(...)
+ DATABASE_URL_RO: str = Field(
+ default="",
+ description="Read-only DSN for user SQL and schema introspection",
+ )
DB_POOL_MIN: int = Field(default=2)
DB_POOL_MAX: int = Field(default=10)
DB_QUERY_TIMEOUT: float = Field(default=30.0)
+ STATEMENT_TIMEOUT_MS: int = Field(default=5000)
# Auth
JWT_SECRET_KEY: str = Field(...)
@@ -33,7 +38,30 @@ class Settings(BaseSettings):
# Agent
MAX_RETRIES: int = Field(default=3)
- MAX_RESULT_ROWS: int = Field(default=500)
+ MAX_ROW_LIMIT: int = Field(default=1000)
+ BUDGET_USD_DAILY: float = Field(default=2.00)
+ CLAUDE_INPUT_USD_PER_MTOK: float = Field(
+ default=3.0,
+ description="USD per 1M input tokens (claude-sonnet-4-5 default)",
+ )
+ CLAUDE_OUTPUT_USD_PER_MTOK: float = Field(
+ default=15.0,
+ description="USD per 1M output tokens (claude-sonnet-4-5 default)",
+ )
+ COST_ALERT_WEBHOOK_URL: str = Field(
+ default="",
+ description="Slack-compatible webhook for daily budget alerts",
+ )
+ CACHE_TTL_SECONDS: int = Field(default=900)
+
+ # Langfuse (Layer 1 observability — optional locally, required in prod)
+ LANGFUSE_PUBLIC_KEY: str = Field(default="")
+ LANGFUSE_SECRET_KEY: str = Field(default="")
+ LANGFUSE_BASE_URL: str = Field(default="https://cloud.langfuse.com")
+
+ @property
+ def langfuse_enabled(self) -> bool:
+ return bool(self.LANGFUSE_PUBLIC_KEY and self.LANGFUSE_SECRET_KEY)
# CORS — CRITICAL for ACA multi-app setup
# Set FRONTEND_URL in Azure Container App env vars via GitHub Secrets
@@ -52,10 +80,20 @@ class Settings(BaseSettings):
LOG_LEVEL: str = Field(default="INFO")
- @field_validator("DATABASE_URL")
+ @staticmethod
+ def _fix_postgres_url(v: str) -> str:
+ return v.replace("postgres://", "postgresql://", 1) if v.startswith("postgres://") else v
+
+ @field_validator("DATABASE_URL", "DATABASE_URL_RO")
@classmethod
def fix_db_url(cls, v: str) -> str:
- return v.replace("postgres://", "postgresql://", 1) if v.startswith("postgres://") else v
+ return cls._fix_postgres_url(v)
+
+ @model_validator(mode="after")
+ def default_readonly_url(self) -> "Settings":
+ if not self.DATABASE_URL_RO:
+ self.DATABASE_URL_RO = self.DATABASE_URL
+ return self
def get_all_cors_origins(self) -> List[str]:
"""
diff --git a/backend/app/core/exceptions.py b/backend/app/core/exceptions.py
index a08050e..a5dd2de 100644
--- a/backend/app/core/exceptions.py
+++ b/backend/app/core/exceptions.py
@@ -14,6 +14,10 @@ class DatabaseError(InsightIQError):
"""Raised when a database operation fails after all retries."""
+class QueryTimeoutError(DatabaseError):
+ """Raised when a query hits statement_timeout or pool command timeout."""
+
+
class SQLGenerationError(InsightIQError):
"""Raised when the LLM cannot produce valid SQL after max retries."""
diff --git a/backend/app/cost/__init__.py b/backend/app/cost/__init__.py
new file mode 100644
index 0000000..37095cc
--- /dev/null
+++ b/backend/app/cost/__init__.py
@@ -0,0 +1,6 @@
+"""InsightIQ — Layer 5 cost metering."""
+
+from app.cost.meter import CostMeter
+from app.cost.pricing import price_tokens
+
+__all__ = ["CostMeter", "price_tokens"]
diff --git a/backend/app/cost/alerts.py b/backend/app/cost/alerts.py
new file mode 100644
index 0000000..6ff77c9
--- /dev/null
+++ b/backend/app/cost/alerts.py
@@ -0,0 +1,59 @@
+"""Budget alert delivery (Slack-compatible webhook)."""
+
+from __future__ import annotations
+
+import json
+import logging
+import urllib.error
+import urllib.request
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+
+def send_budget_alert(*, daily_spend_usd: float, budget_usd: float) -> bool:
+ """
+ Fire a one-shot alert when daily spend crosses BUDGET_USD_DAILY.
+
+ Returns True if a webhook was delivered successfully.
+ """
+ message = f"InsightIQ daily spend ${daily_spend_usd:.4f} " f"exceeded budget ${budget_usd:.2f}"
+ logger.warning(message)
+
+ webhook = (settings.COST_ALERT_WEBHOOK_URL or "").strip()
+ if not webhook:
+ logger.info("COST_ALERT_WEBHOOK_URL unset — budget alert logged only")
+ return False
+
+ payload = json.dumps(
+ {
+ "text": message,
+ "blocks": [
+ {
+ "type": "section",
+ "text": {
+ "type": "mrkdwn",
+ "text": (
+ f":warning: *InsightIQ budget alert*\n"
+ f"Daily spend: `${daily_spend_usd:.4f}`\n"
+ f"Budget: `${budget_usd:.2f}`"
+ ),
+ },
+ }
+ ],
+ }
+ ).encode()
+
+ req = urllib.request.Request(
+ webhook,
+ data=payload,
+ headers={"Content-Type": "application/json"},
+ method="POST",
+ )
+ try:
+ with urllib.request.urlopen(req, timeout=10) as resp:
+ return 200 <= resp.status < 300
+ except urllib.error.URLError as exc:
+ logger.exception("Budget alert webhook failed: %s", exc)
+ return False
diff --git a/backend/app/cost/daily.py b/backend/app/cost/daily.py
new file mode 100644
index 0000000..fc07538
--- /dev/null
+++ b/backend/app/cost/daily.py
@@ -0,0 +1,92 @@
+"""Daily spend ledger and budget enforcement (Layer 5)."""
+
+from __future__ import annotations
+
+import logging
+
+from app.core.config import settings
+from app.cost.alerts import send_budget_alert
+from app.cost.meter import CostMeter
+from app.db.pool import database_pool
+
+logger = logging.getLogger(__name__)
+
+
+async def bump_daily_spend(amount_usd: float) -> float:
+ """Atomically add ``amount_usd`` to today's running total; return new total."""
+ if amount_usd <= 0:
+ return await get_daily_spend()
+
+ row = await database_pool.fetchrow(
+ """
+ INSERT INTO daily_cost_spend (day, usd)
+ VALUES (CURRENT_DATE, $1)
+ ON CONFLICT (day) DO UPDATE
+ SET usd = daily_cost_spend.usd + EXCLUDED.usd,
+ updated_at = NOW()
+ RETURNING usd
+ """,
+ amount_usd,
+ )
+ return float(row["usd"]) if row else 0.0
+
+
+async def get_daily_spend() -> float:
+ row = await database_pool.fetchrow("SELECT usd FROM daily_cost_spend WHERE day = CURRENT_DATE")
+ return float(row["usd"]) if row else 0.0
+
+
+async def _alert_if_needed(daily_total: float) -> None:
+ """Send at most one webhook alert per UTC day when budget is exceeded."""
+ if daily_total <= settings.BUDGET_USD_DAILY:
+ return
+
+ row = await database_pool.fetchrow(
+ "SELECT alerted FROM daily_cost_spend WHERE day = CURRENT_DATE"
+ )
+ if row and row["alerted"]:
+ return
+
+ if send_budget_alert(
+ daily_spend_usd=daily_total,
+ budget_usd=settings.BUDGET_USD_DAILY,
+ ):
+ await database_pool.execute(
+ """
+ UPDATE daily_cost_spend
+ SET alerted = TRUE, updated_at = NOW()
+ WHERE day = CURRENT_DATE
+ """
+ )
+
+
+async def finalize_query_cost(meter: CostMeter) -> float:
+ """
+ Persist per-query spend and check the daily budget.
+
+ Called once per ``run_sql_agent`` invocation (including partial spend on
+ blocked or failed queries that already called Claude).
+ """
+ if meter.usd <= 0:
+ return await get_daily_spend()
+
+ try:
+ daily_total = await bump_daily_spend(meter.usd)
+ await _alert_if_needed(daily_total)
+ logger.info(
+ "Query cost recorded",
+ extra={
+ "cost_usd": meter.usd,
+ "daily_spend_usd": daily_total,
+ "input_tokens": meter.input_tokens,
+ "output_tokens": meter.output_tokens,
+ "claude_calls": meter.call_count,
+ },
+ )
+ return daily_total
+ except Exception:
+ logger.exception(
+ "Failed to record query cost",
+ extra={"cost_usd": meter.usd},
+ )
+ return await get_daily_spend()
diff --git a/backend/app/cost/meter.py b/backend/app/cost/meter.py
new file mode 100644
index 0000000..525e54e
--- /dev/null
+++ b/backend/app/cost/meter.py
@@ -0,0 +1,45 @@
+"""Per-query cost accumulator — sums all Claude calls in one request."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+
+from app.core.config import settings
+from app.cost.pricing import price_tokens
+
+
+@dataclass
+class CostMeter:
+ """Accumulates tokens and USD across SQL (+ retry) generation calls."""
+
+ input_tokens: int = 0
+ output_tokens: int = 0
+ usd: float = 0.0
+ call_count: int = 0
+ _model: str = field(default_factory=lambda: settings.CLAUDE_MODEL, repr=False)
+
+ def add(
+ self,
+ input_tokens: int,
+ output_tokens: int,
+ *,
+ model: str | None = None,
+ ) -> float:
+ """Add one Claude usage record; return incremental USD for this call."""
+ self.input_tokens += input_tokens
+ self.output_tokens += output_tokens
+ self.call_count += 1
+ incremental = price_tokens(
+ input_tokens,
+ output_tokens,
+ model=model or self._model,
+ )
+ self.usd = round(self.usd + incremental, 6)
+ return incremental
+
+ def add_usage(self, usage: dict[str, int], *, model: str | None = None) -> float:
+ return self.add(
+ usage.get("input_tokens", 0) or 0,
+ usage.get("output_tokens", 0) or 0,
+ model=model,
+ )
diff --git a/backend/app/cost/pricing.py b/backend/app/cost/pricing.py
new file mode 100644
index 0000000..1bc9a53
--- /dev/null
+++ b/backend/app/cost/pricing.py
@@ -0,0 +1,30 @@
+"""Anthropic token → USD pricing for cost metering."""
+
+from __future__ import annotations
+
+from app.core.config import settings
+
+# Per-million-token rates (USD) — override via env for new model SKUs
+_DEFAULT_RATES: dict[str, tuple[float, float]] = {
+ "claude-sonnet-4-5": (3.0, 15.0),
+ "claude-sonnet-4-20250514": (3.0, 15.0),
+ "claude-3-5-sonnet-20241022": (3.0, 15.0),
+}
+
+
+def _rates_for_model(model: str | None) -> tuple[float, float]:
+ if model and model in _DEFAULT_RATES:
+ return _DEFAULT_RATES[model]
+ return (settings.CLAUDE_INPUT_USD_PER_MTOK, settings.CLAUDE_OUTPUT_USD_PER_MTOK)
+
+
+def price_tokens(
+ input_tokens: int,
+ output_tokens: int,
+ *,
+ model: str | None = None,
+) -> float:
+ """Return estimated USD for a single Claude API call."""
+ input_rate, output_rate = _rates_for_model(model or settings.CLAUDE_MODEL)
+ usd = (input_tokens / 1_000_000) * input_rate + (output_tokens / 1_000_000) * output_rate
+ return round(usd, 6)
diff --git a/backend/app/db/audit.py b/backend/app/db/audit.py
new file mode 100644
index 0000000..5fd9598
--- /dev/null
+++ b/backend/app/db/audit.py
@@ -0,0 +1,49 @@
+"""
+InsightIQ — query audit log (Layer 4)
+=====================================
+Every query attempt is recorded on the write pool — never on the user
+request read path. Complements ``query_history`` (UI cache) with a
+security-focused audit trail.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.db.pool import database_pool
+
+logger = logging.getLogger(__name__)
+
+VERDICT_ALLOWED = "allowed"
+VERDICT_BLOCKED = "blocked"
+
+
+async def log_query_audit(
+ *,
+ user_id: Optional[int],
+ question: str,
+ sql: str,
+ verdict: str,
+ row_count: int = 0,
+ cost_usd: float = 0.0,
+) -> None:
+ """Best-effort audit insert — failures must not break the user response."""
+ try:
+ await database_pool.execute(
+ """
+ INSERT INTO query_audit (user_id, question, sql, verdict, row_count, cost_usd)
+ VALUES ($1, $2, $3, $4, $5, $6)
+ """,
+ user_id,
+ question,
+ sql or "-- no sql generated --",
+ verdict,
+ row_count,
+ cost_usd,
+ )
+ except Exception:
+ logger.exception(
+ "Failed to write query_audit",
+ extra={"user_id": user_id, "verdict": verdict},
+ )
diff --git a/backend/app/db/data_access.py b/backend/app/db/data_access.py
new file mode 100644
index 0000000..cdbe076
--- /dev/null
+++ b/backend/app/db/data_access.py
@@ -0,0 +1,66 @@
+"""
+InsightIQ — Layer 4 data access (database-side defenses)
+=========================================================
+Assumes Layer 3 guardrails may miss something. User SQL runs on the
+read-only pool with enforced LIMIT, statement_timeout, and read-only txn.
+"""
+
+from __future__ import annotations
+
+import logging
+
+import asyncpg
+import sqlglot
+from sqlglot import exp
+
+from app.core.config import settings
+from app.db.pool import read_database_pool
+
+logger = logging.getLogger(__name__)
+
+
+def _literal_int(node: exp.Expression | None) -> int | None:
+ if isinstance(node, exp.Literal) and not node.is_string:
+ try:
+ return int(node.this)
+ except (TypeError, ValueError):
+ return None
+ return None
+
+
+def enforce_limit(sql: str, cap: int) -> str:
+ """
+ Append LIMIT if missing; cap literal limits that exceed ``cap``.
+
+ Applied at execution time so a guardrail miss cannot return unbounded rows.
+ """
+ tree = sqlglot.parse_one(sql, read="postgres")
+ limit_node = tree.args.get("limit")
+ if limit_node is None:
+ return tree.limit(cap).sql(dialect="postgres")
+
+ current = _literal_int(limit_node.expression)
+ if current is not None and current > cap:
+ tree.set("limit", exp.Limit(expression=exp.Literal.number(cap)))
+ return tree.sql(dialect="postgres")
+
+
+async def _harden_read_session(conn: asyncpg.Connection) -> None:
+ """Per-request session guards on the read-only pool."""
+ await conn.execute(f"SET statement_timeout = '{settings.STATEMENT_TIMEOUT_MS}ms'")
+ await conn.execute("SET default_transaction_read_only = on")
+
+
+async def fetch_readonly(sql: str) -> list[dict]:
+ """
+ Execute user SQL on the read-only pool with Layer 4 session guards.
+
+ - ``enforce_limit`` before execution
+ - ``statement_timeout`` per connection
+ - ``default_transaction_read_only = on`` per connection
+ """
+ bounded_sql = enforce_limit(sql, settings.MAX_ROW_LIMIT)
+ return await read_database_pool.fetch_with_session(
+ bounded_sql,
+ _harden_read_session,
+ )
diff --git a/backend/app/db/pool.py b/backend/app/db/pool.py
index cbae2c2..37e94bb 100644
--- a/backend/app/db/pool.py
+++ b/backend/app/db/pool.py
@@ -1,42 +1,52 @@
"""
-InsightIQ — asyncpg Connection Pool
-=====================================
-Manages a single shared asyncpg pool for the entire application.
-
-Design decisions:
-- Pool is a singleton — created once on startup, closed on shutdown.
-- min_size=2 keeps warm connections ready for the first requests.
-- max_size=10 prevents overwhelming a small Postgres instance.
-- Queries have a configurable timeout to prevent pool exhaustion
- from long-running analytical queries.
-- All public methods log timing for observability.
+InsightIQ — asyncpg Connection Pools
+======================================
+Two pools:
+- database_pool (write): auth, query_history INSERT, migrations path uses owner DSN
+- read_database_pool (read): user SQL, schema introspection, history GETs
+
+The read pool uses DATABASE_URL_RO (insightiq_ro) so the request path cannot
+modify data even if application-level SQL guards fail.
"""
import asyncio
import logging
import time
-from typing import Any, List, Optional
+from typing import Any, Awaitable, Callable, List, Optional
import asyncpg
from asyncpg import Pool, Record
from app.core.config import settings
-from app.core.exceptions import DatabaseError
+from app.core.exceptions import DatabaseError, QueryTimeoutError
logger = logging.getLogger(__name__)
+ConnectionInit = Callable[[asyncpg.Connection], Awaitable[None]]
+
class DatabasePool:
"""
Async context-aware wrapper around asyncpg.Pool.
Usage:
- await database_pool.initialise() # called in lifespan startup
- rows = await database_pool.fetch("SELECT * FROM customers LIMIT 10")
- await database_pool.close() # called in lifespan shutdown
+ await database_pool.initialise()
+ rows = await read_database_pool.fetch("SELECT * FROM customers LIMIT 10")
+ await database_pool.close()
"""
- def __init__(self) -> None:
+ def __init__(
+ self,
+ dsn: str,
+ *,
+ label: str,
+ command_timeout: float,
+ init: Optional[ConnectionInit] = None,
+ ) -> None:
+ self._dsn = dsn
+ self._label = label
+ self._command_timeout = command_timeout
+ self._init = init or self._set_type_codecs
self._pool: Optional[Pool] = None
self._init_lock = asyncio.Lock()
@@ -51,41 +61,38 @@ async def ensure_ready(self) -> None:
await self.initialise()
async def initialise(self) -> None:
- """
- Create the asyncpg pool.
- Called once during FastAPI lifespan startup.
- Raises DatabaseError if the connection cannot be established.
- """
+ """Create the asyncpg pool."""
try:
self._pool = await asyncpg.create_pool(
- dsn=settings.DATABASE_URL,
+ dsn=self._dsn,
min_size=settings.DB_POOL_MIN,
max_size=settings.DB_POOL_MAX,
- command_timeout=settings.DB_QUERY_TIMEOUT,
- # Return rows as plain dicts rather than asyncpg Record objects
- # for easier JSON serialisation downstream
- init=self._set_type_codecs,
+ command_timeout=self._command_timeout,
+ init=self._init,
)
logger.info(
"asyncpg pool created",
extra={
+ "pool": self._label,
"min_size": settings.DB_POOL_MIN,
"max_size": settings.DB_POOL_MAX,
},
)
except Exception as exc:
- logger.exception("Failed to create database pool")
- raise DatabaseError(f"Cannot connect to database: {exc}") from exc
+ logger.exception("Failed to create database pool", extra={"pool": self._label})
+ raise DatabaseError(f"Cannot connect to database ({self._label}): {exc}") from exc
async def close(self) -> None:
- """
- Gracefully close all pool connections.
- Called during FastAPI lifespan shutdown.
- """
+ """Gracefully close all pool connections."""
if self._pool:
await self._pool.close()
self._pool = None
- logger.info("asyncpg pool closed")
+ logger.info("asyncpg pool closed", extra={"pool": self._label})
+
+ async def reset(self) -> None:
+ """Close the pool and drop loop-bound state (pytest / hot reload)."""
+ await self.close()
+ self._init_lock = asyncio.Lock()
# ── Query methods ──────────────────────────────────────────────────────
@@ -95,40 +102,31 @@ async def fetch(
*args: Any,
timeout: Optional[float] = None,
) -> List[dict]:
- """
- Execute a SELECT query and return all rows as a list of dicts.
-
- Args:
- query: SQL query string (parameterised with $1, $2, ...)
- *args: Query parameters passed to asyncpg
- timeout: Per-query timeout override (defaults to DB_QUERY_TIMEOUT)
-
- Returns:
- List of row dicts with column names as keys.
-
- Raises:
- DatabaseError: On any asyncpg or timeout error.
- """
+ """Execute a SELECT query and return all rows as a list of dicts."""
await self.ensure_ready()
start = time.perf_counter()
+ effective_timeout = timeout or self._command_timeout
try:
async with self._pool.acquire() as conn:
- records: List[Record] = await conn.fetch(
- query, *args, timeout=timeout or settings.DB_QUERY_TIMEOUT
- )
+ records: List[Record] = await conn.fetch(query, *args, timeout=effective_timeout)
rows = [dict(r) for r in records]
elapsed_ms = int((time.perf_counter() - start) * 1000)
logger.debug(
"Query executed",
- extra={"rows": len(rows), "elapsed_ms": elapsed_ms},
+ extra={"pool": self._label, "rows": len(rows), "elapsed_ms": elapsed_ms},
)
return rows
except asyncpg.TooManyConnectionsError as exc:
raise DatabaseError("Database connection pool exhausted") from exc
except asyncpg.QueryCanceledError as exc:
- raise DatabaseError(f"Query timed out after {settings.DB_QUERY_TIMEOUT}s") from exc
+ raise QueryTimeoutError(
+ f"Query timed out after {effective_timeout}s ({self._label} pool)"
+ ) from exc
except Exception as exc:
- logger.exception("Database fetch failed", extra={"query": query[:200]})
+ logger.exception(
+ "Database fetch failed",
+ extra={"pool": self._label, "query": query[:200]},
+ )
raise DatabaseError(f"Query failed: {exc}") from exc
async def execute(
@@ -137,20 +135,13 @@ async def execute(
*args: Any,
timeout: Optional[float] = None,
) -> str:
- """
- Execute a non-SELECT statement (INSERT, UPDATE, etc.) or a
- lightweight command like 'SELECT 1' for health checks.
-
- Returns asyncpg status string e.g. 'INSERT 0 1'.
- """
+ """Execute a non-SELECT statement or lightweight command like SELECT 1."""
await self.ensure_ready()
try:
async with self._pool.acquire() as conn:
- return await conn.execute(
- query, *args, timeout=timeout or settings.DB_QUERY_TIMEOUT
- )
+ return await conn.execute(query, *args, timeout=timeout or self._command_timeout)
except Exception as exc:
- logger.exception("Database execute failed")
+ logger.exception("Database execute failed", extra={"pool": self._label})
raise DatabaseError(f"Execute failed: {exc}") from exc
async def fetchrow(
@@ -158,10 +149,7 @@ async def fetchrow(
query: str,
*args: Any,
) -> Optional[dict]:
- """
- Fetch a single row. Returns None if no rows match.
- Useful for INSERT ... RETURNING id queries.
- """
+ """Fetch a single row. Returns None if no rows match."""
await self.ensure_ready()
try:
async with self._pool.acquire() as conn:
@@ -170,15 +158,46 @@ async def fetchrow(
except Exception as exc:
raise DatabaseError(f"Fetchrow failed: {exc}") from exc
+ async def fetch_with_session(
+ self,
+ query: str,
+ setup: ConnectionInit,
+ *,
+ timeout: Optional[float] = None,
+ ) -> List[dict]:
+ """Execute a SELECT with per-connection session setup (Layer 4 read path)."""
+ await self.ensure_ready()
+ start = time.perf_counter()
+ effective_timeout = timeout or self._command_timeout
+ try:
+ async with self._pool.acquire() as conn:
+ await setup(conn)
+ records: List[Record] = await conn.fetch(query, timeout=effective_timeout)
+ rows = [dict(r) for r in records]
+ elapsed_ms = int((time.perf_counter() - start) * 1000)
+ logger.debug(
+ "Governed query executed",
+ extra={"pool": self._label, "rows": len(rows), "elapsed_ms": elapsed_ms},
+ )
+ return rows
+ except asyncpg.TooManyConnectionsError as exc:
+ raise DatabaseError("Database connection pool exhausted") from exc
+ except asyncpg.QueryCanceledError as exc:
+ raise QueryTimeoutError(
+ f"Query timed out after {effective_timeout}s ({self._label} pool)"
+ ) from exc
+ except Exception as exc:
+ logger.exception(
+ "Governed fetch failed",
+ extra={"pool": self._label, "query": query[:200]},
+ )
+ raise DatabaseError(f"Query failed: {exc}") from exc
+
# ── Private helpers ────────────────────────────────────────────────────
@staticmethod
async def _set_type_codecs(conn: asyncpg.Connection) -> None:
- """
- Register custom type codecs on each new connection.
- Converts PostgreSQL numeric → Python float for JSON serialisation.
- Without this, Decimal values cause JSON encoding errors.
- """
+ """Register numeric → float codec for JSON serialisation."""
await conn.set_type_codec(
"numeric",
encoder=str,
@@ -187,7 +206,33 @@ async def _set_type_codecs(conn: asyncpg.Connection) -> None:
format="text",
)
+ @staticmethod
+ async def _init_read_connection(conn: asyncpg.Connection) -> None:
+ await DatabasePool._set_type_codecs(conn)
+ await conn.execute(f"SET statement_timeout = '{settings.STATEMENT_TIMEOUT_MS}ms'")
+ await conn.execute("SET default_transaction_read_only = on")
+
+
+def _read_pool_timeout() -> float:
+ return max(settings.STATEMENT_TIMEOUT_MS / 1000.0, 1.0)
+
+
+# ── Module-level singletons ─────────────────────────────────────────────────
+database_pool = DatabasePool(
+ dsn=settings.DATABASE_URL,
+ label="write",
+ command_timeout=settings.DB_QUERY_TIMEOUT,
+)
+
+read_database_pool = DatabasePool(
+ dsn=settings.DATABASE_URL_RO,
+ label="read",
+ command_timeout=_read_pool_timeout(),
+ init=DatabasePool._init_read_connection,
+)
+
-# ── Module-level singleton ─────────────────────────────────────────────────
-# Imported and used everywhere — pool is initialised in main.py lifespan
-database_pool = DatabasePool()
+async def reset_all_pools() -> None:
+ """Reset module singleton pools — use between pytest async tests."""
+ await database_pool.reset()
+ await read_database_pool.reset()
diff --git a/backend/app/db/schema_loader.py b/backend/app/db/schema_loader.py
index 3fe062d..73bd1fd 100644
--- a/backend/app/db/schema_loader.py
+++ b/backend/app/db/schema_loader.py
@@ -18,15 +18,16 @@
import time
from typing import Any, Dict, List, Optional
+from app.core.config import settings
from app.core.exceptions import SchemaLoadError
-from app.db.pool import database_pool
+from app.db.pool import read_database_pool
logger = logging.getLogger(__name__)
# ── Schema cache ───────────────────────────────────────────────────────────
_schema_cache: Optional[str] = None
_schema_cached_at: float = 0.0
-_CACHE_TTL_SECONDS: float = 300.0 # refresh every 5 minutes
+_CACHE_TTL_SECONDS: float = float(settings.CACHE_TTL_SECONDS)
async def get_schema_for_prompt() -> str:
@@ -61,6 +62,13 @@ async def get_schema_for_prompt() -> str:
raise SchemaLoadError(f"Failed to load schema: {exc}") from exc
+def invalidate_schema_cache() -> None:
+ """Clear in-memory schema cache (e.g. between pytest event loops)."""
+ global _schema_cache, _schema_cached_at
+ _schema_cache = None
+ _schema_cached_at = 0.0
+
+
async def _build_schema_string() -> str:
"""
Build a human-readable schema description for the prompt.
@@ -118,7 +126,7 @@ async def _build_schema_string() -> str:
async def _fetch_tables() -> List[Dict[str, Any]]:
"""Return all user-created tables in the public schema."""
- return await database_pool.fetch(
+ return await read_database_pool.fetch(
"""
SELECT table_name
FROM information_schema.tables
@@ -134,7 +142,7 @@ async def _fetch_columns(table_name: str) -> List[Dict[str, Any]]:
Return columns for a table with their type, nullability,
and any foreign key references.
"""
- return await database_pool.fetch(
+ return await read_database_pool.fetch(
"""
SELECT
c.column_name,
@@ -175,7 +183,7 @@ async def _fetch_sample_values(table_name: str, column_name: str) -> List[Any]:
return []
try:
- rows = await database_pool.fetch(
+ rows = await read_database_pool.fetch(
f"""
SELECT DISTINCT {column_name}
FROM {table_name}
@@ -192,7 +200,7 @@ async def _fetch_sample_values(table_name: str, column_name: str) -> List[Any]:
async def _fetch_views() -> List[Dict[str, Any]]:
"""Return all views in the public schema."""
- return await database_pool.fetch(
+ return await read_database_pool.fetch(
"""
SELECT viewname
FROM pg_views
diff --git a/backend/app/eval/__init__.py b/backend/app/eval/__init__.py
new file mode 100644
index 0000000..a8ac037
--- /dev/null
+++ b/backend/app/eval/__init__.py
@@ -0,0 +1,5 @@
+"""InsightIQ Layer 2 — SQL evaluation (execution accuracy, not string match)."""
+
+from app.eval.grader import load_jsonl, same_result
+
+__all__ = ["load_jsonl", "same_result"]
diff --git a/backend/app/eval/grader.py b/backend/app/eval/grader.py
new file mode 100644
index 0000000..934461e
--- /dev/null
+++ b/backend/app/eval/grader.py
@@ -0,0 +1,99 @@
+"""
+InsightIQ — SQL eval grader (Layer 2)
+======================================
+Grades text-to-SQL on **execution accuracy**: run generated SQL and gold SQL,
+compare result sets. Order-insensitive; ignores column names (values only).
+
+This matches Spider/BIRD-style eval — two syntactically different queries
+that return the same rows are both correct.
+"""
+
+from __future__ import annotations
+
+import json
+from collections import Counter
+from datetime import date, datetime
+from decimal import Decimal
+from pathlib import Path
+from typing import Any, List, Mapping, Sequence
+
+
+def load_jsonl(path: Path) -> List[dict]:
+ """Load a JSONL file; skip blank lines and comments."""
+ rows: List[dict] = []
+ for line in path.read_text(encoding="utf-8").splitlines():
+ line = line.strip()
+ if not line or line.startswith("#"):
+ continue
+ rows.append(json.loads(line))
+ return rows
+
+
+def _normalize_cell(value: Any) -> Any:
+ if value is None:
+ return None
+ if isinstance(value, Decimal):
+ return round(float(value), 6)
+ if isinstance(value, float):
+ return round(value, 6)
+ if isinstance(value, datetime):
+ return value.isoformat()
+ if isinstance(value, date):
+ return value.isoformat()
+ if isinstance(value, str) and value.endswith("+00:00"):
+ return value.replace("+00:00", "Z")
+ return value
+
+
+def _value_sort_key(value: Any) -> tuple:
+ """Stable ordering for heterogeneous SQL cell values."""
+ normalized = _normalize_cell(value)
+ if normalized is None:
+ return (0, "")
+ if isinstance(normalized, bool):
+ return (1, str(normalized))
+ if isinstance(normalized, (int, float)):
+ return (2, f"{normalized:.6f}")
+ if isinstance(normalized, str):
+ return (3, normalized)
+ return (4, str(normalized))
+
+
+def _row_signature(row: Mapping[str, Any]) -> tuple:
+ """Comparable tuple from a row — column-name agnostic, sorted values."""
+ values = (_normalize_cell(v) for v in row.values())
+ return tuple(sorted(values, key=_value_sort_key))
+
+
+def same_result(
+ generated_rows: Sequence[Mapping[str, Any]],
+ gold_rows: Sequence[Mapping[str, Any]],
+) -> bool:
+ """
+ Return True if both result sets are equal (order-insensitive).
+
+ Compares row value-tuples so aliases and column order do not matter.
+ Generated rows may include extra columns (common with LLM output) as long
+ as every gold row's values appear in some generated row (one-to-one).
+ """
+ if len(generated_rows) != len(gold_rows):
+ return False
+ gen_sigs = [_row_signature(r) for r in generated_rows]
+ gold_sigs = [_row_signature(r) for r in gold_rows]
+ used = [False] * len(gen_sigs)
+ for gold_sig in gold_sigs:
+ matched = False
+ for i, gen_sig in enumerate(gen_sigs):
+ if used[i]:
+ continue
+ if Counter(gen_sig) >= Counter(gold_sig):
+ used[i] = True
+ matched = True
+ break
+ if not matched:
+ return False
+ return True
+
+
+def accuracy(passed: int, total: int) -> float:
+ return passed / total if total else 0.0
diff --git a/backend/app/eval/runner.py b/backend/app/eval/runner.py
new file mode 100644
index 0000000..b4ad224
--- /dev/null
+++ b/backend/app/eval/runner.py
@@ -0,0 +1,96 @@
+"""
+Live SQL eval runner — shared by local tests and CI merge gate.
+"""
+
+from __future__ import annotations
+
+import os
+from pathlib import Path
+
+from app.eval.grader import accuracy, load_jsonl, same_result
+
+EVAL_DIR = Path(__file__).resolve().parents[3] / "tests" / "eval"
+PAIRS_PATH = EVAL_DIR / "pairs.jsonl"
+PAIRS_SMOKE_PATH = EVAL_DIR / "pairs_smoke.jsonl"
+RED_TEAM_PATH = EVAL_DIR / "red_team.jsonl"
+
+MIN_EXEC_ACCURACY_FULL = 0.90
+MIN_EXEC_ACCURACY_SMOKE = 0.80
+
+
+def red_team_sql_cases() -> list[str]:
+ return [row["sql"] for row in load_jsonl(RED_TEAM_PATH) if "sql" in row]
+
+
+def require_live_api_key(*, ci: bool = False) -> None:
+ """Skip locally; fail in CI when the secret is missing."""
+ import pytest
+
+ key = os.getenv("ANTHROPIC_API_KEY", "")
+ if not key or key.startswith("sk-ant-ci-placeholder"):
+ msg = "ANTHROPIC_API_KEY required for live SQL eval"
+ if ci:
+ pytest.fail(msg)
+ pytest.skip(msg)
+
+
+async def run_execution_accuracy(
+ pairs_path: Path,
+ min_accuracy: float,
+) -> None:
+ from app.agent.sql_agent import generate_sql, run_readonly, validate_sql
+
+ pairs = load_jsonl(pairs_path)
+ passed = 0
+ failures: list[str] = []
+
+ for pair in pairs:
+ question = pair["q"]
+ gold_sql = pair["gold_sql"]
+ try:
+ gen_sql = await generate_sql(question)
+ except Exception as exc:
+ failures.append(f"{question!r}: generation failed ({exc})")
+ continue
+
+ if validate_sql(gen_sql) == "blocked":
+ failures.append(f"{question!r}: generated SQL blocked by guardrail")
+ continue
+
+ try:
+ gen_rows = await run_readonly(gen_sql)
+ gold_rows = await run_readonly(gold_sql)
+ except Exception as exc:
+ failures.append(f"{question!r}: execution failed ({exc})")
+ continue
+
+ if same_result(gen_rows, gold_rows):
+ passed += 1
+ else:
+ failures.append(
+ f"{question!r}: result mismatch\n"
+ f" gen: {gen_sql[:120]}\n"
+ f" gold: {gold_sql[:120]}"
+ )
+
+ acc = accuracy(passed, len(pairs))
+ if failures:
+ print("\n".join(failures))
+
+ from app.observability.tracing import flush_traces, is_tracing_enabled
+
+ if is_tracing_enabled():
+ flush_traces()
+
+ assert acc >= min_accuracy, (
+ f"exec accuracy {acc:.2f} < {min_accuracy:.2f} "
+ f"({passed}/{len(pairs)} passed, {len(pairs)} Claude calls)"
+ )
+
+
+async def run_gold_sql_pairs(pairs_path: Path = PAIRS_PATH) -> None:
+ from app.agent.sql_agent import run_readonly
+
+ for pair in load_jsonl(pairs_path):
+ rows = await run_readonly(pair["gold_sql"])
+ assert isinstance(rows, list)
diff --git a/backend/app/guardrails/__init__.py b/backend/app/guardrails/__init__.py
new file mode 100644
index 0000000..57c000f
--- /dev/null
+++ b/backend/app/guardrails/__init__.py
@@ -0,0 +1,14 @@
+"""
+InsightIQ — Layer 3 SQL guardrails (AST validation + question scanning)
+"""
+
+from app.guardrails.question import scan_question
+from app.guardrails.sql import USER_BLOCK_MESSAGE, validate_sql_ast
+from app.guardrails.verdict import Verdict
+
+__all__ = [
+ "Verdict",
+ "validate_sql_ast",
+ "scan_question",
+ "USER_BLOCK_MESSAGE",
+]
diff --git a/backend/app/guardrails/question.py b/backend/app/guardrails/question.py
new file mode 100644
index 0000000..06e5065
--- /dev/null
+++ b/backend/app/guardrails/question.py
@@ -0,0 +1,32 @@
+"""Scan natural-language questions for prompt-injection patterns."""
+
+from __future__ import annotations
+
+from app.guardrails.verdict import Verdict
+
+_SUSPICIOUS_PHRASES = (
+ "ignore previous",
+ "ignore all previous",
+ "disregard previous",
+ "drop table",
+ "delete from",
+ "system prompt",
+ "you are now",
+ "; --",
+ "override instructions",
+)
+
+
+def scan_question(question: str) -> Verdict:
+ """The NL question is untrusted input — reject obvious injection attempts."""
+ lowered = question.lower()
+ for phrase in _SUSPICIOUS_PHRASES:
+ if phrase in lowered:
+ return Verdict.block(
+ f"suspicious instruction in question: {phrase!r}",
+ user_message=(
+ "Your question looks like an instruction to bypass safety rules. "
+ "Please ask a normal analytics question."
+ ),
+ )
+ return Verdict.allow()
diff --git a/backend/app/guardrails/sql.py b/backend/app/guardrails/sql.py
new file mode 100644
index 0000000..0830bfc
--- /dev/null
+++ b/backend/app/guardrails/sql.py
@@ -0,0 +1,99 @@
+"""AST-based SQL validation with sqlglot (Layer 3)."""
+
+from __future__ import annotations
+
+import sqlglot
+from sqlglot import exp
+
+from app.guardrails.verdict import USER_BLOCK_MESSAGE, Verdict
+
+# Analytics tables + views only — no users, query_history, or system catalogs
+ALLOWED_TABLES = frozenset(
+ {
+ "customers",
+ "subscriptions",
+ "events",
+ "invoices",
+ "support_tickets",
+ "customer_health",
+ "monthly_mrr",
+ }
+)
+
+PII_COLUMNS = frozenset({"email", "phone", "ssn", "card_number"})
+
+_WRITE_NODES = (
+ exp.Drop,
+ exp.Delete,
+ exp.Update,
+ exp.Insert,
+ exp.Alter,
+ exp.TruncateTable,
+ exp.Create,
+ exp.Merge,
+)
+
+_READ_ROOT_TYPES = (exp.Select, exp.Union)
+
+
+def _table_name(table: exp.Table) -> str:
+ """Normalise table identifier (strip schema/quotes)."""
+ name = table.name
+ return name.lower() if name else ""
+
+
+def validate_sql_ast(sql: str) -> Verdict:
+ """
+ Parse SQL to an AST and enforce read-only, single-statement, allowlisted access.
+
+ Replaces regex ``^SELECT`` checks — CTE-wrapped writes and multi-statement
+ injection are caught by walking the parse tree.
+ """
+ stripped = (sql or "").strip()
+ if not stripped:
+ return Verdict.block("empty sql")
+
+ try:
+ statements = sqlglot.parse(stripped, read="postgres")
+ except Exception as exc:
+ return Verdict.block(f"cannot parse sql: {exc}")
+
+ if len(statements) != 1:
+ return Verdict.block("only one statement allowed")
+
+ tree = statements[0]
+ if not isinstance(tree, _READ_ROOT_TYPES):
+ return Verdict.block("only SELECT is allowed")
+
+ for node in tree.walk():
+ if isinstance(node, _WRITE_NODES):
+ return Verdict.block(
+ f"write operation rejected: {type(node).__name__}",
+ )
+
+ cte_names = {(cte.alias or "").lower() for cte in tree.find_all(exp.CTE) if cte.alias}
+
+ for table in tree.find_all(exp.Table):
+ name = _table_name(table)
+ if not name or name in cte_names:
+ continue
+ if name not in ALLOWED_TABLES:
+ return Verdict.block(
+ f"table not allowed: {name}",
+ user_message=(
+ f"{USER_BLOCK_MESSAGE} " f"Table {name!r} is not available for reporting."
+ ),
+ )
+
+ for column in tree.find_all(exp.Column):
+ col_name = (column.name or "").lower()
+ if col_name in PII_COLUMNS:
+ return Verdict.block(
+ f"protected column: {col_name}",
+ user_message=(
+ "I cannot return protected personal data "
+ f"({col_name}). Try company_name or plan instead."
+ ),
+ )
+
+ return Verdict.allow()
diff --git a/backend/app/guardrails/verdict.py b/backend/app/guardrails/verdict.py
new file mode 100644
index 0000000..84f543f
--- /dev/null
+++ b/backend/app/guardrails/verdict.py
@@ -0,0 +1,26 @@
+"""Guardrail verdict — allow or block with audit + user-facing messages."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+USER_BLOCK_MESSAGE = "I can only run read-only reports on the allowed tables."
+
+
+@dataclass(frozen=True, slots=True)
+class Verdict:
+ allowed: bool
+ reason: str = ""
+ user_message: str = ""
+
+ @classmethod
+ def allow(cls) -> Verdict:
+ return cls(allowed=True)
+
+ @classmethod
+ def block(cls, reason: str, user_message: str | None = None) -> Verdict:
+ return cls(
+ allowed=False,
+ reason=reason,
+ user_message=user_message or USER_BLOCK_MESSAGE,
+ )
diff --git a/backend/app/observability/__init__.py b/backend/app/observability/__init__.py
new file mode 100644
index 0000000..66c228a
--- /dev/null
+++ b/backend/app/observability/__init__.py
@@ -0,0 +1,5 @@
+"""InsightIQ observability — Langfuse tracing (Layer 1)."""
+
+from app.observability.tracing import configure_langfuse, flush_traces, is_tracing_enabled
+
+__all__ = ["configure_langfuse", "flush_traces", "is_tracing_enabled"]
diff --git a/backend/app/observability/tracing.py b/backend/app/observability/tracing.py
new file mode 100644
index 0000000..ccd97ed
--- /dev/null
+++ b/backend/app/observability/tracing.py
@@ -0,0 +1,142 @@
+"""
+InsightIQ — Langfuse tracing (Layer 1 · Observability)
+========================================================
+Traces each query: question → SQL generation → validation → execution.
+
+Uses Langfuse SDK v4 (`observe`, `propagate_attributes`, `get_client`).
+When keys are unset, Langfuse runs in disabled mode — no overhead in CI/local.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from contextlib import contextmanager
+from typing import Any, Callable, Generator, Iterator, List, Optional, TypeVar
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+F = TypeVar("F", bound=Callable[..., Any])
+
+_configured = False
+
+try:
+ from langfuse import observe as observe
+ from langfuse import propagate_attributes as propagate_attributes
+except ImportError: # pragma: no cover — optional in minimal test envs
+
+ def _noop_observe(
+ func: Optional[F] = None,
+ *,
+ name: Optional[str] = None,
+ as_type: Optional[str] = None,
+ **kwargs: Any,
+ ) -> F | Callable[[F], F]:
+ def decorator(fn: F) -> F:
+ return fn
+
+ if func is not None:
+ return func
+ return decorator
+
+ @contextmanager
+ def _noop_propagate(**kwargs: Any) -> Iterator[None]:
+ yield
+
+ observe = _noop_observe
+ propagate_attributes = _noop_propagate
+
+
+def is_tracing_enabled() -> bool:
+ return bool(settings.LANGFUSE_PUBLIC_KEY and settings.LANGFUSE_SECRET_KEY)
+
+
+def configure_langfuse() -> bool:
+ """Push settings into os.environ so Langfuse get_client() picks them up."""
+ global _configured
+ if not is_tracing_enabled():
+ logger.info("Langfuse tracing disabled (no public/secret key)")
+ return False
+ if _configured:
+ return True
+
+ os.environ.setdefault("LANGFUSE_PUBLIC_KEY", settings.LANGFUSE_PUBLIC_KEY)
+ os.environ.setdefault("LANGFUSE_SECRET_KEY", settings.LANGFUSE_SECRET_KEY)
+ os.environ.setdefault("LANGFUSE_BASE_URL", settings.LANGFUSE_BASE_URL)
+ os.environ.setdefault("LANGFUSE_HOST", settings.LANGFUSE_BASE_URL)
+ _configured = True
+ logger.info("Langfuse tracing enabled")
+ return True
+
+
+def flush_traces() -> None:
+ """Drain the Langfuse buffer — call on container shutdown (scale-to-zero)."""
+ if not is_tracing_enabled():
+ return
+ try:
+ from langfuse import get_client
+
+ get_client().flush()
+ logger.info("Langfuse traces flushed")
+ except Exception:
+ logger.exception("Langfuse flush failed")
+
+
+@contextmanager
+def generation_span(
+ name: str,
+ *,
+ input_data: Optional[Any] = None,
+) -> Generator[Any, None, None]:
+ """Context manager for a Langfuse generation span with manual usage updates."""
+ if not is_tracing_enabled():
+ yield None
+ return
+
+ from langfuse import get_client
+
+ langfuse = get_client()
+ with langfuse.start_as_current_observation(
+ name=name,
+ as_type="generation",
+ input=input_data,
+ ) as span:
+ yield span
+
+
+def update_current_trace_cost(cost_usd: float, **extra_metadata: Any) -> None:
+ """Attach per-query USD to the active Langfuse trace."""
+ if not is_tracing_enabled():
+ return
+ try:
+ from langfuse import get_client
+
+ metadata = {"cost_usd": round(cost_usd, 6), **extra_metadata}
+ get_client().update_current_trace(metadata=metadata)
+ except Exception:
+ logger.debug("Could not update Langfuse trace cost", exc_info=True)
+
+
+def trace_tags(
+ *,
+ retries: int,
+ row_count: int,
+ truncated: bool,
+ verdict: str,
+ failure_mode: str | None = None,
+ from_cache: bool = False,
+) -> List[str]:
+ tags: List[str] = []
+ if from_cache:
+ tags.append("cache_hit")
+ if verdict == "blocked":
+ tags.append("blocked")
+ if failure_mode and failure_mode not in ("none", "empty_results"):
+ tags.append("degraded")
+ if retries > 0:
+ tags.append("degraded")
+ if truncated:
+ tags.append("degraded")
+ return tags
diff --git a/backend/app/reliability/__init__.py b/backend/app/reliability/__init__.py
new file mode 100644
index 0000000..c906d2f
--- /dev/null
+++ b/backend/app/reliability/__init__.py
@@ -0,0 +1,7 @@
+"""InsightIQ — Layer 6 reliability (graceful degradation)."""
+
+from app.reliability.answer import answer_safely
+from app.reliability.failure_modes import FailureMode
+from app.reliability.outcome import AgentOutcome
+
+__all__ = ["AgentOutcome", "FailureMode", "answer_safely"]
diff --git a/backend/app/reliability/answer.py b/backend/app/reliability/answer.py
new file mode 100644
index 0000000..7d773bd
--- /dev/null
+++ b/backend/app/reliability/answer.py
@@ -0,0 +1,223 @@
+"""Layer 6 — graceful degradation orchestrator."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from app.core.config import settings
+from app.core.exceptions import DatabaseError, QueryTimeoutError, SQLGenerationError
+from app.cost.meter import CostMeter
+from app.guardrails import scan_question, validate_sql_ast
+from app.reliability.failure_modes import FailureMode
+from app.reliability.messages import (
+ DB_ERROR,
+ DB_TIMEOUT,
+ PROMPT_INJECTION,
+ UNPARSEABLE_SQL,
+)
+from app.reliability.outcome import AgentOutcome
+
+logger = logging.getLogger(__name__)
+
+
+def _meter_snapshot(meter: CostMeter) -> tuple[int, int, float]:
+ return meter.input_tokens, meter.output_tokens, meter.usd
+
+
+async def answer_safely(
+ question: str,
+ user_id: Optional[int] = None,
+ *,
+ meter: CostMeter,
+) -> AgentOutcome:
+ """
+ Run the full query pipeline; return a typed outcome instead of raising.
+
+ Self-heal runs up to ``MAX_RETRIES`` times for parse/execution failures.
+ """
+ from app.agent.sql_agent import (
+ _execute_readonly,
+ _generate_sql,
+ _log_blocked_query,
+ _log_query_history,
+ _serialise_row,
+ _validate_sql,
+ )
+ from app.db.schema_loader import get_schema_for_prompt
+
+ qv = scan_question(question)
+ if not qv.allowed:
+ await _log_blocked_query(question, "", qv.reason, user_id)
+ return AgentOutcome.degraded(
+ FailureMode.PROMPT_INJECTION,
+ PROMPT_INJECTION,
+ question=question,
+ verdict="blocked",
+ )
+
+ schema = await get_schema_for_prompt()
+ from app.agent.sql_agent import _SYSTEM_PROMPT_TEMPLATE
+
+ system = _SYSTEM_PROMPT_TEMPLATE.format(schema=schema)
+ messages: list[dict[str, str]] = [{"role": "user", "content": question}]
+
+ sql: Optional[str] = None
+ explanation = ""
+ last_error: Optional[str] = None
+ retries = 0
+ last_db_timeout = False
+
+ for attempt in range(settings.MAX_RETRIES + 1):
+ if attempt > 0:
+ retries += 1
+ import json
+
+ messages.append(
+ {
+ "role": "assistant",
+ "content": json.dumps({"sql": sql, "explanation": explanation}),
+ }
+ )
+ from app.agent.sql_agent import _RETRY_TEMPLATE
+
+ messages.append(
+ {
+ "role": "user",
+ "content": _RETRY_TEMPLATE.format(error=last_error, failed_sql=sql or ""),
+ }
+ )
+
+ try:
+ sql, explanation, _ = await _generate_sql(system, messages, meter=meter)
+ except SQLGenerationError:
+ if attempt == settings.MAX_RETRIES:
+ inp, out, cost = _meter_snapshot(meter)
+ return AgentOutcome.degraded(
+ FailureMode.UNPARSEABLE_SQL,
+ UNPARSEABLE_SQL,
+ question=question,
+ sql=sql or "",
+ explanation=explanation,
+ retries=retries,
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
+ last_error = "Could not parse Claude response as valid JSON with 'sql' key."
+ continue
+
+ try:
+ _validate_sql(sql)
+ except Exception:
+ verdict = validate_sql_ast(sql or "")
+ await _log_blocked_query(
+ question,
+ sql or "",
+ verdict.reason or "blocked",
+ user_id,
+ meter.usd,
+ )
+ inp, out, cost = _meter_snapshot(meter)
+ return AgentOutcome.degraded(
+ FailureMode.GUARDRAIL_BLOCK,
+ verdict.user_message or verdict.reason,
+ question=question,
+ sql=sql or "",
+ explanation=explanation,
+ retries=retries,
+ verdict="blocked",
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
+
+ try:
+ rows, execution_ms, truncated = await _execute_readonly(sql)
+ rows = [_serialise_row(r) for r in rows]
+ columns = list(rows[0].keys()) if rows else []
+
+ await _log_query_history(
+ question=question,
+ sql=sql,
+ explanation=explanation,
+ rows=rows,
+ columns=columns,
+ row_count=len(rows),
+ execution_ms=execution_ms,
+ retries=retries,
+ user_id=user_id,
+ cost_usd=meter.usd,
+ )
+
+ return AgentOutcome.success(
+ question=question,
+ sql=sql,
+ explanation=explanation,
+ rows=rows,
+ columns=columns,
+ execution_ms=execution_ms,
+ retries=retries,
+ truncated=truncated,
+ input_tokens=meter.input_tokens,
+ output_tokens=meter.output_tokens,
+ cost_usd=meter.usd,
+ )
+
+ except QueryTimeoutError as exc:
+ last_error = str(exc)
+ last_db_timeout = True
+ if attempt == settings.MAX_RETRIES:
+ inp, out, cost = _meter_snapshot(meter)
+ return AgentOutcome.degraded(
+ FailureMode.DB_TIMEOUT,
+ DB_TIMEOUT,
+ question=question,
+ sql=sql or "",
+ explanation=explanation,
+ retries=retries,
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
+
+ except DatabaseError as exc:
+ last_error = str(exc)
+ last_db_timeout = False
+ logger.warning("db error during query execution", extra={"error": last_error})
+ if attempt == settings.MAX_RETRIES:
+ inp, out, cost = _meter_snapshot(meter)
+ return AgentOutcome.degraded(
+ FailureMode.DB_ERROR,
+ DB_ERROR,
+ question=question,
+ sql=sql or "",
+ explanation=explanation,
+ retries=retries,
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
+
+ inp, out, cost = _meter_snapshot(meter)
+ if last_db_timeout:
+ return AgentOutcome.degraded(
+ FailureMode.DB_TIMEOUT,
+ DB_TIMEOUT,
+ question=question,
+ sql=sql or "",
+ retries=retries,
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
+ return AgentOutcome.degraded(
+ FailureMode.UNPARSEABLE_SQL,
+ UNPARSEABLE_SQL,
+ question=question,
+ sql=sql or "",
+ retries=retries,
+ input_tokens=inp,
+ output_tokens=out,
+ cost_usd=cost,
+ )
diff --git a/backend/app/reliability/failure_modes.py b/backend/app/reliability/failure_modes.py
new file mode 100644
index 0000000..bdc160c
--- /dev/null
+++ b/backend/app/reliability/failure_modes.py
@@ -0,0 +1,21 @@
+"""Layer 6 failure modes for tracing and API responses."""
+
+from __future__ import annotations
+
+from enum import Enum
+
+
+class FailureMode(str, Enum):
+ """Production failure taxonomy — each maps to a user message."""
+
+ NONE = "none"
+ PROMPT_INJECTION = "prompt_injection"
+ UNPARSEABLE_SQL = "unparseable_sql"
+ GUARDRAIL_BLOCK = "guardrail_block"
+ DB_TIMEOUT = "db_timeout"
+ DB_ERROR = "db_error"
+ EMPTY_RESULTS = "empty_results"
+
+ def is_degraded(self) -> bool:
+ """Empty results are informational, not degraded."""
+ return self not in (FailureMode.NONE, FailureMode.EMPTY_RESULTS)
diff --git a/backend/app/reliability/messages.py b/backend/app/reliability/messages.py
new file mode 100644
index 0000000..80bc3f8
--- /dev/null
+++ b/backend/app/reliability/messages.py
@@ -0,0 +1,16 @@
+"""User-facing copy for Layer 6 graceful degradation."""
+
+PROMPT_INJECTION = (
+ "I can only run read-only data questions. "
+ "Please ask a normal analytics question about your SaaS metrics."
+)
+
+UNPARSEABLE_SQL = "I couldn't turn that into a safe query. Try rephrasing."
+
+DB_TIMEOUT = "That query took too long — try narrowing the date range or adding filters."
+
+DB_ERROR = "Something went wrong running that query. The team has been notified."
+
+EMPTY_RESULTS = "No matching data for that question."
+
+SCHEMA_UNAVAILABLE = "Database temporarily unavailable. Please try again shortly."
diff --git a/backend/app/reliability/outcome.py b/backend/app/reliability/outcome.py
new file mode 100644
index 0000000..4a6d814
--- /dev/null
+++ b/backend/app/reliability/outcome.py
@@ -0,0 +1,146 @@
+"""Agent outcome — success or graceful degradation (Layer 6)."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import Any, Dict, List
+
+from app.reliability.failure_modes import FailureMode
+from app.reliability.messages import EMPTY_RESULTS
+
+
+@dataclass
+class AgentOutcome:
+ """Result of ``answer_safely`` — never raises for expected failure modes."""
+
+ failure_mode: FailureMode
+ message: str
+ question: str
+ sql: str = ""
+ explanation: str = ""
+ rows: List[dict] = field(default_factory=list)
+ columns: List[str] = field(default_factory=list)
+ row_count: int = 0
+ execution_ms: int = 0
+ retries: int = 0
+ truncated: bool = False
+ input_tokens: int = 0
+ output_tokens: int = 0
+ cost_usd: float = 0.0
+ verdict: str = "allowed"
+ from_cache: bool = False
+
+ @property
+ def is_ok(self) -> bool:
+ return self.failure_mode == FailureMode.NONE
+
+ @property
+ def cacheable(self) -> bool:
+ """Successful answers worth caching (includes empty result sets)."""
+ return self.failure_mode in (FailureMode.NONE, FailureMode.EMPTY_RESULTS)
+
+ @property
+ def status(self) -> str:
+ if self.failure_mode == FailureMode.EMPTY_RESULTS:
+ return "ok"
+ if self.failure_mode != FailureMode.NONE:
+ return "degraded"
+ return "ok"
+
+ def to_trace_output(self) -> Dict[str, Any]:
+ return {
+ "sql": self.sql,
+ "explanation": self.explanation,
+ "row_count": self.row_count,
+ "execution_ms": self.execution_ms,
+ "retries": self.retries,
+ "truncated": self.truncated,
+ "input_tokens": self.input_tokens,
+ "output_tokens": self.output_tokens,
+ "cost_usd": self.cost_usd,
+ "verdict": self.verdict,
+ "failure_mode": self.failure_mode.value,
+ "message": self.message,
+ "status": self.status,
+ "from_cache": self.from_cache,
+ }
+
+ @classmethod
+ def degraded(
+ cls,
+ failure_mode: FailureMode,
+ message: str,
+ *,
+ question: str,
+ sql: str = "",
+ explanation: str = "",
+ retries: int = 0,
+ verdict: str = "blocked",
+ input_tokens: int = 0,
+ output_tokens: int = 0,
+ cost_usd: float = 0.0,
+ ) -> AgentOutcome:
+ return cls(
+ failure_mode=failure_mode,
+ message=message,
+ question=question,
+ sql=sql,
+ explanation=explanation,
+ retries=retries,
+ verdict=verdict,
+ input_tokens=input_tokens,
+ output_tokens=output_tokens,
+ cost_usd=cost_usd,
+ )
+
+ @classmethod
+ def success(
+ cls,
+ *,
+ question: str,
+ sql: str,
+ explanation: str,
+ rows: List[dict],
+ columns: List[str],
+ execution_ms: int,
+ retries: int,
+ truncated: bool,
+ input_tokens: int,
+ output_tokens: int,
+ cost_usd: float,
+ ) -> AgentOutcome:
+ if not rows:
+ return cls(
+ failure_mode=FailureMode.EMPTY_RESULTS,
+ message=EMPTY_RESULTS,
+ question=question,
+ sql=sql,
+ explanation=explanation,
+ rows=[],
+ columns=columns,
+ execution_ms=execution_ms,
+ retries=retries,
+ truncated=truncated,
+ input_tokens=input_tokens,
+ output_tokens=output_tokens,
+ cost_usd=cost_usd,
+ verdict="allowed",
+ )
+
+ return cls(
+ failure_mode=FailureMode.NONE,
+ message="",
+ question=question,
+ sql=sql,
+ explanation=explanation,
+ rows=rows,
+ columns=columns,
+ row_count=len(rows),
+ execution_ms=execution_ms,
+ retries=retries,
+ truncated=truncated,
+ input_tokens=input_tokens,
+ output_tokens=output_tokens,
+ cost_usd=cost_usd,
+ verdict="allowed",
+ )
diff --git a/backend/main.py b/backend/main.py
index a44e7a8..ffc4619 100644
--- a/backend/main.py
+++ b/backend/main.py
@@ -18,7 +18,8 @@
from app.auth.router import router as auth_router
from app.core.config import settings
from app.core.logging import configure_logging
-from app.db.pool import database_pool
+from app.db.pool import database_pool, read_database_pool
+from app.observability import configure_langfuse, flush_traces
configure_logging()
logger = logging.getLogger(__name__)
@@ -27,13 +28,16 @@
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
# Do not block on DB here — ACA scale-from-zero probes /health before Neon wakes.
+ configure_langfuse()
cors_origins = settings.get_all_cors_origins()
logger.info("InsightIQ starting — /health live; DB pool connects on first API call")
logger.info(f"CORS allowed origins: {cors_origins}")
yield
- logger.info("Shutting down — closing database pool...")
+ logger.info("Shutting down — flushing Langfuse traces and closing database pools...")
+ flush_traces()
+ await read_database_pool.close()
await database_pool.close()
@@ -85,8 +89,8 @@ async def health_check() -> dict:
@app.get("/ready", tags=["Health"])
async def readiness_check() -> dict:
"""Readiness — confirms database connectivity."""
- await database_pool.ensure_ready()
- await database_pool.execute("SELECT 1")
+ await read_database_pool.ensure_ready()
+ await read_database_pool.execute("SELECT 1")
return {"status": "ready", "version": "1.0.0"}
diff --git a/backend/requirements.txt b/backend/requirements.txt
index 25aff41..819700f 100644
--- a/backend/requirements.txt
+++ b/backend/requirements.txt
@@ -32,3 +32,6 @@ pytest-mock==3.14.0
# Code quality
ruff==0.6.9
+
+langfuse>=2.0,<5.0
+sqlglot>=25.0 # parse generated SQL into an AST
diff --git a/database/init.sql b/database/init.sql
index 435453d..3c8b400 100644
--- a/database/init.sql
+++ b/database/init.sql
@@ -68,6 +68,36 @@ CREATE TABLE query_history (
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
+CREATE TABLE query_audit (
+ id SERIAL PRIMARY KEY,
+ user_id INTEGER,
+ question TEXT NOT NULL,
+ sql TEXT NOT NULL,
+ verdict VARCHAR(20) NOT NULL CHECK (verdict IN ('allowed', 'blocked')),
+ row_count INTEGER NOT NULL DEFAULT 0,
+ cost_usd NUMERIC(10, 6) NOT NULL DEFAULT 0,
+ ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE TABLE daily_cost_spend (
+ day DATE PRIMARY KEY,
+ usd NUMERIC(12, 6) NOT NULL DEFAULT 0,
+ alerted BOOLEAN NOT NULL DEFAULT FALSE,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_query_audit_ts ON query_audit(ts DESC);
+
+CREATE TABLE query_cache (
+ cache_key VARCHAR(20) PRIMARY KEY,
+ question_norm TEXT NOT NULL,
+ payload JSONB NOT NULL,
+ expires_at TIMESTAMPTZ NOT NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX idx_query_cache_expires ON query_cache(expires_at);
+
CREATE INDEX idx_customers_plan ON customers(plan);
CREATE INDEX idx_customers_active ON customers(is_active);
CREATE INDEX idx_subs_customer ON subscriptions(customer_id);
diff --git a/database/migrations/003_query_audit.sql b/database/migrations/003_query_audit.sql
new file mode 100644
index 0000000..3edda7c
--- /dev/null
+++ b/database/migrations/003_query_audit.sql
@@ -0,0 +1,21 @@
+-- ================================================================
+-- InsightIQ Migration 003 — Layer 4 query audit log
+-- ================================================================
+-- Run:
+-- docker exec -i insightiq_postgres psql \
+-- -U insightiq_user -d insightiq \
+-- < database/migrations/003_query_audit.sql
+
+CREATE TABLE IF NOT EXISTS query_audit (
+ id SERIAL PRIMARY KEY,
+ user_id INTEGER REFERENCES users(id) ON DELETE SET NULL,
+ question TEXT NOT NULL,
+ sql TEXT NOT NULL,
+ verdict VARCHAR(20) NOT NULL CHECK (verdict IN ('allowed', 'blocked')),
+ row_count INTEGER NOT NULL DEFAULT 0,
+ ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_query_audit_user_id ON query_audit(user_id);
+CREATE INDEX IF NOT EXISTS idx_query_audit_ts ON query_audit(ts DESC);
+CREATE INDEX IF NOT EXISTS idx_query_audit_verdict ON query_audit(verdict);
diff --git a/database/migrations/004_readonly_role_grants.sql b/database/migrations/004_readonly_role_grants.sql
new file mode 100644
index 0000000..714fdc9
--- /dev/null
+++ b/database/migrations/004_readonly_role_grants.sql
@@ -0,0 +1,46 @@
+-- ================================================================
+-- InsightIQ Migration 004 — Read-only role grants (Neon / production)
+-- ================================================================
+-- Run once as the database owner (e.g. neondb_owner) after creating
+-- role insightiq_ro. Local Docker dev often uses one superuser for both
+-- DATABASE_URL and DATABASE_URL_RO — this script targets production.
+--
+-- psql "$OWNER_DATABASE_URL" -f database/migrations/004_readonly_role_grants.sql
+
+-- ── Role (skip if already created) ───────────────────────────────────────
+DO $$
+BEGIN
+ IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'insightiq_ro') THEN
+ CREATE ROLE insightiq_ro LOGIN PASSWORD 'CHANGE_ME';
+ END IF;
+END
+$$;
+
+GRANT CONNECT ON DATABASE neondb TO insightiq_ro;
+GRANT USAGE ON SCHEMA public TO insightiq_ro;
+
+-- ── Table SELECT (column-level — PII excluded) ───────────────────────────
+GRANT SELECT (
+ id, company_name, plan, country, industry, employee_count,
+ health_score, created_at, churned_at, is_active
+) ON customers TO insightiq_ro;
+-- email intentionally omitted — blocked in Layer 3 AND not granted here
+
+GRANT SELECT ON subscriptions TO insightiq_ro;
+GRANT SELECT ON events TO insightiq_ro;
+GRANT SELECT ON invoices TO insightiq_ro;
+GRANT SELECT ON support_tickets TO insightiq_ro;
+
+-- Analytics views only — no query_history, users, or system catalogs
+GRANT SELECT ON customer_health TO insightiq_ro;
+GRANT SELECT ON monthly_mrr TO insightiq_ro;
+
+-- ── Hard deny writes (belt + suspenders with role default) ────────────────
+REVOKE INSERT, UPDATE, DELETE, TRUNCATE, REFERENCES, TRIGGER
+ ON ALL TABLES IN SCHEMA public FROM insightiq_ro;
+
+-- ── Verify ───────────────────────────────────────────────────────────────
+SELECT grantee, table_name, privilege_type
+FROM information_schema.role_table_grants
+WHERE grantee = 'insightiq_ro'
+ORDER BY table_name, privilege_type;
diff --git a/database/migrations/005_daily_cost.sql b/database/migrations/005_daily_cost.sql
new file mode 100644
index 0000000..82cae3f
--- /dev/null
+++ b/database/migrations/005_daily_cost.sql
@@ -0,0 +1,17 @@
+-- ================================================================
+-- InsightIQ Migration 005 — Layer 5 daily cost ledger
+-- ================================================================
+-- Run:
+-- docker exec -i insightiq_postgres psql \
+-- -U insightiq_user -d insightiq \
+-- < database/migrations/005_daily_cost.sql
+
+CREATE TABLE IF NOT EXISTS daily_cost_spend (
+ day DATE PRIMARY KEY,
+ usd NUMERIC(12, 6) NOT NULL DEFAULT 0,
+ alerted BOOLEAN NOT NULL DEFAULT FALSE,
+ updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+ALTER TABLE query_audit
+ ADD COLUMN IF NOT EXISTS cost_usd NUMERIC(10, 6);
diff --git a/database/migrations/006_query_cache.sql b/database/migrations/006_query_cache.sql
new file mode 100644
index 0000000..c3d4d32
--- /dev/null
+++ b/database/migrations/006_query_cache.sql
@@ -0,0 +1,17 @@
+-- ================================================================
+-- InsightIQ Migration 006 — normalised question cache (optional layer)
+-- ================================================================
+-- Run:
+-- docker exec -i insightiq_postgres psql \
+-- -U insightiq_user -d insightiq \
+-- < database/migrations/006_query_cache.sql
+
+CREATE TABLE IF NOT EXISTS query_cache (
+ cache_key VARCHAR(20) PRIMARY KEY,
+ question_norm TEXT NOT NULL,
+ payload JSONB NOT NULL,
+ expires_at TIMESTAMPTZ NOT NULL,
+ created_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_query_cache_expires ON query_cache(expires_at);
diff --git a/docs/cost.md b/docs/cost.md
new file mode 100644
index 0000000..db17c13
--- /dev/null
+++ b/docs/cost.md
@@ -0,0 +1,45 @@
+# Layer 5 · Cost observability
+
+Per-query Claude spend is recorded in **Langfuse** (`cost_usd` on each `insightiq_query` trace) and in the `query_audit` table.
+
+## Langfuse cost dashboard
+
+1. Open [Langfuse](https://cloud.langfuse.com) → your InsightIQ project.
+2. Go to **Traces** and filter by name `insightiq_query`.
+3. Open any trace → check metadata `cost_usd`, `input_tokens`, `output_tokens`.
+4. For aggregate cost: **Analytics** → group by trace name or use the cost column if enabled on your plan.
+
+Save a screenshot as `docs/cost.png` for your portfolio README.
+
+## Typical cost per query
+
+At claude-sonnet-4-5 rates ($3/MTok in, $15/MTok out):
+
+| Scenario | Tokens (in / out) | Cost |
+|----------|-------------------|------|
+| Measured (full schema prompt) | ~10,500 / ~130 | **~$0.03** |
+| Optimized (caching / slim schema) | ~1,000 / ~130 | **~$0.004** |
+
+Retries add linearly — a 3-attempt heal costs up to ~3× that.
+
+## Daily budget alert
+
+Set in production:
+
+| Variable | Default | Purpose |
+|----------|---------|---------|
+| `BUDGET_USD_DAILY` | `2.00` | In-app daily spend cap (alert threshold) |
+| `COST_ALERT_WEBHOOK_URL` | _(empty)_ | Slack-compatible webhook; one alert per UTC day |
+
+The GitHub **cost-monitor.yml** workflow still checks Anthropic’s billing API at 09:00 UTC as a second line of defense.
+
+## SQL: average cost from audit log
+
+```sql
+SELECT
+ ROUND(AVG(cost_usd)::numeric, 6) AS avg_cost_usd,
+ COUNT(*) AS queries,
+ ROUND(SUM(cost_usd)::numeric, 4) AS total_usd
+FROM query_audit
+WHERE verdict = 'allowed' AND cost_usd > 0;
+```
diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx
index 0f98809..3fb4997 100644
--- a/frontend/src/App.tsx
+++ b/frontend/src/App.tsx
@@ -74,7 +74,20 @@ const Dashboard: React.FC = () => {
?
+ {result.message} +
+