SQL review, optimization & async jobs (implementation log)

Roadmap: FUTURE_PHASES.md Status: ✅ Complete — 1–5 done
Operator guide: Review hold + async jobs below (2–5)

Goal

High-risk or expensive SQL is gated, optimized, or run asynchronously with cancel support.

Step checklist

Step	Task	Status	Notes
1	`SqlReviewDecider` + store (app DB)	✅ Done	Heuristics + `sql_reviews` table
2	Chat returns `review_required` without rows when pending	✅ Done	Ask + admin approve/reject API
3	`QueryCostController` port + dialect hints	✅ Done	Heuristics + optional EXPLAIN
4	Job store + worker + API routes	✅ Done	`POST/GET/POST cancel` `/chat/jobs`
5	SSE progress events for jobs	✅ Done	`GET /chat/jobs/{id}/stream`

Step 1 — SqlReviewDecider + app DB store (complete)

Risk heuristics score generated SQL before execution. When risk meets the threshold and INSIGHTAI_SQL_REVIEW_ENABLED=true, a row is inserted into the SQL review queue (sql_reviews table). Step 2 gates chat/ask execution until a reviewer approves (or permanently blocks rejected SQL).

Heuristics

Code	Trigger	Typical risk
`cross_join`	Comma join or `CROSS JOIN` without `ON`	High
`missing_where`	No `WHERE` clause	Medium (High when combined)
`no_row_cap`	No `TOP` / `LIMIT` / `FETCH`	Medium (High when combined)
`sql_parse_failed`	sqlglot cannot parse SQL	Medium

Aggregate risk: cross join → high; two or more reasons → high; single reason → medium.

Review required when risk_level >= INSIGHTAI_SQL_REVIEW_RISK_THRESHOLD (default medium).

Settings

Variable	Default	Description
`INSIGHTAI_SQL_REVIEW_ENABLED`	`false`	Persist pending reviews when threshold met
`INSIGHTAI_SQL_REVIEW_FLAG_MISSING_WHERE`	`true`	Enable missing-WHERE heuristic
`INSIGHTAI_SQL_REVIEW_FLAG_CROSS_JOIN`	`true`	Enable cross-join heuristic
`INSIGHTAI_SQL_REVIEW_FLAG_NO_ROW_CAP`	`true`	Enable no row-cap heuristic
`INSIGHTAI_SQL_REVIEW_RISK_THRESHOLD`	`medium`	`low` \| `medium` \| `high`

Domain & infrastructure

src/insightai/domain/models/sql_review.py
src/insightai/domain/ports/sql_review_decider.py
src/insightai/domain/ports/sql_review_store.py
src/insightai/infrastructure/sql_review/decider.py      # SqlReviewDecider
src/insightai/infrastructure/sql_review/sql_hash.py
src/insightai/infrastructure/sql_review/bootstrap.py
src/insightai/infrastructure/app_db/models/sql_reviews.py
src/insightai/infrastructure/app_db/sql_review_store.py
src/insightai/application/use_cases/evaluate_sql_review.py
alembic/versions/005_sql_reviews_table.py

Roles

Role	Purpose (2+)
`sql_reviewer`	Approve/reject pending SQL (`PlatformRole.SQL_REVIEWER`)
`admin`	Full admin API access

Create reviewer key:

insightai-keys create --label "SQL reviewer" --roles sql_reviewer

Use case (library / tests)

from insightai.application.use_cases.evaluate_sql_review import (
    EvaluateSqlReviewRequest,
    EvaluateSqlReviewUseCase,
)
from insightai.infrastructure.sql_review.bootstrap import build_sql_review_decider

result = use_case.execute(
    EvaluateSqlReviewRequest(
        sql="SELECT * FROM dbo.accounts_user",
        question="List all users",
        auth_subject="analyst-key",
    ),
)
# result.review_required, result.review_id, result.decision.risk_level

When review is required, duplicate SQL (same hash) reuses an existing pending row.

Migration

insightai-app-db upgrade
insightai-app-db current   # expect 005_sql_reviews

Verify

.venv/bin/python -m pytest \
  tests/unit/test_sql_review_models.py \
  tests/unit/test_sql_review_decider.py \
  tests/unit/test_sql_review_store.py \
  tests/unit/test_evaluate_sql_review_use_case.py \
  tests/unit/test_app_db_alembic.py -q

Step 2 — Chat/ask review hold + admin queue (complete)

When INSIGHTAI_SQL_REVIEW_ENABLED=true, high-risk SQL is scored after generation and before governance validation / DB execution.

Behaviour

State	Chat / ask response	DB execution
Low risk	Normal answer + rows	Yes
Pending review	HTTP 200, `review_required: true`, `row_count: 0`, stub answer	No
Rejected (same SQL hash)	HTTP 200, `review_required: true`, rejection message	No
Approved (same SQL hash)	Normal answer + rows	Yes

Response fields (chat and debug /ask):

review_required, review_id, sql_review_status, sql_risk_level, sql_risk_reasons
sql still returned when include_sql=true (or always on /ask) so reviewers can inspect the query

Streaming chat emits status phase review_required then done with the same JSON shape as non-streaming.

Admin API

Requires API key with sql_reviewer or admin role:

Method	Path	Description
`GET`	`/api/v1/admin/sql-reviews?status=pending`	List queue
`POST`	`/api/v1/admin/sql-reviews/{id}/approve`	Allow identical SQL to execute
`POST`	`/api/v1/admin/sql-reviews/{id}/reject`	Block identical SQL

Optional body: { "reviewer_note": "..." }.

Wiring

src/insightai/application/use_cases/ask.py          # _maybe_hold_for_sql_review
src/insightai/api/deps.py                           # EvaluateSqlReviewUseCase → AskUseCase
src/insightai/api/schemas/chat.py                   # review_required fields
src/insightai/api/schemas/ask.py                    # review hold debug response
src/insightai/api/v1/routes/admin.py                # sql-reviews routes
src/insightai/api/auth/sql_reviewer.py              # require_sql_reviewer_role
src/insightai/application/use_cases/list_sql_reviews.py
src/insightai/application/use_cases/approve_sql_review.py
src/insightai/application/use_cases/reject_sql_review.py

Verify

# Enable review + migrate app DB
export INSIGHTAI_SQL_REVIEW_ENABLED=true
insightai-app-db upgrade

# Create reviewer key
insightai-keys create --label "SQL reviewer" --roles sql_reviewer

.venv/bin/python -m pytest \
  tests/unit/test_ask_sql_review.py \
  tests/unit/test_evaluate_sql_review_use_case.py \
  tests/integration/test_chat_sql_review.py -q

Step 3 — QueryCostController + dialect hints (complete)

Pre-execution cost gate that extends row caps and timeouts. Runs after SQL validation in RunQueryUseCase and before the database round-trip.

Behaviour

Mode	On threshold exceeded
`reject` (default)	Raises `QueryCostExceededError` → HTTP 400
`warn`	Logs `query_cost_warning` and executes anyway

Heuristics (always when enabled):

Code	Trigger
`too_many_table_refs`	Distinct tables exceed `INSIGHTAI_QUERY_COST_MAX_TABLE_REFS`
`deep_subqueries`	Nested subquery depth exceeds `INSIGHTAI_QUERY_COST_MAX_SUBQUERY_DEPTH`
`unbounded_sort`	`ORDER BY` without `WHERE` or row cap in SQL
`sql_parse_failed`	sqlglot cannot parse SQL

Optional dialect EXPLAIN hints (INSIGHTAI_QUERY_COST_EXPLAIN_ENABLED=true):

Dialect	Mechanism
SQLite	`EXPLAIN QUERY PLAN` — flags `SCAN TABLE`
PostgreSQL	`EXPLAIN (FORMAT JSON)` — flags `Seq Scan`, high `Plan Rows`
MSSQL	Heuristics only (no EXPLAIN in 18.3)

Codes from EXPLAIN: full_table_scan, high_estimated_rows.

Settings

Variable	Default	Description
`INSIGHTAI_QUERY_COST_ENABLED`	`false`	Enable pre-execution cost gate
`INSIGHTAI_QUERY_COST_ON_EXCEEDED`	`reject`	`reject` \| `warn`
`INSIGHTAI_QUERY_COST_EXPLAIN_ENABLED`	`false`	Run dialect EXPLAIN before execute
`INSIGHTAI_QUERY_COST_MAX_TABLE_REFS`	`4`	Heuristic table-ref cap
`INSIGHTAI_QUERY_COST_MAX_SUBQUERY_DEPTH`	`3`	Heuristic subquery depth cap
`INSIGHTAI_QUERY_COST_MAX_ESTIMATED_ROWS`	`500000`	EXPLAIN planner row estimate cap

Wiring

src/insightai/domain/models/query_cost.py
src/insightai/domain/ports/query_cost_controller.py
src/insightai/infrastructure/query_cost/controller.py
src/insightai/infrastructure/query_cost/heuristics.py
src/insightai/infrastructure/query_cost/explain_hints.py
src/insightai/infrastructure/query_cost/bootstrap.py
src/insightai/application/use_cases/run_query.py   # _enforce_query_cost
src/insightai/infrastructure/database/bootstrap.py # builds controller with engine

Verify

export INSIGHTAI_QUERY_COST_ENABLED=true

.venv/bin/python -m pytest \
  tests/unit/test_query_cost_heuristics.py \
  tests/unit/test_query_cost_controller.py \
  tests/unit/test_query_cost_explain_hints.py -q

Step 4 — Async query jobs (complete)

Long-running chat/ask requests can be enqueued for background processing by the in-process asyncio worker.

API

Requires INSIGHTAI_QUERY_JOBS_ENABLED=true and API key auth.

Method	Path	Description
`POST`	`/api/v1/chat/jobs`	Enqueue question → 202 with `job.id`
`GET`	`/api/v1/chat/jobs/{id}`	Poll status, phase, result, or error
`POST`	`/api/v1/chat/jobs/{id}/cancel`	Cancel pending or request cancel on running job

Request body matches POST /api/v1/chat (ChatRequest). Poll until status is completed, failed, or cancelled.

Worker

When INSIGHTAI_QUERY_JOBS_WORKER_ENABLED=true (default), the API process runs an asyncio loop that:

Claims the oldest pending job (app DB row lock)
Replays the stored AskRequest JSON through the full ask pipeline
Persists a result snapshot (answer, row_count, sql, review_required, …)
Records sql_hash for audit linkage

Cancel is best effort for running jobs: the worker checks cancel_requested_at before starting pipeline execution.

Settings

Variable	Default	Description
`INSIGHTAI_QUERY_JOBS_ENABLED`	`false`	Enable async job API + store
`INSIGHTAI_QUERY_JOBS_WORKER_ENABLED`	`true`	In-process worker in API lifespan
`INSIGHTAI_QUERY_JOBS_POLL_INTERVAL_SECONDS`	`1.0`	Idle poll interval

Wiring

src/insightai/domain/models/query_job.py
src/insightai/domain/ports/query_job_store.py
src/insightai/infrastructure/app_db/models/query_jobs.py
src/insightai/infrastructure/app_db/query_job_store.py
src/insightai/infrastructure/query_jobs/worker.py
src/insightai/application/factory/ask_pipeline.py
src/insightai/api/v1/routes/chat_jobs.py
alembic/versions/006_query_jobs_table.py

Migration

insightai-app-db upgrade
insightai-app-db current   # expect 006_query_jobs

Verify

export INSIGHTAI_QUERY_JOBS_ENABLED=true

.venv/bin/python -m pytest \
  tests/unit/test_query_job_store.py \
  tests/integration/test_chat_jobs_api.py \
  tests/unit/test_app_db_alembic.py -q

Step 5 — SSE progress events for jobs (complete)

Clients can subscribe to job progress instead of polling JSON.

API

Method	Path	Description
`GET`	`/api/v1/chat/jobs/{id}/stream`	SSE: `status` → `done` or `error`

Events mirror chat streaming:

Event	Payload
`status`	`{ "phase": "sql_generation", "status": "running" }`
`done`	Full `ChatJobResponse` when `status=completed` or `cancelled`
`error`	`{ "error_code", "error_message", "job_id" }` when `status=failed`

Phases: queued, sql_generation, query_execution, answer_generation, review_required, done.

The worker publishes phase updates through an in-process event broker; the stream endpoint also polls the app DB on timeout so late subscribers still catch up.

Settings

Variable	Default	Description
`INSIGHTAI_QUERY_JOBS_STREAM_POLL_INTERVAL_SECONDS`	`0.5`	SSE fallback poll interval

Wiring

src/insightai/infrastructure/query_jobs/events.py      # QueryJobEventBroker
src/insightai/infrastructure/query_jobs/stream.py      # stream_query_job_events
src/insightai/infrastructure/query_jobs/phase_map.py
src/insightai/application/use_cases/ask.py             # on_phase callback
src/insightai/api/v1/routes/chat_jobs.py               # GET .../stream

Verify

export INSIGHTAI_QUERY_JOBS_ENABLED=true

.venv/bin/python -m pytest \
  tests/unit/test_query_job_sse.py \
  tests/integration/test_chat_jobs_stream_api.py -q

Document history

Date	Change
2026-05-31	Step 5 — SSE job progress stream
2026-05-31	Step 1 — SqlReviewDecider, sql_reviews store, EvaluateSqlReviewUseCase

FilesExpand file tree

SQL_REVIEW.md

Latest commit

History

SQL_REVIEW.md

File metadata and controls

SQL review, optimization & async jobs (implementation log)

Goal

Step checklist

Step 1 — SqlReviewDecider + app DB store (complete)

Heuristics

Settings

Domain & infrastructure

Roles

Use case (library / tests)

Migration

Verify

Step 2 — Chat/ask review hold + admin queue (complete)

Behaviour

Admin API

Wiring

Verify

Step 3 — QueryCostController + dialect hints (complete)

Behaviour

Settings

Wiring

Verify

Step 4 — Async query jobs (complete)

API

Worker

Settings

Wiring

Migration

Verify

Step 5 — SSE progress events for jobs (complete)

API

Settings

Wiring

Verify

Document history