feat(AF-941): bytes-scanned cost caps and estimated_bytes_scanned routing

CLAUDE.md

@@ -341,6 +341,14 @@ nullable or has a DEFAULT. `ALTER TYPE … ADD VALUE` needs a `.sql.conf` sideca
                           once as SQL_REVIEW_BLOCKED when it changed the outcome)
   PENDING_AI → REJECTED  (routing-policy AUTO_REJECT — AF-379; no review_decisions row,
                           audited via QueryAutoRejectedEvent)
+  PENDING_AI → REJECTED  (bytes-scanned cap — #941; the warehouse's pre-flight bytes estimate
+                          exceeds the datasource/grant cap, or there is none and the datasource's
+                          bytes_cap_missing_estimate=REJECT. Decided before routing and the
+                          AI-failed path, no routing_decision row, QueryAutoRejectedEvent with a
+                          null policy id; audited as QUERY_BYTES_SCANNED_CAP_ENFORCED)
+  PENDING_AI → PENDING_REVIEW (bytes-scanned cap, no estimate, bytes_cap_missing_estimate=
+                          REQUIRE_REVIEW — #941; suppresses the same auto-approve paths as a SQL
+                          review BLOCK, never softens AUTO_REJECT)
   PENDING_REVIEW → APPROVED or REJECTED (external ticket resolution — AF-453; a channel with
                           bidirectional_sync=true maps a ServiceNow/Jira ticket resolution onto a
                           decision via workflow.api.ExternalDecisionService. System-attributed:
@@ -363,7 +371,9 @@ nullable or has a DEFAULT. `ALTER TYPE … ADD VALUE` needs a `.sql.conf` sideca
                               recurrence_next_run_at, records recurrence_halted_reason, and
                               audits RECURRING_SERIES_HALTED)
   APPROVED       → EXECUTED  (break-glass run — audit action QUERY_BREAK_GLASS_EXECUTED — AF-385)
-  APPROVED       → FAILED    (execution error)
+  APPROVED       → FAILED    (execution error; also the bytes-scanned cap re-checked just before
+                              execution refusing the run — #941, scheduled / recurring /
+                              break-glass included)
   ```
 
   Illegal transitions must throw a domain exception, not silently succeed. **Break-glass /

README.md

@@ -67,7 +67,7 @@ A glance at the day-to-day flows engineers and approvers actually use.
 
 - **Proxy-first execution** — no user ever holds production credentials; the proxy holds them encrypted and opens connections only after approval. Single SQL statements run with autocommit; multi-statement INSERT/UPDATE/DELETE batches wrapped in `BEGIN; … COMMIT;` execute atomically inside one JDBC transaction (mixed SELECT/DML batches are rejected at parse time), with homogeneous INSERT runs collapsed into JDBC `executeBatch()` for bulk-load throughput. Optional **multi-replica read load balancing**: attach any number of replica endpoints to a datasource and SELECT traffic round-robins across the healthy ones — per-node health checks with circuit-breaker failover skip downed replicas, and only full replica-set exhaustion falls back to the primary (with an audit row). Optional **SELECT result caching**: opt a datasource into a Redis-backed result cache (per-datasource TTL) keyed over the security-rewritten query — masking and row-level security still apply — and invalidated on any proxied write to a referenced table.
 - **Configurable review workflows** — per-datasource review plans, multi-stage sequential approval chains, optional auto-approve for reads, approval timeouts with auto-reject. Reviewers can be scoped **per-datasource** (directly or via groups) so different teams see only the queues that belong to them. Reviewers going away can set an **out-of-office delegation** naming a colleague to cover their review duty for a window — across queries, governed API calls, and grouped requests — with every decision recording both identities. A delegate can never act on the delegator's own requests, delegation never grants a permission they lack, and it does not chain. Plans can also **escalate** a request that nobody has decided on to the reviewers at its current stage plus your admins before the approval timeout auto-rejects it, and **nudge** those same reviewers on a cadence — both optional, both notify-only, so waiting never changes who may approve.
-- **Policy-as-code routing** — ordered, attribute-based routing policies decide a query's path after AI analysis and before reviewers see it: **auto-approve**, **auto-reject**, **require N approvals**, or **escalate**. Conditions match on query type, referenced tables (glob), AI risk level / score, requester role or group, time-of-day / day-of-week, WHERE / LIMIT presence, **query shape** (joins, set operations, subqueries, CTEs, GROUP BY / HAVING, aggregates, window functions), the transactional flag, and the submission **client context** — source IP / CIDR, user-agent, time-since-last-approval, and CI/CD origin (API key or `X-AccessFlow-CI` header) — combined with AND / OR / NOT. Client-context conditions **fail closed** (missing context never auto-approves), so an off-network or stale-approval query escalates to stricter review instead. First match by priority wins; on no match the query falls through to the datasource's review plan. Every automated decision is recorded in the audit log.
+- **Policy-as-code routing** — ordered, attribute-based routing policies decide a query's path after AI analysis and before reviewers see it: **auto-approve**, **auto-reject**, **require N approvals**, or **escalate**. Conditions match on query type, referenced tables (glob), AI risk level / score, requester role or group, time-of-day / day-of-week, WHERE / LIMIT presence, **query shape** (joins, set operations, subqueries, CTEs, GROUP BY / HAVING, aggregates, window functions), the transactional flag, the pre-flight estimate (**estimated rows, estimated bytes scanned**, scan type), and the submission **client context** — source IP / CIDR, user-agent, time-since-last-approval, and CI/CD origin (API key or `X-AccessFlow-CI` header) — combined with AND / OR / NOT. Client-context conditions **fail closed** (missing context never auto-approves), so an off-network or stale-approval query escalates to stricter review instead. First match by priority wins; on no match the query falls through to the datasource's review plan. Every automated decision is recorded in the audit log.
 - **Policy simulator** — dry-run a draft routing, row-security, or masking policy against your own historical query traffic before you save it. AccessFlow replays the window **twice** — once against your current policies, once with the draft applied — and reports the difference between those two runs: how many past queries would change, which users would be affected, and which queries a row predicate would newly filter, deny, or reject outright. It is strictly read-only — no connection to your database, nothing executed, nothing persisted — and it names its own approximations (memberships are read as they are now; an engine that cannot classify a query shape offline is reported as *unclassifiable*, never as safe) instead of implying a precision the data does not have.
 - **Deterministic SQL review rules (#860)** — a named rule catalog that judges every SQL query alongside the AI verdict and routing policies, and shows its findings **in the editor as you type**. Fourteen built-in rules derived from the parsed statement alone (missing `WHERE` on `UPDATE` / `DELETE`, an always-true `WHERE`, `SELECT *`, unbounded reads, cross joins, leading-wildcard `LIKE`, `DROP` / `TRUNCATE` / DDL, banned functions, protected tables by glob, DML outside a transaction), each at an admin-set **`OFF` / `WARN` / `BLOCK`** severity configured per **environment** (`DEVELOPMENT` / `TEST` / `STAGING` / `PRODUCTION`, plus an organisation default). **`BLOCK` escalates, it never rejects**: a blocking finding suppresses every auto-approve path and sends the query to a human reviewer; a routing auto-reject still rejects. Evaluated synchronously at submission so findings exist even when AI analysis is off or fails, rendered in each reader's language on the query detail, review queue, break-glass retro-review and request-group detail; break-glass runs record findings but are never gated by them, and non-relational engines report *not applicable*. See [`docs/19-sql-review.md`](https://github.com/bablsoft/accessflow/blob/main/docs/19-sql-review.md).
 - **Just-in-time (JIT) access requests** — users self-request temporary, scoped access to a datasource (read/write/DDL, optional schema/table scope) or an API connection (read/write, optional operation allow-list) for an ISO-8601 duration. Requests flow through the same approval engine, a time-boxed permission is granted on approval, and a clustered scheduler auto-revokes it on expiry (admins can also revoke early). A grant can opt into **query pre-approval**: while it is active, queries it covers skip human review and are auto-approved with the grant recorded as the approval provenance — routing policies, high-risk AI verdicts, and behavioural anomalies still override.
@@ -84,6 +84,7 @@ A glance at the day-to-day flows engineers and approvers actually use.
 - **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. Cloud warehouses additionally report the estimated bytes the query would scan (BigQuery's native dry-run job) — the direct cost signal for bytes-billed engines. 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.
+- **Bytes-scanned cost caps** — on BigQuery, Snowflake and Databricks, where a ten-row result can scan a terabyte, set a maximum bytes-scanned per query on the datasource and, tighter, per grant. A query whose pre-flight scan estimate exceeds the cap is refused before it runs, with a message naming both numbers; the cap is checked again right before execution, so scheduled, recurring, grouped and break-glass runs honour it too. When no estimate exists, the datasource decides explicitly: send the query to a person, or refuse it. For an advisory signal instead, route on `estimated_bytes_scanned` to escalate expensive queries.
 - **Approval-likelihood prediction (AF-645)** — a triage signal for busy review queues. AccessFlow trains a small per-organization statistical model on your own historical review decisions and shows reviewers the probability that a query gets approved — a bare percentage badge on the review queue, and a labelled "Historical approval likelihood: 78%" card on the query detail page. No LLM call and no external service — it's plain logistic regression over data you already collect (query type, AI risk verdict, cost estimate, time of day, per-submitter and per-datasource approval history), retrained daily. **Advisory only**: it never approves or rejects anything, and it is never an input to routing, grant coverage, or any other decision path. Auto-approved, break-glass, and grant-covered queries are excluded from training because they carry no reviewer judgment, and until an organization has enough decided history for the model to clear its quality gate the UI says "not enough review history yet" rather than showing a number.
 - **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.

backend/src/main/java/com/bablsoft/accessflow/access/internal/AccessGrantMaterializer.java

@@ -54,7 +54,8 @@ void materialize(UUID accessRequestId, UUID approvedByUserId) {
             return;
         }
         // A replaced row's denials carry over (#939): a JIT approval widens capabilities and expiry,
-        // never lifts a denial an admin set — JIT requests cannot even ask for deny-lists.
+        // never lifts a denial an admin set — JIT requests cannot even ask for deny-lists. The
+        // bytes-scanned cap carries over for the same reason (#941): a JIT approval never widens cost.
         var replaced = replaceExistingTimeBoxedPermission(entity);
         var command = new CreatePermissionCommand(
                 entity.getRequesterId(),
@@ -63,6 +64,7 @@ void materialize(UUID accessRequestId, UUID approvedByUserId) {
                 entity.isCanDdl(),
                 false,
                 null,
+                replaced.map(DatasourceUserPermissionView::bytesScannedLimitOverride).orElse(null),
                 toList(entity.getAllowedSchemas()),
                 toList(entity.getAllowedTables()),
                 null,

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

@@ -8,6 +8,7 @@
 import com.bablsoft.accessflow.ai.api.AiBudgetExceededException;
 import com.bablsoft.accessflow.ai.api.AiRateLimitExceededException;
 import com.bablsoft.accessflow.ai.internal.persistence.repo.AiConfigRepository;
+import com.bablsoft.accessflow.core.api.ByteSizeFormat;
 import com.bablsoft.accessflow.core.api.AiAnalysisPersistenceService;
 import com.bablsoft.accessflow.core.api.AiProviderType;
 import com.bablsoft.accessflow.core.api.DataClassification;
@@ -296,6 +297,14 @@ private String buildCostEstimateContext(UUID queryRequestId) {
             }
             sb.append("Plan root operation: ").append(estimate.scanType()).append('.');
         }
+        if (estimate.estimatedBytesScanned() != null) {
+            if (!sb.isEmpty()) {
+                sb.append(' ');
+            }
+            sb.append("The warehouse estimates this query will scan ")
+                    .append(ByteSizeFormat.format(estimate.estimatedBytesScanned()))
+                    .append('.');
+        }
         return sb.isEmpty() ? null : sb.toString();
     }
 

backend/src/main/java/com/bablsoft/accessflow/attestation/internal/DefaultAttestationLifecycleService.java

@@ -211,6 +211,11 @@ private String toSnapshotJson(DatasourcePermissionView view) {
         } else {
             node.putNull("row_limit_override");
         }
+        if (view.bytesScannedLimitOverride() != null) {
+            node.put("bytes_scanned_limit_override", view.bytesScannedLimitOverride());
+        } else {
+            node.putNull("bytes_scanned_limit_override");
+        }
         putStringArray(node, "allowed_schemas", view.allowedSchemas());
         putStringArray(node, "allowed_tables", view.allowedTables());
         putStringArray(node, "restricted_columns", view.restrictedColumns());

backend/src/main/java/com/bablsoft/accessflow/audit/api/AuditAction.java

@@ -132,6 +132,14 @@ public enum AuditAction {
      * when the guard changed the outcome; never for {@code WARN}, never on a rejection.
      */
     SQL_REVIEW_BLOCKED,
+    /**
+     * The bytes-scanned cap (#941) changed a query's outcome: refused it when it left
+     * {@code PENDING_AI} ({@code stage=decision}), turned an automatic approval into human review,
+     * or failed it just before execution ({@code stage=execution}). System-attributed — null actor,
+     * {@code trigger=bytes_scanned_cap}; metadata carries {@code limit}, {@code source},
+     * {@code estimated_bytes} and {@code outcome}.
+     */
+    QUERY_BYTES_SCANNED_CAP_ENFORCED,
     ROW_SECURITY_POLICY_CREATED,
     ROW_SECURITY_POLICY_UPDATED,
     ROW_SECURITY_POLICY_DELETED,

backend/src/main/java/com/bablsoft/accessflow/core/api/AppliedBytesCap.java

@@ -0,0 +1,36 @@
+package com.bablsoft.accessflow.core.api;
+
+import java.util.Objects;
+
+/**
+ * The bytes-scanned cap (#941) binding one user on one datasource: the most restrictive of the
+ * datasource's {@code max_bytes_scanned_per_query} and the user's merged grant override, together
+ * with the datasource's missing-estimate policy.
+ */
+public record AppliedBytesCap(long limit, BytesScannedCapSource source,
+                              BytesCapMissingEstimateAction missingEstimate) {
+
+    public AppliedBytesCap {
+        if (limit <= 0) {
+            throw new IllegalArgumentException("limit must be positive");
+        }
+        Objects.requireNonNull(source, "source");
+        missingEstimate = missingEstimate == null
+                ? BytesCapMissingEstimateAction.REQUIRE_REVIEW : missingEstimate;
+    }
+
+    /**
+     * Compares an estimate against the cap. {@code null} means no estimate exists; the
+     * datasource's missing-estimate policy then decides.
+     */
+    public BytesScannedCapOutcome check(Long estimatedBytesScanned) {
+        if (estimatedBytesScanned == null) {
+            return missingEstimate == BytesCapMissingEstimateAction.REJECT
+                    ? BytesScannedCapOutcome.NO_ESTIMATE_REJECTED
+                    : BytesScannedCapOutcome.NO_ESTIMATE_REVIEW;
+        }
+        return estimatedBytesScanned > limit
+                ? BytesScannedCapOutcome.EXCEEDED
+                : BytesScannedCapOutcome.WITHIN;
+    }
+}