feat(AF-623): automated sensitive-data discovery & classification scans

CLAUDE.md

@@ -117,6 +117,9 @@ com.bablsoft.accessflow/
 │   ├── api/
 │   ├── events/
 │   └── internal/   # persistence, scheduled (run + timeout jobs), web
+├── discovery/      # Automated sensitive-data discovery (AF-623): DiscoveryScanJob samples column data via the engine sampling path, regex+checksum detectors (email, PAN+Luhn, SSN, IBAN, phone) + optional fail-safe AI pass propose classification tags an admin confirms (AF-447 derivation) or dismisses
+│   ├── api/
+│   └── internal/   # config, persistence, detect (pure detectors), scheduled, web
 └── mcp/            # Spring AI stateless MCP server — @Tool callbacks for AI agents
     ├── api/
     └── internal/
@@ -270,6 +273,11 @@ com.bablsoft.accessflow/
 | `ACCESSFLOW_LIFECYCLE_REVIEW_TIMEOUT` | ISO-8601 duration. How long an erasure request may sit in `PENDING_REVIEW` before `ErasureReviewTimeoutJob` auto-rejects it (default `PT168H`, AF-519). Binds `accessflow.lifecycle.review-timeout`. |
 | `ACCESSFLOW_REQUESTGROUPS_RUN_POLL_INTERVAL` | ISO-8601 duration. Cadence at which `ScheduledGroupRunJob` (the `requestgroups` module, AF-501) scans for `APPROVED` grouped requests whose `scheduled_for ≤ now()` and executes their ordered member sequence (default `PT1M`). Binds `accessflow.requestgroups.run-poll-interval`. |
 | `ACCESSFLOW_REQUESTGROUPS_TIMEOUT_POLL_INTERVAL` | ISO-8601 duration. Cadence at which `GroupTimeoutJob` (the `requestgroups` module, AF-501) scans for grouped requests stuck in `PENDING_REVIEW` past the review timeout and auto-rejects them (`TIMED_OUT`) (default `PT5M`). Binds `accessflow.requestgroups.timeout-poll-interval`. |
+| `ACCESSFLOW_DISCOVERY_SCAN_POLL_INTERVAL` | ISO-8601 duration. Cadence at which `DiscoveryScanJob` (the `discovery` module, AF-623) checks for datasources whose sensitive-data discovery scan is due — enabled `discovery_scan_config` rows past their own `scan_interval_hours` (default `PT15M`). Binds `accessflow.discovery.scan-poll-interval`. |
+| `ACCESSFLOW_DISCOVERY_SCAN_TIME_BUDGET` | ISO-8601 duration. Wall-clock budget for one discovery scan; tables past the deadline are skipped and the run is flagged partial (default `PT10M`). Binds `accessflow.discovery.scan-time-budget`. |
+| `ACCESSFLOW_DISCOVERY_SAMPLE_STATEMENT_TIMEOUT` | ISO-8601 duration. Per-table statement timeout for the discovery scan's bounded sample read (default `PT10S`). Binds `accessflow.discovery.sample-statement-timeout`. |
+| `ACCESSFLOW_DISCOVERY_MAX_TABLES_PER_SCAN` | Hard cap on tables sampled in one discovery scan (default `200`). Binds `accessflow.discovery.max-tables-per-scan`. |
+| `ACCESSFLOW_DISCOVERY_MAX_AI_TABLES_PER_SCAN` | Hard cap on tables sent through the optional discovery AI pass per scan, bounding provider spend (default `25`). Binds `accessflow.discovery.max-ai-tables-per-scan`. |
 | `ACCESSFLOW_APIGOV_REVIEW_TIMEOUT` | ISO-8601 duration. How long an API request may sit in `PENDING_REVIEW` before `ApiRequestTimeoutJob` auto-rejects it (default `PT24H`). |
 | `ACCESSFLOW_APIGOV_MAX_REQUEST_BODY_BYTES` | Cap on the total encoded size of a submitted API request body (#517) — raw text, x-www-form-urlencoded, or the base64-decoded size of form-data / binary file parts (files ride inline as bounded base64 since AccessFlow has no object storage). Exceeding it rejects the submission with HTTP 422. Default `5242880` (5 MiB). Binds `accessflow.apigov.max-request-body-bytes`. |
 | `ACCESSFLOW_APIGOV_MAX_RESPONSE_BYTES` | System-wide hard ceiling (#521) on a stored — and therefore downloadable — API response body; the absolute backstop above any per-connector `max_response_bytes` (effective cap is the **min** of the two). The body lives in a Postgres `text` column and is read fully into JVM memory during the call. Default `10485760` (10 MiB). Binds `accessflow.apigov.max-response-bytes`. |

README.md

@@ -72,6 +72,7 @@ A glance at the day-to-day flows engineers and approvers actually use.
 - **Dynamic data masking** — per-column masking policies (full, partial last-N, stable hash, email-preserving, format-preserving) with role / group / user **reveal** conditions evaluated per requester. Masking is applied at result-read time before results are serialized or stored, so unmasked values never persist; applied policy ids are recorded in the audit log. Extends the static `restricted_columns` masking.
 - **Row-level security** — per-table row predicates the proxy injects into the parsed SQL so a scoped user only sees (SELECT) or affects (UPDATE/DELETE) the rows they are authorised for. Admins author a structured `column operator value` predicate where the value is a fixed literal or a `:user.*` variable (built-in id / email / role / groups, or an admin-set per-user attribute). Values are bound as JDBC parameters — never concatenated; predicates that can't be safely applied are rejected, never run unfiltered. Composes with column masking and the schema/table allow-list.
 - **Data classification tagging** — tag tables and columns as PII, PCI, PHI, GDPR, FINANCIAL, or SENSITIVE right in the schema explorer. Tagging a column auto-applies a masking policy, the AI analyzer raises a query's risk score when it touches a tagged object, and a derivation preview suggests a stricter review posture. Tags are audited and queryable org-wide as the evidence base for compliance reporting.
+- **Automated sensitive-data discovery** — an opt-in per-datasource scanner samples column data through the same governed sampling path, detects sensitive values with local regex + checksum detectors (emails, credit-card PANs with Luhn, SSNs, IBANs, phone numbers) and optionally your bound AI analyzer (which only ever sees column names, types, and redacted samples), and **proposes** classification tags in a review worklist. Confirming a finding applies the tag — deriving masking automatically — while dismissing suppresses it permanently; scans and decisions are audited, and an on-demand "Scan now" complements the scheduled cadence.
 - **User groups** — bundle reviewers (and other users) into groups; groups can be assigned as datasource reviewers, **granted data/API access** (a whole team inherits a datasource or API-connector grant — read/write/DDL/break-glass — instead of a row per member; effective access is the most-permissive union of a user's direct and group grants), and auto-synced from SAML / OAuth2 IdP claims via per-IdP `group_mappings` (admin-managed memberships are preserved across logins).
 - **AI query analysis** — pluggable adapters for OpenAI, Anthropic Claude, self-hosted Ollama, any OpenAI API–compatible backend (vLLM, LM Studio, Together, Groq, OpenRouter, …) via a custom endpoint, and Hugging Face (Inference Providers router or a local / self-hosted TGI server); risk scoring (0–100), missing-index detection, anti-pattern hints. Per-organization configuration via the admin UI, including an **editable analyzer prompt** per configuration (override the built-in template, or leave it blank to use the default); admin **AI analyses dashboard** charts average risk over time, top issue categories, and most active submitters. A per-organization **rate limit** (requests/minute, default 30) and optional **monthly token budget** cap AI usage so a runaway editor or compromised account can't drain the provider API key — enforced before every analysis call; exceedance returns HTTP 429 in the editor or records a sentinel `CRITICAL` analysis row on the async path. A per-organization **fallback pool** marks configurations (typically a local Ollama) to retry in priority order when the primary provider is unreachable — air-gap-friendly resilience without weakening any failure semantics.
 - **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.

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

@@ -0,0 +1,45 @@
+package com.bablsoft.accessflow.ai.api;
+
+import com.bablsoft.accessflow.core.api.DataClassification;
+
+import java.util.List;
+import java.util.UUID;
+
+/**
+ * Optional AI pass of the sensitive-data discovery scan (AF-623). Classifies a table's columns
+ * using the organization's first usable {@code ai_config}, complementing the local regex/checksum
+ * detectors (e.g. a {@code national_id} column no regex covers).
+ *
+ * <p><strong>Privacy:</strong> callers must pass only redacted sample values (format-preserving
+ * masking) — raw sampled data never reaches the AI provider.
+ *
+ * <p><strong>Fail-safe by contract:</strong> returns an empty list when the org has no usable AI
+ * config, the provider errors, or the response is not the expected JSON — it never throws and
+ * never blocks the scan (same posture as the UBA anomaly summary, AF-383).
+ */
+public interface DataDiscoveryAiService {
+
+    List<DiscoveryColumnSuggestion> classifyColumns(UUID organizationId,
+                                                    DiscoveryTableContext context);
+
+    /** One table's worth of column metadata + redacted samples. */
+    record DiscoveryTableContext(String tableName, List<DiscoveryColumnContext> columns) {
+
+        public DiscoveryTableContext {
+            columns = columns == null ? List.of() : List.copyOf(columns);
+        }
+    }
+
+    /** {@code redactedSamples} must already be redacted by the caller. */
+    record DiscoveryColumnContext(String name, String type, List<String> redactedSamples) {
+
+        public DiscoveryColumnContext {
+            redactedSamples = redactedSamples == null ? List.of() : List.copyOf(redactedSamples);
+        }
+    }
+
+    /** A proposed classification for one column; {@code confidence} is clamped to 0–100. */
+    record DiscoveryColumnSuggestion(String columnName, DataClassification classification,
+                                     int confidence, String rationale) {
+    }
+}

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

@@ -84,6 +84,18 @@ class AiAnalyzerStrategyHolder implements AiAnalyzerStrategy {
             database-access pattern looks anomalous relative to the user's historical baseline, and \
             what a reviewer should check. Do not use markdown, headings, or bullet points.""";
 
+    // AF-623: sensitive-data discovery AI pass. Sample values are redacted (format-preserving)
+    // before they reach this prompt — the model judges by column name, type, and value shape.
+    private static final String DISCOVERY_CLASSIFY_PREAMBLE = """
+            You are a data-classification analyst. Given a database table with its column names, \
+            types, and redacted sample values (digits replaced by *, letters by x — only the shape \
+            is preserved), identify columns likely to contain sensitive data. Respond with ONLY a \
+            JSON object of the form \
+            {"columns":[{"column":"<name>","classification":"PII|PCI|PHI|GDPR|FINANCIAL|SENSITIVE",\
+            "confidence":<0-100>,"rationale":"<one short sentence>"}]}. \
+            Propose only columns you genuinely believe are sensitive; when none are, respond \
+            {"columns":[]}. No markdown, no code fences, no prose outside the JSON.""";
+
     @Override
     public AiAnalysisResult analyze(String sql, DbType dbType, String schemaContext,
                                     String costEstimateContext, String language, UUID aiConfigId) {
@@ -184,22 +196,38 @@ private static void rethrowIfNotFallbackEligible(RuntimeException ex) {
      * config exists or any provider/parse error occurs — it never throws and never blocks detection.
      */
     Optional<String> summarizeFreeform(UUID organizationId, String userPrompt) {
+        return completeFreeform(organizationId, ANOMALY_SUMMARY_PREAMBLE, userPrompt,
+                "anomaly summary");
+    }
+
+    /**
+     * Freeform call for the sensitive-data discovery AI pass (AF-623). Same fail-safe contract as
+     * {@link #summarizeFreeform}; the preamble demands strict JSON, which the discovery service
+     * parses leniently and drops on any mismatch.
+     */
+    Optional<String> classifyDiscoveryColumns(UUID organizationId, String userPrompt) {
+        return completeFreeform(organizationId, DISCOVERY_CLASSIFY_PREAMBLE, userPrompt,
+                "discovery classification");
+    }
+
+    private Optional<String> completeFreeform(UUID organizationId, String systemPreamble,
+                                              String userPrompt, String logContext) {
         if (organizationId == null || userPrompt == null || userPrompt.isBlank()) {
             return Optional.empty();
         }
         try {
             var entity = resolveUsableConfig(organizationId).orElse(null);
             if (entity == null) {
-                log.debug("No usable ai_config for org {}; skipping anomaly summary", organizationId);
+                log.debug("No usable ai_config for org {}; skipping {}", organizationId, logContext);
                 return Optional.empty();
             }
             var chatModel = chatModelCache.computeIfAbsent(entity.getId(), key -> buildChatModel(entity));
-            var invocation = ChatModelInvoker.invoke(chatModel, ANOMALY_SUMMARY_PREAMBLE, userPrompt,
+            var invocation = ChatModelInvoker.invoke(chatModel, systemPreamble, userPrompt,
                     entity.getProvider().name());
             var text = invocation.text();
             return text == null || text.isBlank() ? Optional.empty() : Optional.of(text.strip());
         } catch (RuntimeException ex) {
-            log.warn("Anomaly AI summary failed for org {}: {}", organizationId, ex.getMessage());
+            log.warn("AI {} failed for org {}: {}", logContext, organizationId, ex.getMessage());
             return Optional.empty();
         }
     }

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

@@ -0,0 +1,134 @@
+package com.bablsoft.accessflow.ai.internal;
+
+import com.bablsoft.accessflow.ai.api.DataDiscoveryAiService;
+import com.bablsoft.accessflow.core.api.DataClassification;
+import lombok.RequiredArgsConstructor;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.ObjectProvider;
+import org.springframework.stereotype.Service;
+import tools.jackson.databind.JsonNode;
+import tools.jackson.databind.ObjectMapper;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.UUID;
+
+/**
+ * Discovery AI classification pass (AF-623). Builds a plain-text prompt from the (already
+ * redacted) column contexts, calls the org's first usable {@code ai_config} through the holder's
+ * freeform lane, and leniently parses the demanded strict-JSON response — unknown columns,
+ * unknown classifications, and malformed payloads are dropped, never thrown.
+ *
+ * <p>The holder is resolved through an {@link ObjectProvider} (lazy, by-name) rather than injected
+ * by concrete type, matching {@code AnomalySummaryService}: integration tests that
+ * {@code @MockitoBean AiAnalyzerStrategy} replace the holder bean with a bare interface mock,
+ * which is not assignable to the concrete field type.
+ */
+@Service
+@RequiredArgsConstructor
+class DefaultDataDiscoveryAiService implements DataDiscoveryAiService {
+
+    private static final Logger log = LoggerFactory.getLogger(DefaultDataDiscoveryAiService.class);
+    private static final int MAX_RATIONALE_LENGTH = 500;
+
+    private final ObjectProvider<AiAnalyzerStrategyHolder> strategyHolder;
+    private final ObjectMapper objectMapper;
+
+    @Override
+    public List<DiscoveryColumnSuggestion> classifyColumns(UUID organizationId,
+                                                           DiscoveryTableContext context) {
+        if (organizationId == null || context == null || context.columns().isEmpty()) {
+            return List.of();
+        }
+        var response = strategyHolder.getObject()
+                .classifyDiscoveryColumns(organizationId, buildPrompt(context))
+                .orElse(null);
+        if (response == null) {
+            return List.of();
+        }
+        return parse(response, context);
+    }
+
+    private static String buildPrompt(DiscoveryTableContext context) {
+        var sb = new StringBuilder();
+        sb.append("Table: ").append(context.tableName()).append('\n');
+        sb.append("Columns:\n");
+        for (var column : context.columns()) {
+            sb.append("- ").append(column.name());
+            if (column.type() != null && !column.type().isBlank()) {
+                sb.append(" (").append(column.type()).append(')');
+            }
+            if (!column.redactedSamples().isEmpty()) {
+                sb.append(" samples: ").append(String.join(", ", column.redactedSamples()));
+            }
+            sb.append('\n');
+        }
+        return sb.toString();
+    }
+
+    private List<DiscoveryColumnSuggestion> parse(String response, DiscoveryTableContext context) {
+        JsonNode root;
+        try {
+            root = objectMapper.readTree(stripCodeFences(response));
+        } catch (RuntimeException ex) {
+            log.warn("Discovery AI response was not valid JSON, ignoring: {}", ex.getMessage());
+            return List.of();
+        }
+        var columnsNode = root.path("columns");
+        if (!columnsNode.isArray()) {
+            log.warn("Discovery AI response missing 'columns' array, ignoring");
+            return List.of();
+        }
+        var knownColumns = new HashSet<String>();
+        for (var column : context.columns()) {
+            knownColumns.add(column.name().toLowerCase(Locale.ROOT));
+        }
+        var suggestions = new ArrayList<DiscoveryColumnSuggestion>();
+        for (var node : columnsNode) {
+            var suggestion = toSuggestion(node, knownColumns);
+            if (suggestion != null) {
+                suggestions.add(suggestion);
+            }
+        }
+        return List.copyOf(suggestions);
+    }
+
+    private DiscoveryColumnSuggestion toSuggestion(JsonNode node, HashSet<String> knownColumns) {
+        var columnName = node.path("column").asString("");
+        if (columnName.isBlank() || !knownColumns.contains(columnName.toLowerCase(Locale.ROOT))) {
+            return null;
+        }
+        DataClassification classification;
+        try {
+            classification = DataClassification.valueOf(
+                    node.path("classification").asString("").toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException ex) {
+            return null;
+        }
+        var confidenceNode = node.path("confidence");
+        var confidence = confidenceNode.isNumber()
+                ? Math.clamp(confidenceNode.asInt(), 0, 100) : 0;
+        var rationale = node.path("rationale").asString("");
+        if (rationale.length() > MAX_RATIONALE_LENGTH) {
+            rationale = rationale.substring(0, MAX_RATIONALE_LENGTH);
+        }
+        return new DiscoveryColumnSuggestion(columnName, classification, confidence,
+                rationale.isBlank() ? null : rationale);
+    }
+
+    /** Models sometimes wrap JSON in markdown fences despite instructions — tolerate it. */
+    private static String stripCodeFences(String response) {
+        var trimmed = response.strip();
+        if (trimmed.startsWith("```")) {
+            var firstNewline = trimmed.indexOf('\n');
+            var lastFence = trimmed.lastIndexOf("```");
+            if (firstNewline >= 0 && lastFence > firstNewline) {
+                return trimmed.substring(firstNewline + 1, lastFence).strip();
+            }
+        }
+        return trimmed;
+    }
+}