feat(AF-624): pre-flight query cost & blast-radius estimation

CLAUDE.md

@@ -286,6 +286,7 @@ com.bablsoft.accessflow/
 | `ACCESSFLOW_PROXY_IDLE_TIMEOUT` | HikariCP `idleTimeout` (default `10m`). |
 | `ACCESSFLOW_PROXY_MAX_LIFETIME` | HikariCP `maxLifetime` (default `30m`). |
 | `ACCESSFLOW_PROXY_LEAK_DETECTION_THRESHOLD` | HikariCP leak-detection threshold (default `0s` = disabled). |
+| `ACCESSFLOW_PROXY_ESTIMATE_TIMEOUT` | ISO-8601 duration. Statement timeout for the automatic pre-flight cost estimate (AF-624) — the async EXPLAIN + affected-row COUNT run after each submission use this tighter bound instead of the full execution statement timeout (default `PT5S`). |
 | `ACCESSFLOW_PROXY_EXECUTION_MAX_ROWS` | Hard cap on rows returned by a single query (default `10000`). |
 | `ACCESSFLOW_PROXY_EXECUTION_STATEMENT_TIMEOUT` | Statement-level timeout applied to customer-DB JDBC statements (default `30s`). |
 | `ACCESSFLOW_PROXY_EXECUTION_DEFAULT_FETCH_SIZE` | Default JDBC fetch size (default `1000`). |
@@ -861,18 +862,20 @@ The `AiAnalyzerStrategy` interface must be implemented by all three adapters:
 ```java
 public interface AiAnalyzerStrategy {
     AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
-                             String language, UUID organizationId);
+                             String costEstimateContext, String language, UUID aiConfigId);
 }
 ```
 
 `schemaContext` may be `null` or empty when introspection is unavailable; the prompt template
-substitutes `(no schema introspection available)` in that case.
+substitutes `(no schema introspection available)` in that case. `costEstimateContext` (AF-624)
+carries the same contract for the pre-flight cost estimate — `null`/empty substitutes
+`(no cost estimate available)`.
 
 - Adapters route their HTTP calls through **Spring AI 2.0** (`spring-ai-bom:2.0.0`). The autowired `AiAnalyzerStrategy` is `AiAnalyzerStrategyHolder` — it builds `AnthropicChatModel` / `OpenAiChatModel` / `OllamaChatModel` per-org from the `ai_config` row, caches the delegate, and evicts on `AiConfigUpdatedEvent` (no Spring context refresh, no restart).
 - The same Spring AI BOM also pins `spring-ai-starter-mcp-server-webmvc` (`2.0.0`) — the stateless MCP server starter used by the `mcp` module. The starter ships with `spring-ai-autoconfigure-mcp-server-common` as an explicit dependency in `backend/pom.xml` to work around a missing transitive in the starter POM (`McpServerStdioDisabledCondition` lives in the common artifact). See `docs/13-mcp.md` and `docs/05-backend.md` → "MCP server and user API keys".
 - Active provider per org is the `ai_config.provider` column. There is no `accessflow.ai.provider` property and no `@ConditionalOnProperty` on the strategy classes — they are plain classes, not Spring beans.
 - Connection settings (API key, base URL, model, max-tokens, timeout) come from `ai_config`, not from `spring.ai.*` properties. `application.yml` sets `spring.ai.model.{chat,embedding,image,audio.speech,audio.transcription,moderation}=none` so no `ChatModel` is auto-built at startup.
-- The default system prompt template (`SystemPromptRenderer.DEFAULT_TEMPLATE`) is in `docs/05-backend.md` — use it verbatim; do not invent a different prompt. It uses named placeholders `{{db_type}}`, `{{schema_context}}`, `{{sql}}`, `{{language}}` (substituted at render time; `{{sql}}` replaced last). Admins may override the prompt **per `ai_config` row** via the nullable `system_prompt_template` column (blank ⇒ default); a custom template **must contain `{{sql}}`** or the service throws `AiConfigInvalidPromptException` (HTTP 400 `AI_CONFIG_INVALID_PROMPT`). The default is served to the admin UI via `GET /admin/ai-configs/prompt-default`.
+- The default system prompt template (`SystemPromptRenderer.DEFAULT_TEMPLATE`) is in `docs/05-backend.md` — use it verbatim; do not invent a different prompt. It uses named placeholders `{{db_type}}`, `{{schema_context}}`, `{{rag_context}}`, `{{cost_estimate}}` (AF-624 — the pre-flight estimate summary), `{{sql}}`, `{{language}}` (substituted at render time; `{{sql}}` replaced last). Admins may override the prompt **per `ai_config` row** via the nullable `system_prompt_template` column (blank ⇒ default); a custom template **must contain `{{sql}}`** or the service throws `AiConfigInvalidPromptException` (HTTP 400 `AI_CONFIG_INVALID_PROMPT`). The default is served to the admin UI via `GET /admin/ai-configs/prompt-default`.
 - AI calls are asynchronous — publish a `QuerySubmittedEvent` and handle in the strategy asynchronously using virtual threads.
 - The response must be parsed strictly as JSON matching the `AiAnalysisResult` schema. If the AI returns non-JSON or an unexpected schema, log and mark the analysis as failed; do not propagate the exception to the query request.
 - For Anthropic: use `claude-sonnet-4-20250514` as the default model.

README.md

@@ -77,6 +77,7 @@ A glance at the day-to-day flows engineers and approvers actually use.
 - **Multi-model orchestration, voting & guardrails** — an AI configuration can run several models in parallel (the primary plus weighted `{provider, model, weight}` members — e.g. a fast local Ollama pre-score alongside a deep Claude analysis) and combine their verdicts by a configurable voting strategy (**weighted average**, **highest risk**, or **majority**); issues are merged and risk aggregated. **Guardrails** block configured prompt patterns (case-insensitive regex) before any model is called, and the strict output-schema validation rejects malformed responses without failing the query. Per-model **cost (tokens) and latency** are recorded for every analysis and charted on the admin dashboard.
 - **Query-optimization suggestions** — alongside the risk verdict, the analyzer returns concrete, dialect-aware optimization suggestions: index recommendations (`CREATE INDEX …`) and query rewrites. Each suggestion has a one-click **"Apply as draft"** that pre-fills the editor with the suggested statement and routes it through the normal review pipeline — nothing auto-executes, and the draft is audited as AI-suggested.
 - **Query playground / dry-run sandbox** — preview a query's impact before submitting it for review. A **Dry run** action returns the engine's execution plan (node type, estimated rows, cost, filters) and a best-effort estimated row impact **without executing or mutating data**, shown in the editor next to the AI panel. It runs through the same governance as a real execution (datasource access, table allow-list, row-level security) but creates no review. Dialect-aware across every engine with a plan concept — relational `EXPLAIN` (PostgreSQL / MySQL / MariaDB / Oracle / SQL Server), MongoDB `explain`, Couchbase / Neo4j `EXPLAIN`, Elasticsearch / OpenSearch query validation; engines without one degrade gracefully. SELECT dry-runs prefer the read replica when configured.
+- **Pre-flight cost & blast-radius estimation (AF-624)** — every submitted query automatically gets a persisted cost estimate before review: the engine's own `EXPLAIN` plan (estimated rows, scan type, cost) plus, for UPDATE/DELETE, an **exact affected-row count** computed with a governed, non-mutating count (relational `SELECT COUNT(*)` rewrite; MongoDB `countDocuments`, SQL++ `COUNT(*)`, Cypher `count(*)`, Elasticsearch `_count`). Reviewers see it on the query detail page ("this DELETE touches ~2.4M rows via Seq Scan"), routing policies can match on `estimated_rows` / `scan_type` (e.g. auto-escalate full-scan DELETEs over 100k rows), and the estimate is folded into the AI analyzer's prompt so risk scoring reflects the actual blast radius. Engines without a plan concept degrade gracefully to an "estimate unavailable" state that never blocks submission.
 - **Text-to-query** — opt-in per datasource. Users describe what they want in plain language ("order numbers for the last 5 days") and the AI drafts a schema-grounded query into the editor — in the datasource engine's **native query language** (SQL for relational engines, plus MongoDB shell/JSON, Cypher, CQL, Elasticsearch Query DSL, redis-cli, SQL++, and PartiQL for the NoSQL engines). Reuses the datasource's AI configuration and is grounded in its introspected schema (restricted columns are never referenced). The generated query is only a draft — it's still submitted through the normal pipeline, so AI risk analysis and human review always apply.
 - **RAG knowledge base** — opt-in per AI configuration. Attach knowledge documents (data-governance policies, naming conventions, schema notes); AccessFlow embeds them with a dedicated embedding model and, at analysis / text-to-SQL time, retrieves the most relevant chunks and injects them into the prompt — so analysis reflects your house rules. Pluggable vector store: in-app **pgvector** (self-contained, auto-provisioned where the DB role permits — and optional: AccessFlow still starts if the extension is absent, with the in-app store disabled) or external **Qdrant**. Retrieval is best-effort and never blocks analysis.
 - **Langfuse integration** — optional, per-organization. Send a trace of every AI analysis (input SQL, structured output, model, token usage, latency) to [Langfuse](https://langfuse.com) for LLM observability, and/or fetch analyzer prompts from Langfuse by name at runtime (prompt management) so you can iterate on prompts without redeploying. Best-effort and non-blocking — a Langfuse outage never affects query workflow.

backend/src/main/java/com/bablsoft/accessflow/ai/api/AiAnalyzerStrategy.java

@@ -16,7 +16,9 @@ public interface AiAnalyzerStrategy {
     /**
      * Analyze {@code sql} for risks, returning a structured result. The {@code schemaContext}
      * argument is an opaque, provider-renderable description of the target schema (or
-     * {@code null}/empty if introspection is unavailable).
+     * {@code null}/empty if introspection is unavailable). {@code costEstimateContext} (AF-624)
+     * carries the same contract for the pre-flight cost / blast-radius estimate — an opaque,
+     * provider-renderable summary (or {@code null}/empty when no estimate is available).
      *
      * <p>{@code language} is a BCP-47 code (e.g. {@code "en"}, {@code "es"}, {@code "zh-CN"}) that
      * tells the AI which language to use for free-form fields. When {@code null} or unrecognised
@@ -30,8 +32,14 @@ public interface AiAnalyzerStrategy {
      * @throws com.bablsoft.accessflow.ai.api.AiAnalysisParseException provider response did not
      *                                                                match the expected schema
      */
-    AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext, String language,
-                             UUID aiConfigId);
+    AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
+                             String costEstimateContext, String language, UUID aiConfigId);
+
+    /** Convenience overload for callers with no cost-estimate context. */
+    default AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
+                                     String language, UUID aiConfigId) {
+        return analyze(sql, dbType, schemaContext, null, language, aiConfigId);
+    }
 
     /**
      * Translate the natural-language {@code prompt} into a single SQL statement for the target

backend/src/main/java/com/bablsoft/accessflow/ai/internal/AiAnalyzerStrategyHolder.java

@@ -85,18 +85,19 @@ class AiAnalyzerStrategyHolder implements AiAnalyzerStrategy {
             what a reviewer should check. Do not use markdown, headings, or bullet points.""";
 
     @Override
-    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext, String language,
-                                    UUID aiConfigId) {
+    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
+                                    String costEstimateContext, String language, UUID aiConfigId) {
         if (aiConfigId == null) {
             throw notConfigured();
         }
         try {
-            return delegateFor(aiConfigId).analyze(sql, dbType, schemaContext, language, aiConfigId);
+            return delegateFor(aiConfigId).analyze(sql, dbType, schemaContext, costEstimateContext,
+                    language, aiConfigId);
         } catch (AiAnalysisException | AiAnalysisParseException primary) {
             rethrowIfNotFallbackEligible(primary);
             return failover(aiConfigId, primary, fallback ->
-                    delegateFor(fallback.getId()).analyze(sql, dbType, schemaContext, language,
-                            fallback.getId()));
+                    delegateFor(fallback.getId()).analyze(sql, dbType, schemaContext,
+                            costEstimateContext, language, fallback.getId()));
         }
     }
 

backend/src/main/java/com/bablsoft/accessflow/ai/internal/AnthropicAnalyzerStrategy.java

@@ -42,11 +42,11 @@ class AnthropicAnalyzerStrategy implements AiAnalyzerStrategy {
     private final RagRetriever ragRetriever;
 
     @Override
-    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext, String language,
-                                    UUID aiConfigId) {
+    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
+                                    String costEstimateContext, String language, UUID aiConfigId) {
         var ragContext = ragRetriever.retrieve(sql);
         var userPrompt = promptRenderer.render(promptSource.template(), sql, dbType, schemaContext,
-                ragContext, language);
+                ragContext, costEstimateContext, language);
         var prompt = new Prompt(List.of(
                 new SystemMessage(SYSTEM_PROMPT_PREAMBLE),
                 new UserMessage(userPrompt)));

backend/src/main/java/com/bablsoft/accessflow/ai/internal/DefaultAiAnalyzerService.java

@@ -27,6 +27,8 @@
 import com.bablsoft.accessflow.core.events.AiAnalysisCompletedEvent;
 import com.bablsoft.accessflow.core.events.AiAnalysisFailedEvent;
 import com.bablsoft.accessflow.core.events.AiAnalysisSkippedEvent;
+import com.bablsoft.accessflow.core.api.QueryEstimateSnapshot;
+import com.bablsoft.accessflow.proxy.api.QueryCostEstimateService;
 import com.bablsoft.accessflow.proxy.api.SqlParserService;
 import io.micrometer.core.instrument.MeterRegistry;
 import io.micrometer.observation.Observation;
@@ -69,6 +71,7 @@ class DefaultAiAnalyzerService implements AiAnalyzerService {
     private final ApplicationEventPublisher eventPublisher;
     private final AiRateLimiter aiRateLimiter;
     private final ObservationRegistry observationRegistry;
+    private final QueryCostEstimateService queryCostEstimateService;
     private final MeterRegistry meterRegistry;
 
     @Override
@@ -143,9 +146,14 @@ public void analyzeSubmittedQuery(UUID queryRequestId) {
         } catch (RuntimeException e) {
             log.warn("Schema introspection failed for query {}: {}", queryRequestId, e.getMessage());
         }
+        // AF-624: ensure the pre-flight cost estimate exists (idempotent — the proxy listener may
+        // already have computed it) and fold it into the prompt so risk scoring sees the actual
+        // blast radius. Best-effort: an estimate failure never blocks analysis.
+        String costEstimateContext = buildCostEstimateContext(queryRequestId);
         try {
             aiRateLimiter.enforce(snapshot.organizationId());
-            var analysis = analyzeWithObservation(snapshot, descriptor, schemaContext);
+            var analysis = analyzeWithObservation(snapshot, descriptor, schemaContext,
+                    costEstimateContext);
             var result = ClassificationRiskBooster.boost(analysis, ClassificationRiskBooster.bumpFor(
                     referencedClassifications(snapshot.sqlText(), classificationTags)));
             var issuesJson = responseParser.issuesAsJson(result.issues());
@@ -180,13 +188,15 @@ public void analyzeSubmittedQuery(UUID queryRequestId) {
      */
     private AiAnalysisResult analyzeWithObservation(QueryRequestSnapshot snapshot,
                                                     DatasourceConnectionDescriptor descriptor,
-                                                    String schemaContext) {
+                                                    String schemaContext,
+                                                    String costEstimateContext) {
         // Each tag is set exactly once per branch (no re-add) so the meter series carry the right
         // provider/risk on success and consistent keys on failure.
         Observation observation = Observation.start("accessflow.ai.analyze", observationRegistry);
         try (Observation.Scope ignored = observation.openScope()) {
             var analysis = strategy.analyze(snapshot.sqlText(), descriptor.dbType(), schemaContext,
-                    resolveLanguage(snapshot.organizationId()), descriptor.aiConfigId());
+                    costEstimateContext, resolveLanguage(snapshot.organizationId()),
+                    descriptor.aiConfigId());
             observation.lowCardinalityKeyValue("provider", providerTag(analysis.aiProvider()))
                     .lowCardinalityKeyValue("risk_level",
                             analysis.riskLevel() != null ? analysis.riskLevel().name() : "none")
@@ -243,6 +253,52 @@ private UUID requireBoundAiConfig(DatasourceConnectionDescriptor descriptor) {
         return aiConfigId;
     }
 
+    /**
+     * Best-effort AF-624 prompt context: triggers/reads the persisted estimate and renders it as a
+     * short plain-text summary. Null (→ the template's "(no cost estimate available)" fallback)
+     * when the estimate is absent, unsupported, or failed.
+     */
+    private String buildCostEstimateContext(UUID queryRequestId) {
+        QueryEstimateSnapshot estimate;
+        try {
+            estimate = queryCostEstimateService.estimateSubmittedQuery(queryRequestId).orElse(null);
+        } catch (RuntimeException e) {
+            log.warn("Cost estimate unavailable for query {}: {}", queryRequestId, e.getMessage());
+            return null;
+        }
+        if (estimate == null || estimate.failed() || !estimate.supported()) {
+            return null;
+        }
+        var sb = new StringBuilder();
+        if (estimate.affectedRowCount() != null) {
+            sb.append("Exact affected-row count for this ").append(estimate.queryType())
+                    .append(" (governed SELECT COUNT(*)): ")
+                    .append(String.format(Locale.ROOT, "%,d", estimate.affectedRowCount()))
+                    .append(" rows.");
+        }
+        if (estimate.estimatedRows() != null) {
+            if (!sb.isEmpty()) {
+                sb.append(' ');
+            }
+            sb.append("The engine's execution plan estimates ~")
+                    .append(String.format(Locale.ROOT, "%,d", estimate.estimatedRows()))
+                    .append(" rows");
+            if (estimate.scanType() != null) {
+                sb.append(" via ").append(estimate.scanType());
+            }
+            if (estimate.estimatedCost() != null) {
+                sb.append(" (cost ").append(estimate.estimatedCost()).append(')');
+            }
+            sb.append('.');
+        } else if (estimate.scanType() != null) {
+            if (!sb.isEmpty()) {
+                sb.append(' ');
+            }
+            sb.append("Plan root operation: ").append(estimate.scanType()).append('.');
+        }
+        return sb.isEmpty() ? null : sb.toString();
+    }
+
     private String resolveLanguage(UUID organizationId) {
         if (organizationId == null) {
             return SupportedLanguage.EN.code();

backend/src/main/java/com/bablsoft/accessflow/ai/internal/GuardrailAiAnalyzerStrategy.java

@@ -37,10 +37,11 @@ class GuardrailAiAnalyzerStrategy implements AiAnalyzerStrategy {
     }
 
     @Override
-    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext, String language,
-                                    UUID aiConfigId) {
+    public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
+                                    String costEstimateContext, String language, UUID aiConfigId) {
         guard(sql);
-        return delegate.analyze(sql, dbType, schemaContext, language, aiConfigId);
+        return delegate.analyze(sql, dbType, schemaContext, costEstimateContext, language,
+                aiConfigId);
     }
 
     @Override