Merge remote-tracking branch 'origin/feature/compare' into feature/watches

Watches evaluate a metric per time bucket as their owner (SPIKE, DROP,
OUT_OF_RANGE, NEW_CATEGORY, NULL_RATE_CREEP with robust baselines) on a
hidden scheduled process, raise and resolve data.analysis.anomaly alerts
through QAlerts, and offer Evaluate now / Enable / Disable record actions
and a preview endpoint (extension 1.3.0). Pivot and crosstab labels are
human (Yes/No, Average Annual Salary). Contingent fields: a watch's saved
analysis depends on its table, table names use the core tables PVS, and
analyzable/excluded fields pick from the chosen table's fields.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

CHANGELOG.md

@@ -6,6 +6,12 @@ Following [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and [Semantic
 
 ### Added
 
+- Phase e anomaly watches: owner-locked configuration, validated metrics/detector parameters, complete calendar buckets, five pure detectors, and scheduled evaluation through owner sessions.
+- Core alerts SPI type `data.analysis.anomaly` (core disallows the spec's underscore), stable watch keys, relative drill links, raise/recovery transitions, and persisted outcomes without an alerts implementation.
+- Standard streamed ETL Evaluate now / Enable / Disable record actions and firing-first linked watches on the Analyses home.
+- Read-only watch preview HTTP endpoint, semantic chart/baseline annotations, extension 1.3.0 and regenerated OpenAPI fragment.
+- Detector edge cases, real-row owner/security/alert transition tests, calendar boundary and action tests, and HTTP preview coverage.
+
 - Phase d compare, shift, correlation, scatter and crosstab service operations and HTTP specs; extension version 1.2.0 with typed `DataAnalysis*V1` schemas and regenerated OpenAPI.
 - Pure-Java two-proportion, Welch, chi-square/Cramer V, PSI and Pearson statistics, including regularized incomplete beta/gamma and independent SciPy references.
 - Core-pushed numeric moments and batched correlations, sample trend lines, calendar distribution shifts, category pooling, privacy suppression and exact count drills.
@@ -23,6 +29,9 @@ Following [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and [Semantic
 
 ### Changed
 
+- Boolean pivot/crosstab headers now say Yes / No / (Blank). Pivot measures use human labels such as Count, Average Amount, Sum of Amount, Distinct Email, Median Amount and 90th percentile of Amount.
+- Watch DDL includes owner/date fields and last value/message; existing databases need an owner/date backfill before enabling watches.
+
 - Scalable estimates sample each native aggregate without primary-key IN lists, scale counts using observed sample sizes, retain exact population totals and unscaled ratios, and drill over the full population. Portable estimates use one primary-key boundary and report unscaled prefix counts.
 
 - Target QQQ 4.1.0-RC.6; local verification uses `-Dqqq.snapshot.version=4.1.0-DATAANALYSIS-SNAPSHOT`.

README.md

@@ -6,7 +6,7 @@ Statistical analysis and data dives on any QQQ table, filter or saved view: prof
 
 The host API provides column and table profiles, pivots, comparisons, distribution shifts, correlations, scatter samples, crosstabs, distinct-value pages, duplicate groups, drill-through filters, privacy suppression, and asynchronous exact profile runs with a persistent per-user cache. Metadata, six visible tables, owner locks, PostgreSQL DDL generation and the Analyses home are also available.
 
-Thirteen HTTP operations are available under `/qqq/v1/qbit/data-analysis`; the Analyze UI is a later phase. Pivot and relationship backends are available. Anomalies, saved-analysis execution, scheduling and exports remain later phases.
+Fourteen HTTP operations are available under `/qqq/v1/qbit/data-analysis`; the Analyze UI is a later phase. Pivot and relationship backends are available. Anomaly watches, their scheduled evaluator and alert transitions are available. Saved-analysis execution, scheduled delivery and exports remain later phases.
 
 ## Requirements
 
@@ -47,6 +47,8 @@ QQQ names and physical SQL table names are identical. `backendName` in configura
 
 Only Saved Analyses, Watches, Analysis Settings and Analysis Runs appear in navigation. Schedules, stars and linked watches appear on saved records. Enum PVS cover analysis kind, chart kind, detector, watch state, run status, schedule frequency and format. JSON definitions and results are stored as text. Profile, distinct-value and duplicate requests have typed Java contracts and are validated by the facade.
 
+The four `table_name` fields use core's `tables` possible-value source; an existing host source is preserved. A watch's `saved_id` options are filtered by its selected `table_name`. Analysis Settings' `analyzable_fields` and `excluded_fields` use a custom field picker scoped to the selected table. Core currently has no multi-select PVS field, so these settings use a single PVS with comma-separated TEXT storage (for example, `amount,status`); the configuration resolver continues to parse those lists. Explicit dependency metadata awaits core's `possibleValueSourceDependsOn` support.
+
 ## Permissions
 
 | Permission | Grants |
@@ -138,7 +140,7 @@ Count drills combine an untouched copy of the request filter with their producin
 
 ## HTTP endpoints
 
-Extension `data-analysis` version **1.2.0**, prefix `/qbit/data-analysis` inside `/qqq/v1`:
+Extension `data-analysis` version **1.3.0**, prefix `/qbit/data-analysis` inside `/qqq/v1`:
 
 | Method | Path after `/qqq/v1` | Body or purpose |
 | --- | --- | --- |
@@ -155,10 +157,11 @@ Extension `data-analysis` version **1.2.0**, prefix `/qbit/data-analysis` inside
 | POST | `/qbit/data-analysis/tables/{table}/crosstab` | `{rowField, columnField, filter, limit}`; default/maximum 20 categories per axis plus Other |
 | POST | `/qbit/data-analysis/runs` | `{operation: COLUMN_PROFILE or TABLE_PROFILE, tableName, request}` returns `{runId, status}` |
 | GET | `/qbit/data-analysis/runs/{runId}` | Status, timestamps, error, typed `columnProfile` or `tableProfile` |
+| POST | `/qbit/data-analysis/watches/{id}/preview` | No body; complete bucket series, `TIME_SERIES` chart with baseline annotations and detector verdict; no writes or alerts |
 
 Every operation uses the request session through `DataAnalysisService`. Capabilities returns 200 with `canAnalyze: false` and no fields when denied. Other permission denials return 403, bad bodies or unavailable fields return 400, and unknown tables return 404. Core middleware supplies plain messages in its standard `{error: message}` JSON envelope. Run rows remain owner-locked.
 
-Requests and drills reuse core's `QueryFilter` JSON shape and schema reference. Versioned `DataAnalysis...V1` schemas cover profiles, metrics, drills, charts, series, points, field summaries, distinct and duplicate results, capabilities and runs. The [OpenAPI fragment](src/main/resources/openapi/qqq-v1/data-analysis.yaml) is checked against regeneration. The deployed `/qqq/v1/openapi.json` and `UiApiInvoker.listOperations()` include all eight operations.
+Requests and drills reuse core's `QueryFilter` JSON shape and schema reference. Versioned `DataAnalysis...V1` schemas cover profiles, metrics, drills, charts, series, points, field summaries, distinct and duplicate results, capabilities and runs. The [OpenAPI fragment](src/main/resources/openapi/qqq-v1/data-analysis.yaml) is checked against regeneration. The deployed `/qqq/v1/openapi.json` and `UiApiInvoker.listOperations()` include all fourteen operations.
 
 Regenerate deliberately after an API change:
 
@@ -202,6 +205,34 @@ Shift uses session-timezone calendar buckets. Only buckets whose end has passed
 
 The minimum-group-size guard removes small numeric summaries, suppresses crosstab cells and revealing marginal shares, disables drills on privacy-merged Other groups, and suppresses scatter points below the minimum. Statistical tests and PSI are unavailable when protected values would contribute. No runtime statistics dependency is added; beta/gamma, Welch, chi-square, proportions, PSI and Pearson are pure Java, checked against independently computed SciPy references.
 
+## Anomaly watches
+
+Watches are ordinary owner-locked `data_analysis_watch` rows. Create and edit them through core table CRUD with `data_analysis.watch.read`, `.insert`, `.edit`, and `.delete`. Standard table/record actions **Evaluate now**, **Enable**, and **Disable** use streamed ETL Review and Result summaries. They never appear in navigation. The home lists firing watches first and links to their records.
+
+A watch has `owner_user_id`, `date_field`, `bucket` (`HOUR`, `DAY`, `WEEK`, `MONTH`), `detector`, `parameters_json`, `severity` (`INFO`, `WARNING`, `ERROR`, `CRITICAL`) and `enabled`. It targets either:
+
+- `table_name`, optional core `filter_json`, and `metric_json`, such as `{"operator":"COUNT"}`, `{"operator":"SUM","field":"amount"}`, `{"operator":"AVG","field":"amount"}`, `{"operator":"NULL_SHARE","field":"email"}`, or `{"operator":"CATEGORY_SET","field":"status"}`.
+- `saved_id`: the owner's saved table and filter, with an explicit watch `metric_json` or one saved definition `metric` / single pivot `values` measure. The date field and bucket remain explicit on the watch. Ambiguous saved measures are rejected; saved filters cannot be overridden by `filter_json` on the watch.
+
+Detector parameters are validated by core pre-insert/update customizers, including partial updates. Unknown parameters and invalid metric/detector combinations return clear record errors.
+
+| Detector | Parameters and behavior |
+| --- | --- |
+| SPIKE / DROP | `k=14`, `threshold=3.5`, `relativeChange=0.5`; robust z = `0.6745 * (latest - median) / MAD`, strict directional threshold. With zero MAD, compare relative change against the absolute median. A zero baseline fires only for a nonzero change in the requested direction. |
+| OUT_OF_RANGE | Required `min` and `max`; fire outside the inclusive range. Needs only the latest bucket. |
+| NEW_CATEGORY | `k=14`; requires `CATEGORY_SET`; compare latest categories with the union of previous buckets. Null is excluded from the category set. |
+| NULL_RATE_CREEP | `k=14`, `delta=0.05`, `n=3`; requires `NULL_SHARE`; fire for a rise above the baseline median greater than delta, or `n` consecutive increases (`n+1` values). |
+
+`k` and `n` are bounded to 1–1000. The trailing window contains at most `k+1` complete buckets, excludes the current bucket, and follows the owner's time zone (Monday-based weeks). Leading buckets with no observed rows do not invent history. After the first observed bucket, missing COUNT/SUM/category buckets are zero/empty; AVG and NULL_SHARE with no population are unavailable. Too little history, incomplete streamed results, unavailable fields, and privacy-suppressed buckets never cause a recovery. No suppressed series, categories, values or drills leave the computation. All source reads enforce READ, Analyze, record locks and FULL field access, including filter fields.
+
+`data_analysis_evaluate_watches` is hidden and repeats every 900 seconds when `schedulerName` is configured. Give the scheduler session its standard process permission, `data_analysis_evaluate_watches.hasAccess`. Inventory and status writes use grants restricted to watch rows; source aggregation always switches to the owner. The default integration calls the host authentication module's `createAutomatedSessionForUser`. Hosts can instead configure `withOwnerSessionSupplier(userId -> session)`. The session must identify that owner and include the `data_analysis_user` owner key plus their current permissions. Missing/mismatched sessions record **Owner session unavailable**; there is no system-session fallback. Manual Evaluate now can also evaluate a disabled watch.
+
+A transition from OK to FIRING raises through core `QAlerts`, with the watch record reference and a relative table-query `AlertLink`; recovery resolves the same `data_analysis_watch:<id>` key. The declared type is **`data.analysis.anomaly`**, labelled **Data anomaly**. This spelling is necessary because the supplied core SPI validator rejects the design's `data_analysis.anomaly` underscore. No qbit-alerts dependency is required. With no sink installed, core logs the signal and the watch row still records the outcome. Repeated FIRING/OK evaluations do not send duplicate transitions. Calls are serialized within each runtime; hosts should run this scheduler on one node (cross-node transactional coordination is not provided).
+
+`state`, `alert_key`, `last_evaluated_at`, `last_value` and `last_message` describe the most recent attempt. Skips clear the last value and leave the existing OK/FIRING state intact. Preview uses the requesting viewer's permissions and never updates these fields.
+
+`DataAnalysisSchema.postgresScript(config)` generates the new fields and owner requirement. Existing installations must add `date_field VARCHAR(250)`, `owner_user_id VARCHAR(250)`, `last_value NUMERIC`, and `last_message TEXT`, then backfill each watch's actual owner/date field before making the first two columns NOT NULL and enabling evaluation. The CREATE-IF-NOT-EXISTS generator does not migrate existing tables.
+
 ## Adopting in another app
 
 1. Add a backend, install the producer and run the usual `QInstanceValidator`.
@@ -210,7 +241,7 @@ The minimum-group-size guard removes small numeric summaries, suppresses crossta
 4. Review per-table include/exclude sets, limits and minimum group sizes before enabling analysis. Integrate Saved Analyses and Watches with host ESB/Quick Search when those optional host features are installed; this scaffold does not add those dependencies.
 5. Use the Analyses home as the owner of analysis data. Other dashboards should link there until saved-analysis widgets are implemented.
 
-Phase **b** profiles, drill-through, runs/cache and HTTP endpoints, phase **c** pivot backend, and phase **d** compare/relationships backend are implemented. Remaining work: **b** analysis UI; **c** pivot UI; **d** comparison/relationship UI; **e** watch evaluation and alerts; **f** saved-analysis execution, sharing, stars integration, schedules, chart widgets, full home and exports. The current home summarizes stored rows; the host facade executes profiles. The home does not send deliveries.
+Phase **b** profiles, drill-through, runs/cache and HTTP endpoints, phase **c** pivot backend, phase **d** compare/relationships backend, and phase **e** anomaly watches are implemented. Remaining work: **b** analysis UI; **c** pivot UI; **d** comparison/relationship UI; **f** saved-analysis execution, sharing, stars integration, schedules, chart widgets, full home and exports. The current home summarizes stored rows; the host facade executes profiles. The home does not send deliveries.
 
 ## Build
 

src/main/java/com/kingsrook/qbits/dataanalysis/DataAnalysisQBitConfig.java

@@ -39,6 +39,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitConfig;
 public class DataAnalysisQBitConfig implements QBitConfig
 {
    private Supplier<FieldAnalysisAccessResolver> fieldAnalysisAccessResolver = CoreTierFieldAccessResolver::defaultResolver;
+   private transient java.util.function.Function<String, com.kingsrook.qqq.backend.core.model.session.QSession> ownerSessionSupplier;
    private String backendName;
    private String schedulerName;
    private String appName = "dataAnalysis";
@@ -408,4 +409,21 @@ public class DataAnalysisQBitConfig implements QBitConfig
       this.fieldAnalysisAccessResolver = java.util.Objects.requireNonNull(resolver);
       return (this);
    }
+
+   /*******************************************************************************
+    ** Optional host adapter for scheduled owner permissions and record-lock keys.
+    *******************************************************************************/
+   public java.util.function.Function<String, com.kingsrook.qqq.backend.core.model.session.QSession> getOwnerSessionSupplier()
+   {
+      return (ownerSessionSupplier);
+   }
+
+   /*******************************************************************************
+    ** Without an adapter, use the host authentication module's automated session.
+    *******************************************************************************/
+   public DataAnalysisQBitConfig withOwnerSessionSupplier(java.util.function.Function<String, com.kingsrook.qqq.backend.core.model.session.QSession> supplier)
+   {
+      ownerSessionSupplier = supplier;
+      return (this);
+   }
 }

src/main/java/com/kingsrook/qbits/dataanalysis/DataAnalysisQBitProducer.java

@@ -29,6 +29,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.permissions.TableFeaturePer
 import com.kingsrook.qqq.backend.core.model.metadata.permissions.TableFeaturePermissionsMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitMetaDataProducer;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.TablesPossibleValueSourceMetaDataProvider;
 
 
 /*******************************************************************************
@@ -129,9 +130,20 @@ public class DataAnalysisQBitProducer implements QBitMetaDataProducer<DataAnalys
    public MetaDataProducerMultiOutput produce(QInstance instance) throws QException
    {
       qBitConfig.validate(instance);
+      if(instance.getPossibleValueSource(TablesPossibleValueSourceMetaDataProvider.NAME) == null)
+      {
+         instance.addPossibleValueSource(TablesPossibleValueSourceMetaDataProvider.defineTablesPossibleValueSource(instance));
+      }
       MetaDataProducerMultiOutput output = QBitMetaDataProducer.super.produce(instance);
       output.addSelfToInstance(instance);
       new DataAnalysisRuntime(qBitConfig).addSelfToInstance(instance);
+      com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData.ofOrWithNew(instance).withAlertType(
+         new com.kingsrook.qqq.backend.core.alerts.QAlertTypeMetaData().withName(com.kingsrook.qbits.dataanalysis.watch.WatchEvaluator.ALERT_TYPE)
+            .withLabel("Data anomaly").withSourceName(ARTIFACT_ID).withGroupName("data-analysis")
+            .withDefaultSeverity(com.kingsrook.qqq.backend.core.alerts.AlertSeverity.WARNING)
+            .withResolveMode(com.kingsrook.qqq.backend.core.alerts.ResolveMode.PRODUCER)
+            .withSubjectTableName(com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables.WATCH)
+            .withVisibilityPermission("data_analysis.watch.read"));
       var included = java.util.Set.copyOf(qBitConfig.getIncludeTables());
       var excluded = java.util.Set.copyOf(qBitConfig.getExcludeTables());
       TableFeaturePermissionsMetaData.ofOrWithNew(instance).withPermission(new TableFeaturePermission()

src/main/java/com/kingsrook/qbits/dataanalysis/DataAnalysisRuntime.java

@@ -30,6 +30,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.QSupplementalInstanceMetaDa
 public class DataAnalysisRuntime implements QSupplementalInstanceMetaData
 {
    private final transient DataAnalysisService service;
+   private final transient com.kingsrook.qbits.dataanalysis.watch.WatchEvaluator watches;
 
    /*******************************************************************************
     ** Create a runtime for one configured instance.
@@ -37,6 +38,7 @@ public class DataAnalysisRuntime implements QSupplementalInstanceMetaData
    public DataAnalysisRuntime(DataAnalysisQBitConfig config)
    {
       service = new DataAnalysisService(config);
+      watches = new com.kingsrook.qbits.dataanalysis.watch.WatchEvaluator(config, service);
    }
 
 
@@ -70,4 +72,12 @@ public class DataAnalysisRuntime implements QSupplementalInstanceMetaData
    {
       return (((DataAnalysisRuntime) QContext.getQInstance().getSupplementalMetaData("dataAnalysisRuntime")).getService());
    }
+
+   /*******************************************************************************
+    ** Shared evaluator serializes manual and scheduled watch transitions.
+    *******************************************************************************/
+   public static com.kingsrook.qbits.dataanalysis.watch.WatchEvaluator watches()
+   {
+      return (((DataAnalysisRuntime) QContext.getQInstance().getSupplementalMetaData("dataAnalysisRuntime")).watches);
+   }
 }

src/main/java/com/kingsrook/qbits/dataanalysis/DataAnalysisSchema.java

@@ -83,6 +83,7 @@ public final class DataAnalysisSchema
       {
          case INTEGER -> "INTEGER";
          case LONG -> "BIGINT";
+         case DECIMAL -> "NUMERIC";
          case BOOLEAN -> "BOOLEAN";
          case DATE_TIME -> "TIMESTAMP WITHOUT TIME ZONE";
          case TEXT -> "TEXT";

src/main/java/com/kingsrook/qbits/dataanalysis/DataAnalysisService.java

@@ -338,4 +338,17 @@ public class DataAnalysisService
       validate(request);
       return (new ShiftExplorer(security, configs.resolve(request.getTableName()), request.getTableName()).shift(request));
    }
+
+   /*******************************************************************************
+    ** Watch aggregates use the same current-user table and FULL-field policy.
+    *******************************************************************************/
+   public com.kingsrook.qbits.dataanalysis.watch.WatchPreview watchPreview(com.kingsrook.qqq.backend.core.model.data.QRecord row, java.time.Instant now) throws QException
+   {
+      var resolved = com.kingsrook.qbits.dataanalysis.watch.WatchRecords.resolve(row);
+      String table = resolved.getValueString("table_name");
+      security.table(table);
+      var definition = com.kingsrook.qbits.dataanalysis.watch.WatchDefinition.parse(resolved);
+      return (new com.kingsrook.qbits.dataanalysis.watch.WatchSeries(security, configs.resolve(table), table)
+         .preview(definition, com.kingsrook.qbits.dataanalysis.watch.WatchRecords.filter(resolved), now));
+   }
 }

src/main/java/com/kingsrook/qbits/dataanalysis/api/DataAnalysisApi.java

@@ -76,11 +76,11 @@ public class DataAnalysisApi
     *******************************************************************************/
    public static MiddlewareSpecExtension extension()
    {
-      return (new MiddlewareSpecExtension().withName("data-analysis").withSemanticVersion("1.2.0")
-         .withDescription("Profiles, pivots, comparisons, distribution shifts, correlations, scatter samples and crosstabs as the signed-in user.")
+      return (new MiddlewareSpecExtension().withName("data-analysis").withSemanticVersion("1.3.0")
+         .withDescription("Profiles, pivots, comparisons, distribution shifts, correlations, scatter samples, crosstabs and anomaly watch previews as the signed-in user.")
          .withSpec(CapabilitiesSpec.class).withSpec(ColumnProfileSpec.class).withSpec(DistinctValuesSpec.class)
          .withSpec(CompareSpec.class).withSpec(ShiftSpec.class).withSpec(CorrelationSpec.class).withSpec(ScatterSpec.class).withSpec(CrosstabSpec.class)
-         .withSpec(PivotSpec.class).withSpec(TableProfileSpec.class).withSpec(DuplicatesSpec.class).withSpec(StartRunSpec.class).withSpec(GetRunSpec.class));
+         .withSpec(WatchPreviewSpec.class).withSpec(PivotSpec.class).withSpec(TableProfileSpec.class).withSpec(DuplicatesSpec.class).withSpec(StartRunSpec.class).withSpec(GetRunSpec.class));
    }
 
    /*******************************************************************************
@@ -123,9 +123,9 @@ public class DataAnalysisApi
       @Override
       public List<Parameter> defineRequestParameters()
       {
-         String name = path.contains("{table}") ? "table" : path.contains("{runId}") ? "runId" : null;
+         String name = path.contains("{table}") ? "table" : path.contains("{runId}") ? "runId" : path.contains("{id}") ? "id" : null;
          return (name == null ? List.of() : List.of(new Parameter().withName(name).withIn(In.PATH).withRequired(true)
-            .withDescription(name.equals("table") ? "QQQ table name to analyze." : "Owner-scoped analysis run id.").withSchema(new Schema().withType(Type.STRING))));
+            .withDescription(name.equals("table") ? "QQQ table name to analyze." : "Owner-scoped analysis run or watch id.").withSchema(new Schema().withType(Type.STRING))));
       }
 
       /*******************************************************************************
@@ -757,4 +757,37 @@ public class DataAnalysisApi
          return (transport(DataAnalysisRuntime.service().crosstab(request), RelationshipSchemas.DataAnalysisCrosstabV1.class));
       }
    }
+
+   /*******************************************************************************
+    ** Read-only watch preview; standard table endpoints own watch CRUD.
+    *******************************************************************************/
+   public static class WatchPreviewSpec extends ProfileSpec
+   {
+      /*******************************************************************************
+       ** Publish a side-effect-free detector preview for a saved watch.
+       *******************************************************************************/
+      public WatchPreviewSpec()
+      {
+         super("/watches/{id}/preview", HttpMethod.POST, "Preview an anomaly watch",
+            "Compute complete time buckets and a detector verdict under the signed-in user's READ, Analyze and FULL-field permissions. Includes a baseline band annotation. Does not update the watch or raise alerts.", null, "WatchPreview");
+      }
+
+      /*******************************************************************************
+       ** Missing or inaccessible watch ids return the same not-found response.
+       *******************************************************************************/
+      @Override
+      protected Object handle(JsonInput input) throws Exception
+      {
+         Integer id;
+         try
+         {
+            id = Integer.valueOf(input.getPathParam("id"));
+         }
+         catch(NumberFormatException e)
+         {
+            throw new QBadRequestException("Watch id must be an integer.");
+         }
+         return (transport(DataAnalysisRuntime.watches().preview(id, java.time.Instant.now()), WatchSchemas.DataAnalysisWatchPreviewV1.class));
+      }
+   }
 }

src/main/java/com/kingsrook/qbits/dataanalysis/api/DataAnalysisSchemas.java

@@ -115,6 +115,7 @@ public class DataAnalysisSchemas
       result.put("DataAnalysisPivotCellV1", new DataAnalysisPivotCellV1().toSchema());
       result.put("DataAnalysisPivotV1", new DataAnalysisPivotV1().toSchema());
       result.putAll(RelationshipSchemas.components());
+      result.putAll(WatchSchemas.components());
       return (result);
    }
 

src/main/java/com/kingsrook/qbits/dataanalysis/api/WatchSchemas.java

@@ -0,0 +1,126 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.api;
+
+import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qqq.middleware.javalin.schemabuilder.ToSchema;
+import com.kingsrook.qqq.openapi.model.Schema;
+import com.kingsrook.qqq.openapi.model.Type;
+
+/*******************************************************************************
+ ** Standalone watch preview contracts referencing the existing chart and drill.
+ *******************************************************************************/
+public class WatchSchemas
+{
+   /*******************************************************************************
+    ** Register every referenced version-one watch response component.
+    *******************************************************************************/
+   public static Map<String, Schema> components()
+   {
+      return (Map.of("DataAnalysisWatchPreviewV1", new DataAnalysisWatchPreviewV1().toSchema(),
+         "DataAnalysisWatchBucketV1", new DataAnalysisWatchBucketV1().toSchema(),
+         "DataAnalysisWatchVerdictV1", new DataAnalysisWatchVerdictV1().toSchema()));
+   }
+
+   /*******************************************************************************
+    ** One complete bucket; suppressed series never expose categories or drills.
+    *******************************************************************************/
+   public static class DataAnalysisWatchBucketV1 implements ToSchema
+   {
+      public Serializable start;
+      public Double value;
+      public List<String> categories;
+      public DataAnalysisSchemas.DataAnalysisDrillV1 drill;
+
+      /*******************************************************************************
+       ** Explicit nullable types and component references.
+       *******************************************************************************/
+      @Override
+      public Schema toSchema()
+      {
+         Map<String, Schema> properties = new LinkedHashMap<>();
+         properties.put("start", new Schema().withType(Type.STRING).withDescription("Calendar date or hour-start instant in the user's time zone."));
+         properties.put("value", new Schema().withType(Type.NUMBER).withNullable(true));
+         properties.put("categories", new Schema().withType(Type.ARRAY).withItems(new Schema().withType(Type.STRING)));
+         properties.put("drill", new Schema().withRef("#/components/schemas/DataAnalysisDrillV1").withNullable(true));
+         return (new Schema().withType(Type.OBJECT).withProperties(properties));
+      }
+   }
+
+   /*******************************************************************************
+    ** Unevaluated means no state change; baseline bounds describe the annotation.
+    *******************************************************************************/
+   public static class DataAnalysisWatchVerdictV1 implements ToSchema
+   {
+      public Boolean evaluated;
+      public Boolean firing;
+      public Double value;
+      public Double baseline;
+      public Double baselineLow;
+      public Double baselineHigh;
+      public String message;
+      public DataAnalysisSchemas.DataAnalysisDrillV1 drill;
+
+      /*******************************************************************************
+       ** Carry numeric baseline bounds as well as the plain-English explanation.
+       *******************************************************************************/
+      @Override
+      public Schema toSchema()
+      {
+         Map<String, Schema> properties = new LinkedHashMap<>();
+         properties.put("evaluated", new Schema().withType(Type.BOOLEAN).withDescription("False for insufficient history or privacy suppression; must not resolve an alert."));
+         properties.put("firing", new Schema().withType(Type.BOOLEAN));
+         for(String field : List.of("value", "baseline", "baselineLow", "baselineHigh"))
+         {
+            properties.put(field, new Schema().withType(Type.NUMBER).withNullable(true));
+         }
+         properties.put("message", new Schema().withType(Type.STRING));
+         properties.put("drill", new Schema().withRef("#/components/schemas/DataAnalysisDrillV1").withNullable(true));
+         return (new Schema().withType(Type.OBJECT).withProperties(properties));
+      }
+   }
+
+   /*******************************************************************************
+    ** Complete preview, using the same semantic chart as other analysis endpoints.
+    *******************************************************************************/
+   public static class DataAnalysisWatchPreviewV1 implements ToSchema
+   {
+      public List<DataAnalysisWatchBucketV1> series;
+      public DataAnalysisSchemas.DataAnalysisChartV1 chart;
+      public DataAnalysisWatchVerdictV1 verdict;
+
+      /*******************************************************************************
+       ** Stable references let clients vendor this fragment without middleware.
+       *******************************************************************************/
+      @Override
+      public Schema toSchema()
+      {
+         Map<String, Schema> properties = new LinkedHashMap<>();
+         properties.put("series", new Schema().withType(Type.ARRAY).withItems(new Schema().withRef("#/components/schemas/DataAnalysisWatchBucketV1")));
+         properties.put("chart", new Schema().withRef("#/components/schemas/DataAnalysisChartV1").withDescription("TIME_SERIES with baseline band annotations."));
+         properties.put("verdict", new Schema().withRef("#/components/schemas/DataAnalysisWatchVerdictV1"));
+         return (new Schema().withType(Type.OBJECT).withProperties(properties));
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/app/DataAnalysisHome.java

@@ -158,6 +158,27 @@ public class DataAnalysisHome extends AbstractWidgetRenderer
          {
             html.append("<p>No watches yet.</p>");
          }
+         QueryInput watches = new QueryInput(DataAnalysisTables.WATCH).withShouldFetchHeavyFields(true)
+            .withFilter(new QQueryFilter().withOrderBy(new QFilterOrderBy("state", true)).withOrderBy(new QFilterOrderBy("id", false)).withLimit(5));
+         watches.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+         List<QRecord> recent = new QueryAction().execute(watches).getRecords();
+         if(!recent.isEmpty())
+         {
+            html.append("<ul>");
+            for(QRecord watch : recent)
+            {
+               String path = QContext.getQInstance().getTablePath(DataAnalysisTables.WATCH) + "/" + watch.getValueInteger("id");
+               html.append("<li><a href=\"").append(escape(path)).append("\">")
+                  .append(escape(words(watch.getValueString("state")))).append(" &middot; ")
+                  .append(escape(words(watch.getValueString("detector")))).append("</a>");
+               if(watch.getValueString("last_message") != null)
+               {
+                  html.append(" — ").append(escape(watch.getValueString("last_message")));
+               }
+               html.append("</li>");
+            }
+            html.append("</ul>");
+         }
          link(html, DataAnalysisTables.WATCH);
       }
       else

src/main/java/com/kingsrook/qbits/dataanalysis/engine/AnalysisLabels.java

@@ -85,6 +85,10 @@ public final class AnalysisLabels
       {
          return (null);
       }
+      if(field.getType() == com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType.BOOLEAN)
+      {
+         return (com.kingsrook.qqq.backend.core.utils.ValueUtils.getValueAsBoolean(value) ? "Yes" : "No");
+      }
       String fallback = value.toString();
       QPossibleValueSource source = QContext.getQInstance().getPossibleValueSource(field.getPossibleValueSourceName());
       if(source == null || visited.size() >= 8 || !visited.add(source.getName()))

src/main/java/com/kingsrook/qbits/dataanalysis/engine/PivotExplorer.java

@@ -326,7 +326,42 @@ public class PivotExplorer
       measures.add(aggregate);
       ValueType type = operator == PivotMeasure.Operator.COUNT || operator == PivotMeasure.Operator.COUNT_DISTINCT ? ValueType.COUNT
          : numeric(field) ? ValueType.DECIMAL : field.getType() == QFieldType.DATE ? ValueType.DATE : field.getType() == QFieldType.DATE_TIME ? ValueType.DATE_TIME : ValueType.TEXT;
-      measureInfo.add(new PivotMeasureInfo().withKey("value" + index).withLabel(operator.name() + (field == null ? " records" : " " + field.getLabel())).withValueType(type));
+      measureInfo.add(new PivotMeasureInfo().withKey("value" + index).withLabel(measureLabel(measure, field)).withValueType(type));
+   }
+
+   /*******************************************************************************
+    ** Human measure labels shared by headers, totals and cells.
+    *******************************************************************************/
+   private static String measureLabel(PivotMeasure measure, QFieldMetaData field)
+   {
+      String label = field == null ? "" : field.getLabel();
+      return (switch(measure.getOperator())
+      {
+         case COUNT -> "Count";
+         case AVG -> "Average " + label;
+         case SUM -> "Sum of " + label;
+         case COUNT_DISTINCT -> "Distinct " + label;
+         case MIN -> "Minimum " + label;
+         case MAX -> "Maximum " + label;
+         case PERCENTILE -> measure.getPercentile().compareTo(new BigDecimal("0.5")) == 0 ? "Median " + label
+            : ordinal(measure.getPercentile().multiply(BigDecimal.valueOf(100))) + " percentile of " + label;
+      });
+   }
+
+   /*******************************************************************************
+    ** English ordinal suffixes, including the eleventh through thirteenth.
+    *******************************************************************************/
+   private static String ordinal(BigDecimal value)
+   {
+      int number = value.intValue();
+      String suffix = number % 100 >= 11 && number % 100 <= 13 ? "th" : switch(number % 10)
+      {
+         case 1 -> "st";
+         case 2 -> "nd";
+         case 3 -> "rd";
+         default -> "th";
+      };
+      return (value.stripTrailingZeros().toPlainString() + suffix);
    }
 
    /*******************************************************************************

src/main/java/com/kingsrook/qbits/dataanalysis/engine/RelationshipSupport.java

@@ -177,7 +177,7 @@ public class RelationshipSupport extends StandInAnalysisBackend
          Serializable value = row.getGroupByValues().get(group);
          long n = ((Number) row.getAggregateValues().get(count)).longValue();
          boolean hidden = n < config.getMinGroupSize();
-         Category category = new Category(value, hidden ? "Other" : value == null ? "(blank)" : security.label(metadata, value),
+         Category category = new Category(value, hidden ? "Other" : value == null ? "(Blank)" : security.label(metadata, value),
             DrillFilters.value(field, value, metadata.getType() == QFieldType.STRING), hidden);
          if(hidden || result.size() >= limit)
          {

src/main/java/com/kingsrook/qbits/dataanalysis/metadata/WatchProcesses.java

@@ -0,0 +1,93 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.metadata;
+
+import java.util.List;
+import com.kingsrook.qbits.dataanalysis.DataAnalysisQBitConfig;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qbits.dataanalysis.processes.WatchActions;
+import com.kingsrook.qqq.backend.core.model.metadata.MetaDataProducerMultiOutput;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QIcon;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.PermissionLevel;
+import com.kingsrook.qqq.backend.core.model.metadata.permissions.QPermissionRules;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QBackendStepMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.processes.QProcessMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.qbits.QBitComponentMetaDataProducer;
+import com.kingsrook.qqq.backend.core.model.metadata.scheduleing.QScheduleMetaData;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ExtractViaQueryStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess;
+
+/*******************************************************************************
+ ** Watch record actions and one hidden scheduled process, never app children.
+ *******************************************************************************/
+public class WatchProcesses extends QBitComponentMetaDataProducer<MetaDataProducerMultiOutput, DataAnalysisQBitConfig>
+{
+   /*******************************************************************************
+    ** Build after the watch table has been registered.
+    *******************************************************************************/
+   @Override
+   public int getSortOrder()
+   {
+      return (200);
+   }
+
+   /*******************************************************************************
+    ** Component scanning also encounters unconfigured producers.
+    *******************************************************************************/
+   @Override
+   public boolean isEnabled()
+   {
+      return (getQBitConfig() != null);
+   }
+
+   /*******************************************************************************
+    ** Ordinary core actions provide selected records, Review, and Result screens.
+    *******************************************************************************/
+   @Override
+   public MetaDataProducerMultiOutput produce(QInstance instance)
+   {
+      MetaDataProducerMultiOutput output = new MetaDataProducerMultiOutput();
+      for(String name : List.of(WatchActions.EVALUATE, WatchActions.ENABLE, WatchActions.DISABLE))
+      {
+         String label = WatchActions.EVALUATE.equals(name) ? "Evaluate now" : WatchActions.ENABLE.equals(name) ? "Enable" : "Disable";
+         QProcessMetaData process = StreamedETLWithFrontendProcess.processMetaDataBuilder()
+            .withName(name).withLabel(label).withIcon(new QIcon("notifications_active"))
+            .withTableName(DataAnalysisTables.WATCH).withSourceTable(DataAnalysisTables.WATCH).withDestinationTable(DataAnalysisTables.WATCH)
+            .withExtractStepClass(ExtractViaQueryStep.class).withTransformStepClass(WatchActions.Review.class).withLoadStepClass(WatchActions.Result.class)
+            .withSupportsFullValidation(true).withDoFullValidation(true).withMinInputRecords(1)
+            .withPreviewMessage("Review the selected watches.").getProcessMetaData();
+         // Table READ/EDIT and owner locks are enforced by extraction and both steps.
+         process.withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.NOT_PROTECTED));
+         output.add(process);
+      }
+      QProcessMetaData scheduled = new QProcessMetaData().withName(WatchActions.SCHEDULED).withLabel("Evaluate watches").withIsHidden(true)
+         .withPermissionRules(new QPermissionRules().withLevel(PermissionLevel.HAS_ACCESS_PERMISSION).withPermissionBaseName(WatchActions.SCHEDULED))
+         .withStep(new QBackendStepMetaData().withName("run").withCode(new QCodeReference(WatchActions.Scheduled.class)));
+      if(getQBitConfig().getSchedulerName() != null)
+      {
+         scheduled.withSchedule(new QScheduleMetaData().withSchedulerName(getQBitConfig().getSchedulerName()).withRepeatSeconds(900).withInitialDelaySeconds(900));
+      }
+      output.add(scheduled);
+      return (output);
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/model/DataAnalysisFieldPossibleValueProvider.java

@@ -0,0 +1,112 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package com.kingsrook.qbits.dataanalysis.model;
+
+
+import java.io.Serializable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import com.kingsrook.qqq.backend.core.actions.values.QCustomPossibleValueProvider;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.values.SearchPossibleValueSourceInput;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValue;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.utils.CollectionUtils;
+import com.kingsrook.qqq.backend.core.utils.StringUtils;
+import com.kingsrook.qqq.backend.core.utils.ValueUtils;
+
+
+/*******************************************************************************
+ ** Fields of the table selected in the settings form. Core has no multi-select
+ ** PVS field yet; settings retain their comma-separated TEXT representation.
+ *******************************************************************************/
+public class DataAnalysisFieldPossibleValueProvider implements QCustomPossibleValueProvider<String>
+{
+   public static final String NAME = "data_analysis_fields";
+
+
+
+   /*******************************************************************************
+    ** Without form values the table is unknown; preserve names, including CSV lists.
+    *******************************************************************************/
+   @Override
+   public QPossibleValue<String> getPossibleValue(Serializable idValue)
+   {
+      String name = ValueUtils.getValueAsString(idValue);
+      return (name == null ? null : new QPossibleValue<>(name, name));
+   }
+
+
+
+   /*******************************************************************************
+    ** Offer only fields of the selected table, honoring search text and id lookups.
+    *******************************************************************************/
+   @Override
+   public List<QPossibleValue<String>> search(SearchPossibleValueSourceInput input)
+   {
+      List<QPossibleValue<String>> values = new ArrayList<>();
+      QTableMetaData table = resolveTable(input.getOtherValues());
+      if(table == null)
+      {
+         return (values);
+      }
+      for(QFieldMetaData field : table.getFields().values())
+      {
+         String label = field.getLabel() == null ? field.getName() : field.getLabel();
+         QPossibleValue<String> value = new QPossibleValue<>(field.getName(), label + " (" + field.getName() + ")");
+         if(CollectionUtils.nullSafeHasContents(input.getIdList()))
+         {
+            if(input.getIdList().stream().anyMatch(id -> value.getId().equals(ValueUtils.getValueAsString(id))))
+            {
+               values.add(value);
+            }
+         }
+         else
+         {
+            String term = input.getSearchTerm();
+            if(!StringUtils.hasContent(term) || value.getLabel().toLowerCase(Locale.ROOT).contains(term.toLowerCase(Locale.ROOT)))
+            {
+               values.add(value);
+            }
+         }
+      }
+      return (values);
+   }
+
+
+
+   /*******************************************************************************
+    ** Resolve the table from the frontend form values, never from the config table.
+    *******************************************************************************/
+   private QTableMetaData resolveTable(Map<String, Serializable> otherValues)
+   {
+      if(otherValues == null || QContext.getQInstance() == null)
+      {
+         return (null);
+      }
+      String tableName = ValueUtils.getValueAsString(otherValues.get("table_name"));
+      return (StringUtils.hasContent(tableName) ? QContext.getQInstance().getTable(tableName) : null);
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/model/DataAnalysisTables.java

@@ -29,8 +29,12 @@ import com.kingsrook.qbits.dataanalysis.DataAnalysisQBitConfig;
 import com.kingsrook.qbits.dataanalysis.permissions.DataAnalysisPermissions;
 import com.kingsrook.qqq.backend.core.actions.dashboard.widgets.ChildRecordListRenderer;
 import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QFilterCriteria;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
 import com.kingsrook.qqq.backend.core.model.metadata.audits.AuditLevel;
 import com.kingsrook.qqq.backend.core.model.metadata.audits.QAuditRules;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
 import com.kingsrook.qqq.backend.core.model.metadata.dashboard.QWidgetMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
@@ -49,6 +53,7 @@ import com.kingsrook.qqq.backend.core.model.metadata.tables.ExposedJoin;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QFieldSection;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.SectionFactory;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.TablesPossibleValueSourceMetaDataProvider;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.Tier;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.UniqueKey;
 
@@ -111,10 +116,15 @@ public final class DataAnalysisTables
    public static List<QPossibleValueSource> possibleValueSources()
    {
       List<QPossibleValueSource> sources = new ArrayList<>();
+      sources.add(new QPossibleValueSource().withName(DataAnalysisFieldPossibleValueProvider.NAME)
+         .withType(QPossibleValueSourceType.CUSTOM).withIdType(QFieldType.STRING)
+         .withCustomCodeReference(new QCodeReference(DataAnalysisFieldPossibleValueProvider.class)));
       sources.add(enumSource("kind", "COLUMN_PROFILE", "TABLE_PROFILE", "PIVOT", "COMPARE", "CORRELATION", "CROSSTAB", "SHIFT"));
       sources.add(enumSource("chart_kind", "HISTOGRAM", "BOX", "BAR", "PARETO", "TIME_SERIES", "SCATTER", "HEATMAP"));
       sources.add(enumSource("detector", "SPIKE", "DROP", "OUT_OF_RANGE", "NEW_CATEGORY", "NULL_RATE_CREEP"));
       sources.add(enumSource("watch_state", "OK", "FIRING"));
+      sources.add(enumSource("watch_bucket", "HOUR", "DAY", "WEEK", "MONTH"));
+      sources.add(enumSource("watch_severity", "INFO", "WARNING", "ERROR", "CRITICAL"));
       sources.add(enumSource("run_status", "RUNNING", "COMPLETE", "FAILED"));
       sources.add(enumSource("schedule_frequency", "DAILY", "WEEKLY", "MONTHLY"));
       sources.add(enumSource("schedule_format", "HTML_SUMMARY", "CSV", "XLSX"));
@@ -205,14 +215,16 @@ public final class DataAnalysisTables
       List<QTableMetaData> tables = new ArrayList<>();
       tables.add(base(config, TABLE_CONFIG, "Analysis Settings", "settings", "data_analysis.settings")
          .withRecordLabelFormatAndFields("%s", "table_name")
-         .withField(field("table_name", QFieldType.STRING).withIsRequired(true))
+         .withField(field("table_name", QFieldType.STRING).withPossibleValueSourceName(TablesPossibleValueSourceMetaDataProvider.NAME).withIsRequired(true))
          .withField(field("enabled", QFieldType.BOOLEAN).withDefaultValue(true))
          .withField(field("max_rows", QFieldType.INTEGER))
          .withField(field("sample_threshold_rows", QFieldType.INTEGER))
          .withField(field("sample_rows", QFieldType.INTEGER))
          .withField(field("min_group_size", QFieldType.INTEGER))
-         .withField(field("analyzable_fields", QFieldType.TEXT))
-         .withField(field("excluded_fields", QFieldType.TEXT))
+         // add possibleValueSourceDependsOn("table_name") once core has it
+         .withField(field("analyzable_fields", QFieldType.TEXT).withPossibleValueSourceName(DataAnalysisFieldPossibleValueProvider.NAME))
+         // add possibleValueSourceDependsOn("table_name") once core has it
+         .withField(field("excluded_fields", QFieldType.TEXT).withPossibleValueSourceName(DataAnalysisFieldPossibleValueProvider.NAME))
          .withField(field("cache_ttl_minutes", QFieldType.INTEGER))
          .withAuditRules(new QAuditRules().withAuditLevel(AuditLevel.FIELD))
          .withUniqueKey(new UniqueKey("table_name")));
@@ -220,7 +232,7 @@ public final class DataAnalysisTables
          .withRecordLabelFormatAndFields("%s", "name")
          .withField(field("name", QFieldType.STRING).withIsRequired(true))
          .withField(field("description", QFieldType.TEXT))
-         .withField(field("table_name", QFieldType.STRING).withIsRequired(true))
+         .withField(field("table_name", QFieldType.STRING).withPossibleValueSourceName(TablesPossibleValueSourceMetaDataProvider.NAME).withIsRequired(true))
          .withField(field("filter_json", QFieldType.TEXT))
          .withField(field("kind", QFieldType.STRING).withPossibleValueSourceName("data_analysis_kind").withIsRequired(true))
          .withField(field("definition_json", QFieldType.TEXT).withIsRequired(true))
@@ -252,21 +264,31 @@ public final class DataAnalysisTables
          .withRecordSecurityLock(ownerLock(SAVED + ".owner_user_id").withJoinNameChain(List.of(joinName(SCHEDULE)))));
       tables.add(base(config, WATCH, "Watches", "notifications_active", "data_analysis.watch")
          .withRecordLabelFormatAndFields("%s", "detector")
-         .withField(field("saved_id", QFieldType.INTEGER).withPossibleValueSourceName("data_analysis_saved"))
-         .withField(field("table_name", QFieldType.STRING))
+         .withField(field("saved_id", QFieldType.INTEGER).withPossibleValueSourceName("data_analysis_saved")
+            .withPossibleValueSourceFilter(new QQueryFilter(new QFilterCriteria("table_name", QCriteriaOperator.EQUALS, "${input.table_name}"))))
+         .withField(field("table_name", QFieldType.STRING).withPossibleValueSourceName(TablesPossibleValueSourceMetaDataProvider.NAME))
          .withField(field("filter_json", QFieldType.TEXT))
          .withField(field("metric_json", QFieldType.TEXT))
          .withField(field("detector", QFieldType.STRING).withPossibleValueSourceName("data_analysis_detector").withIsRequired(true))
          .withField(field("parameters_json", QFieldType.TEXT))
-         .withField(field("bucket", QFieldType.STRING))
-         .withField(field("severity", QFieldType.STRING))
+         .withField(field("date_field", QFieldType.STRING).withIsRequired(true))
+         .withField(field("bucket", QFieldType.STRING).withDefaultValue("DAY").withIsRequired(true).withPossibleValueSourceName("data_analysis_watch_bucket"))
+         .withField(field("severity", QFieldType.STRING).withDefaultValue("WARNING").withIsRequired(true).withPossibleValueSourceName("data_analysis_watch_severity"))
+         .withField(field("owner_user_id", QFieldType.STRING).withIsRequired(true))
          .withField(field("enabled", QFieldType.BOOLEAN).withDefaultValue(false))
-         .withField(field("last_evaluated_at", QFieldType.DATE_TIME))
-         .withField(field("state", QFieldType.STRING).withDefaultValue("OK").withPossibleValueSourceName("data_analysis_watch_state"))
-         .withField(field("alert_key", QFieldType.STRING)));
+         .withField(field("last_evaluated_at", QFieldType.DATE_TIME).withIsEditable(false))
+         .withField(field("last_value", QFieldType.DECIMAL).withIsEditable(false))
+         .withField(field("last_message", QFieldType.TEXT).withIsEditable(false))
+         .withField(field("state", QFieldType.STRING).withDefaultValue("OK").withIsEditable(false).withPossibleValueSourceName("data_analysis_watch_state"))
+         .withField(field("alert_key", QFieldType.STRING).withIsEditable(false))
+         .withRecordSecurityLock(ownerLock("owner_user_id"))
+         .withCustomizer(com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers.PRE_INSERT_RECORD,
+            new com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference(com.kingsrook.qbits.dataanalysis.watch.WatchCustomizer.class))
+         .withCustomizer(com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizers.PRE_UPDATE_RECORD,
+            new com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference(com.kingsrook.qbits.dataanalysis.watch.WatchCustomizer.class)));
       tables.add(base(config, RUN, "Analysis Runs", "history", "data_analysis.run")
          .withRecordLabelFormatAndFields("%s", "operation")
-         .withField(field("table_name", QFieldType.STRING).withIsRequired(true))
+         .withField(field("table_name", QFieldType.STRING).withPossibleValueSourceName(TablesPossibleValueSourceMetaDataProvider.NAME).withIsRequired(true))
          .withField(field("operation", QFieldType.STRING).withIsRequired(true))
          .withField(field("request_hash", QFieldType.STRING))
          .withField(field("filter_hash", QFieldType.STRING))

src/main/java/com/kingsrook/qbits/dataanalysis/processes/WatchActions.java

@@ -0,0 +1,186 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.processes;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import com.kingsrook.qbits.dataanalysis.DataAnalysisRuntime;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper;
+import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType;
+import com.kingsrook.qqq.backend.core.actions.processes.BackendStep;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLine;
+import com.kingsrook.qqq.backend.core.model.actions.processes.ProcessSummaryLineInterface;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.Status;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractLoadStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.AbstractTransformStep;
+import com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.ProcessSummaryProviderInterface;
+
+/*******************************************************************************
+ ** Ordinary streamed ETL Review/Result steps; no custom forms or navigation.
+ *******************************************************************************/
+public final class WatchActions
+{
+   public static final String EVALUATE = "data_analysis_watch_evaluate";
+   public static final String ENABLE = "data_analysis_watch_enable";
+   public static final String DISABLE = "data_analysis_watch_disable";
+   public static final String SCHEDULED = "data_analysis_evaluate_watches";
+
+   /*******************************************************************************
+    ** Steps only.
+    *******************************************************************************/
+   private WatchActions()
+   {
+   }
+
+   /*******************************************************************************
+    ** Preview selected records without evaluating or mutating them.
+    *******************************************************************************/
+   public static class Review extends AbstractTransformStep implements ProcessSummaryProviderInterface
+   {
+      private final ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+
+      /*******************************************************************************
+       ** Extract runs as caller; enforce EDIT before promising any action.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         if(!PermissionsHelper.hasTablePermission(new UpdateInput(DataAnalysisTables.WATCH), DataAnalysisTables.WATCH, TablePermissionSubType.EDIT))
+         {
+            throw new QPermissionDeniedException("Watch editing is unavailable.");
+         }
+         String verb = verb(input.getProcessName());
+         for(QRecord record : input.getRecords())
+         {
+            lines.add(new ProcessSummaryLine(Status.INFO, null, "Watch " + record.getValueString("id") + " (" + record.getValueString("detector") + ") will be " + verb + "."));
+            output.addRecord(record);
+         }
+      }
+
+      /*******************************************************************************
+       ** Standard summary lines are rendered by the core Review screen.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean result)
+      {
+         return (result ? new ArrayList<>() : lines);
+      }
+   }
+
+   /*******************************************************************************
+    ** Apply selected records and report exact evaluated/firing/OK/skipped counts.
+    *******************************************************************************/
+   public static class Result extends AbstractLoadStep implements ProcessSummaryProviderInterface
+   {
+      private int done;
+      private int firing;
+      private int ok;
+      private String process;
+      private final ArrayList<ProcessSummaryLineInterface> failures = new ArrayList<>();
+
+      /*******************************************************************************
+       ** Re-read and authorize each id when executing after Review.
+       *******************************************************************************/
+      @Override
+      public void runOnePage(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         process = input.getProcessName();
+         verb(process);
+         for(QRecord record : input.getRecords())
+         {
+            try
+            {
+               if(EVALUATE.equals(process))
+               {
+                  var verdict = DataAnalysisRuntime.watches().evaluate(record.getValue("id"), Instant.now());
+                  if(!verdict.evaluated())
+                  {
+                     failures.add(new ProcessSummaryLine(Status.WARNING, null, "Watch " + record.getValueString("id") + " skipped: " + verdict.message()));
+                     continue;
+                  }
+                  firing += verdict.firing() ? 1 : 0;
+                  ok += verdict.firing() ? 0 : 1;
+               }
+               else
+               {
+                  DataAnalysisRuntime.watches().setEnabled(record.getValue("id"), ENABLE.equals(process));
+               }
+               done++;
+               output.addRecord(record);
+            }
+            catch(QException e)
+            {
+               failures.add(new ProcessSummaryLine(Status.ERROR, null, "Watch " + record.getValueString("id") + " failed: " + e.getMessage()));
+            }
+         }
+      }
+
+      /*******************************************************************************
+       ** Counts span all streamed pages, so the Result screen is a true total.
+       *******************************************************************************/
+      @Override
+      public ArrayList<ProcessSummaryLineInterface> getProcessSummary(RunBackendStepOutput output, boolean result)
+      {
+         ArrayList<ProcessSummaryLineInterface> lines = new ArrayList<>();
+         String message = done + (done == 1 ? " watch " : " watches ") + (EVALUATE.equals(process) ? "evaluated: " + firing + " firing, " + ok + " OK"
+            : ENABLE.equals(process) ? "enabled" : "disabled");
+         lines.add(new ProcessSummaryLine(Status.OK, null, message));
+         lines.addAll(failures);
+         return (lines);
+      }
+   }
+
+   /*******************************************************************************
+    ** Scheduler invokes a hidden backend-only process every fifteen minutes.
+    *******************************************************************************/
+   public static class Scheduled implements BackendStep
+   {
+      /*******************************************************************************
+       ** Store the real count of successfully evaluated enabled watches.
+       *******************************************************************************/
+      @Override
+      public void run(RunBackendStepInput input, RunBackendStepOutput output) throws QException
+      {
+         output.addValue("count", DataAnalysisRuntime.watches().evaluateEnabled(Instant.now()));
+      }
+   }
+
+   /*******************************************************************************
+    ** Reject unknown processes instead of silently choosing a mutation.
+    *******************************************************************************/
+   private static String verb(String process) throws QException
+   {
+      return (switch(process)
+      {
+         case EVALUATE -> "evaluated";
+         case ENABLE -> "enabled";
+         case DISABLE -> "disabled";
+         default -> throw new QException("Unknown watch action.");
+      });
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchBucket.java

@@ -0,0 +1,32 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.io.Serializable;
+import java.util.Set;
+import com.kingsrook.qbits.dataanalysis.result.Drill;
+
+/*******************************************************************************
+ ** One complete calendar bucket, ordered oldest first for detector input.
+ *******************************************************************************/
+public record WatchBucket(Serializable start, Double value, Set<String> categories, Drill drill)
+{
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchCustomizer.java

@@ -0,0 +1,97 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qbits.dataanalysis.permissions.DataAnalysisPermissions;
+import com.kingsrook.qqq.backend.core.actions.customizers.TableCustomizerInterface;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.statusmessages.BadInputStatusMessage;
+
+/*******************************************************************************
+ ** Validate standard table CRUD, including partial updates and owner changes.
+ *******************************************************************************/
+public class WatchCustomizer implements TableCustomizerInterface
+{
+   private static final Set<String> CONFIGURATION = Set.of("saved_id", "table_name", "filter_json", "metric_json", "date_field", "bucket",
+      "detector", "parameters_json", "severity", "owner_user_id", "enabled");
+
+   /*******************************************************************************
+    ** Validate every inserted watch, including disabled drafts.
+    *******************************************************************************/
+   @Override
+   public List<QRecord> preInsert(InsertInput input, List<QRecord> records, boolean preview) throws QException
+   {
+      for(QRecord record : records)
+      {
+         validate(record, record);
+      }
+      return (records);
+   }
+
+   /*******************************************************************************
+    ** Merge partial patches with the stored row before checking detector parameters.
+    *******************************************************************************/
+   @Override
+   public List<QRecord> preUpdate(UpdateInput input, List<QRecord> records, boolean preview, Optional<List<QRecord>> oldRecords) throws QException
+   {
+      for(QRecord record : records)
+      {
+         if(record.getValues().keySet().stream().noneMatch(CONFIGURATION::contains))
+         {
+            continue;
+         }
+         QRecord old = WatchRecords.get(DataAnalysisTables.WATCH, record.getValue("id"));
+         QRecord merged = new QRecord(old);
+         merged.getValues().putAll(record.getValues());
+         validate(record, merged);
+      }
+      return (records);
+   }
+
+   /*******************************************************************************
+    ** Record-level messages are shown by normal core insert/update screens.
+    *******************************************************************************/
+   private void validate(QRecord output, QRecord merged)
+   {
+      try
+      {
+         String owner = merged.getValueString("owner_user_id");
+         var session = QContext.getQSession();
+         WatchDefinition.require(owner != null && !owner.isBlank(), "owner_user_id is required.");
+         WatchDefinition.require(session != null && (session.hasPermission(DataAnalysisPermissions.ADMIN)
+            || session.getUser() != null && owner.equals(session.getUser().getIdReference())), "Only an administrator may assign another watch owner.");
+         WatchDefinition.parse(WatchRecords.resolve(merged));
+      }
+      catch(QException e)
+      {
+         output.addError(new BadInputStatusMessage(e.getMessage()));
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchDefinition.java

@@ -0,0 +1,152 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import com.kingsrook.qqq.backend.core.alerts.AlertSeverity;
+import com.kingsrook.qqq.backend.core.exceptions.QBadRequestException;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
+
+/*******************************************************************************
+ ** Validated watch configuration; JSON accepts only documented, typed parameters.
+ *******************************************************************************/
+public record WatchDefinition(String detector, String metric, String field, String dateField, String bucket,
+   Integer k, Double threshold, Double relativeChange, Double min, Double max, Double delta, Integer n)
+{
+   /*******************************************************************************
+    ** Parse a definition after resolving an optional saved analysis.
+    *******************************************************************************/
+   public static WatchDefinition parse(QRecord row) throws QBadRequestException
+   {
+      String detector = row.getValueString("detector");
+      require(List.of("SPIKE", "DROP", "OUT_OF_RANGE", "NEW_CATEGORY", "NULL_RATE_CREEP").contains(detector == null ? "" : detector), "Choose a supported detector.");
+      Map<String, Object> metric = object(row.getValueString("metric_json"), "metric_json");
+      require(metric.keySet().stream().allMatch(Set.of("operator", "field")::contains), "metric_json accepts operator and field only.");
+      String operator = string(metric.get("operator"));
+      String field = string(metric.get("field"));
+      require(List.of("COUNT", "SUM", "AVG", "NULL_SHARE", "CATEGORY_SET").contains(operator == null ? "" : operator), "Metric operator must be COUNT, SUM, AVG, NULL_SHARE or CATEGORY_SET.");
+      require("COUNT".equals(operator) || field != null && !field.isBlank(), "This metric requires a field.");
+      require("NEW_CATEGORY".equals(detector) == "CATEGORY_SET".equals(operator), "NEW_CATEGORY requires CATEGORY_SET; other detectors require a numeric metric.");
+      require(!"NULL_RATE_CREEP".equals(detector) || "NULL_SHARE".equals(operator), "NULL_RATE_CREEP requires NULL_SHARE.");
+      String date = row.getValueString("date_field");
+      String bucket = row.getValueString("bucket");
+      require(date != null && !date.isBlank(), "Choose a date field.");
+      require(List.of("HOUR", "DAY", "WEEK", "MONTH").contains(bucket == null ? "" : bucket), "Bucket must be HOUR, DAY, WEEK or MONTH.");
+      try
+      {
+         AlertSeverity.valueOf(row.getValueString("severity"));
+      }
+      catch(RuntimeException e)
+      {
+         throw new QBadRequestException("Choose a valid alert severity.");
+      }
+      Map<String, Object> parameters = object(row.getValueString("parameters_json"), "parameters_json");
+      Set<String> allowed = switch(detector)
+      {
+         case "SPIKE", "DROP" -> Set.of("k", "threshold", "relativeChange");
+         case "OUT_OF_RANGE" -> Set.of("k", "min", "max");
+         case "NULL_RATE_CREEP" -> Set.of("k", "delta", "n");
+         default -> Set.of("k");
+      };
+      require(allowed.containsAll(parameters.keySet()), "Unsupported parameter for " + detector + ". Allowed: " + String.join(", ", allowed.stream().sorted().toList()) + ".");
+      int k = integer(parameters, "k", 14, 1, 1000);
+      int n = integer(parameters, "n", 3, 1, 1000);
+      double threshold = number(parameters, "threshold", 3.5);
+      double relative = number(parameters, "relativeChange", 0.5);
+      double delta = number(parameters, "delta", 0.05);
+      require(threshold > 0, "threshold must be positive.");
+      require(relative >= 0, "relativeChange must be non-negative (0.5 means 50%).");
+      require(delta >= 0 && delta <= 1, "delta must be between 0 and 1.");
+      Double min = parameters.containsKey("min") ? number(parameters, "min", 0) : null;
+      Double max = parameters.containsKey("max") ? number(parameters, "max", 0) : null;
+      require(!"OUT_OF_RANGE".equals(detector) || min != null && max != null && min <= max, "OUT_OF_RANGE requires min and max with min <= max.");
+      return (new WatchDefinition(detector, operator, field, date, bucket, k, threshold, relative, min, max, delta, n));
+   }
+
+   /*******************************************************************************
+    ** Strict JSON objects keep malformed definitions out of standard CRUD.
+    *******************************************************************************/
+   @SuppressWarnings("unchecked")
+   public static Map<String, Object> object(String json, String name) throws QBadRequestException
+   {
+      try
+      {
+         Object value = JsonUtils.toObject(json == null || json.isBlank() ? "{}" : json, Object.class);
+         if(value instanceof Map<?, ?> map)
+         {
+            return ((Map<String, Object>) map);
+         }
+      }
+      catch(Exception expected)
+      {
+         // Report a stable validation error, without echoing stored JSON.
+      }
+      throw new QBadRequestException(name + " must be a JSON object.");
+   }
+
+   /*******************************************************************************
+    ** Numeric values must be finite JSON numbers, never coerced strings.
+    *******************************************************************************/
+   private static double number(Map<String, Object> values, String key, double fallback) throws QBadRequestException
+   {
+      if(!values.containsKey(key))
+      {
+         return (fallback);
+      }
+      require(values.get(key) instanceof Number, key + " must be a number.");
+      double value = ((Number) values.get(key)).doubleValue();
+      require(Double.isFinite(value), key + " must be finite.");
+      return (value);
+   }
+
+   /*******************************************************************************
+    ** Bound work and reject fractional window sizes.
+    *******************************************************************************/
+   private static int integer(Map<String, Object> values, String key, int fallback, int min, int max) throws QBadRequestException
+   {
+      double value = number(values, key, fallback);
+      require(value == Math.rint(value) && value >= min && value <= max, key + " must be an integer from " + min + " to " + max + ".");
+      return ((int) value);
+   }
+
+   /*******************************************************************************
+    ** Optional strings still reject incompatible scalar types.
+    *******************************************************************************/
+   private static String string(Object value) throws QBadRequestException
+   {
+      require(value == null || value instanceof String, "Metric operator and field must be strings.");
+      return ((String) value);
+   }
+
+   /*******************************************************************************
+    ** Clear errors shared by table validation and preview.
+    *******************************************************************************/
+   public static void require(boolean valid, String message) throws QBadRequestException
+   {
+      if(!valid)
+      {
+         throw new QBadRequestException(message);
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchDetectors.java

@@ -0,0 +1,131 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import com.kingsrook.qbits.dataanalysis.engine.AnalysisFormatting;
+
+/*******************************************************************************
+ ** Pure detectors over complete buckets, with no context, queries or side effects.
+ *******************************************************************************/
+public final class WatchDetectors
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private WatchDetectors()
+   {
+   }
+
+   /*******************************************************************************
+    ** Compare the latest value to the preceding k complete buckets.
+    *******************************************************************************/
+   public static WatchVerdict evaluate(WatchDefinition definition, List<WatchBucket> series, String label)
+   {
+      int k = definition.k();
+      if(series.isEmpty() || !"OUT_OF_RANGE".equals(definition.detector()) && series.size() < k + 1)
+      {
+         return (WatchVerdict.skipped("Not enough history."));
+      }
+      WatchBucket latest = series.getLast();
+      List<WatchBucket> history = series.subList(Math.max(0, series.size() - k - 1), series.size() - 1);
+      if("NEW_CATEGORY".equals(definition.detector()))
+      {
+         Set<String> previous = new HashSet<>();
+         history.forEach(bucket -> previous.addAll(bucket.categories()));
+         long added = latest.categories().stream().filter(value -> !previous.contains(value)).count();
+         return (new WatchVerdict(true, added > 0, (double) latest.categories().size(), (double) previous.size(), null, null,
+            label + (added > 0 ? " has " + added + " new " + (added == 1 ? "category." : "categories.") : " has no new categories."), latest.drill()));
+      }
+      if(latest.value() == null || !Double.isFinite(latest.value()) || !"OUT_OF_RANGE".equals(definition.detector()) && history.stream().anyMatch(bucket -> bucket.value() == null || !Double.isFinite(bucket.value())))
+      {
+         return (WatchVerdict.skipped("Not enough history: a bucket has no numeric value."));
+      }
+      double x = latest.value();
+      double[] values = history.stream().filter(bucket -> bucket.value() != null && Double.isFinite(bucket.value())).mapToDouble(WatchBucket::value).sorted().toArray();
+      Double baseline = values.length == 0 ? null : median(values);
+      Double low = values.length == 0 ? null : values[0];
+      Double high = values.length == 0 ? null : values[values.length - 1];
+      boolean firing;
+      switch(definition.detector())
+      {
+         case "SPIKE", "DROP" ->
+         {
+            double mad = median(java.util.Arrays.stream(values).map(value -> Math.abs(value - baseline)).sorted().toArray());
+            double change = x - baseline;
+            double score = mad == 0 ? baseline == 0 ? Math.signum(change) * (change == 0 ? 0 : Double.POSITIVE_INFINITY)
+               : change / Math.abs(baseline) : 0.6745 * change / mad;
+            double threshold = mad == 0 ? definition.relativeChange() : definition.threshold();
+            firing = "SPIKE".equals(definition.detector()) ? score > threshold : score < -threshold;
+         }
+         case "OUT_OF_RANGE" ->
+         {
+            low = definition.min();
+            high = definition.max();
+            firing = x < low || x > high;
+         }
+         case "NULL_RATE_CREEP" -> firing = x - baseline > definition.delta() || rising(series, definition.n());
+         default -> throw new IllegalArgumentException("Unsupported numeric detector.");
+      }
+      String verb = firing ? "DROP".equals(definition.detector()) ? " dropped to " : "SPIKE".equals(definition.detector())
+         || "NULL_RATE_CREEP".equals(definition.detector()) ? " rose to " : " is out of range at " : " is ";
+      String band = low == null ? "" : " (" + ("OUT_OF_RANGE".equals(definition.detector()) ? "expected " : "usual ") + format(low) + "–" + format(high) + ")";
+      return (new WatchVerdict(true, firing, x, baseline, low, high, label + verb + format(x) + band + ".", latest.drill()));
+   }
+
+   /*******************************************************************************
+    ** n consecutive rises require n+1 values, including the latest bucket.
+    *******************************************************************************/
+   private static boolean rising(List<WatchBucket> series, int n)
+   {
+      if(series.size() <= n)
+      {
+         return (false);
+      }
+      for(int i = series.size() - n; i < series.size(); i++)
+      {
+         if(series.get(i).value() <= series.get(i - 1).value())
+         {
+            return (false);
+         }
+      }
+      return (true);
+   }
+
+   /*******************************************************************************
+    ** Median of a sorted non-empty array, including even-sized baselines.
+    *******************************************************************************/
+   private static double median(double[] values)
+   {
+      int middle = values.length / 2;
+      return (values.length % 2 == 0 ? values[middle - 1] / 2 + values[middle] / 2 : values[middle]);
+   }
+
+   /*******************************************************************************
+    ** Compact English numeric labels with no scientific notation.
+    *******************************************************************************/
+   private static String format(double value)
+   {
+      return (AnalysisFormatting.number(value, 3));
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchEvaluator.java

@@ -0,0 +1,236 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+import com.kingsrook.qbits.dataanalysis.DataAnalysisQBitConfig;
+import com.kingsrook.qbits.dataanalysis.DataAnalysisService;
+import com.kingsrook.qbits.dataanalysis.engine.DrillFilters;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qbits.dataanalysis.permissions.DataAnalysisPermissions;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionEnforcement;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionsHelper;
+import com.kingsrook.qqq.backend.core.actions.permissions.TablePermissionSubType;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.alerts.AlertLink;
+import com.kingsrook.qqq.backend.core.alerts.AlertRecordRef;
+import com.kingsrook.qqq.backend.core.alerts.AlertSeverity;
+import com.kingsrook.qqq.backend.core.alerts.AlertSignal;
+import com.kingsrook.qqq.backend.core.alerts.QAlerts;
+import com.kingsrook.qqq.backend.core.context.CapturedContext;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QPermissionDeniedException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.session.QUser;
+import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleDispatcher;
+
+/*******************************************************************************
+ ** Transition watches through the core alerts SPI, computing exclusively as owner.
+ *******************************************************************************/
+public class WatchEvaluator
+{
+   public static final String ALERT_TYPE = "data.analysis.anomaly";
+   private final DataAnalysisQBitConfig config;
+   private final DataAnalysisService service;
+
+   /*******************************************************************************
+    ** One evaluator per instance serializes scheduled and manual transitions.
+    *******************************************************************************/
+   public WatchEvaluator(DataAnalysisQBitConfig config, DataAnalysisService service)
+   {
+      this.config = config;
+      this.service = service;
+   }
+
+   /*******************************************************************************
+    ** Preview uses the viewer's rights, never the saved owner's cached data.
+    *******************************************************************************/
+   public WatchPreview preview(Serializable id, Instant now) throws QException
+   {
+      return (service.watchPreview(WatchRecords.get(DataAnalysisTables.WATCH, id), now));
+   }
+
+   /*******************************************************************************
+    ** Manual evaluation first requires the caller's read and edit rights.
+    *******************************************************************************/
+   public synchronized WatchVerdict evaluate(Serializable id, Instant now) throws QException
+   {
+      requireEdit();
+      QRecord row = WatchRecords.get(DataAnalysisTables.WATCH, id);
+      WatchVerdict verdict = asOwner(row, now);
+      store(row, verdict, now);
+      return (verdict);
+   }
+
+   /*******************************************************************************
+    ** Only watch inventory/status use maintenance grants, never source data.
+    *******************************************************************************/
+   public synchronized Integer evaluateEnabled(Instant now) throws QException
+   {
+      QSession maintenance = new QSession().withUser(new QUser().withIdReference("data-analysis-scheduler"))
+         .withPermissions("data_analysis.watch.read", "data_analysis.watch.edit")
+         .withSecurityKeyValue(DataAnalysisPermissions.ADMIN, true);
+      if(QContext.getQSession().getBackendVariants() != null)
+      {
+         maintenance.setBackendVariants(new java.util.HashMap<>(QContext.getQSession().getBackendVariants()));
+      }
+      int[] count = {0};
+      QContext.withTemporaryContext(new CapturedContext(QContext.getQInstance(), maintenance), () ->
+      {
+         QueryInput input = new QueryInput(DataAnalysisTables.WATCH).withShouldFetchHeavyFields(true)
+            .withFilter(DrillFilters.condition("enabled", QCriteriaOperator.EQUALS, true));
+         input.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+         for(QRecord row : new QueryAction().execute(input).getRecords())
+         {
+            WatchVerdict verdict = evaluate(row.getValue("id"), now);
+            if(verdict.evaluated())
+            {
+               count[0]++;
+            }
+         }
+      });
+      return (count[0]);
+   }
+
+   /*******************************************************************************
+    ** An unavailable or mismatched host session must never fall back to system.
+    *******************************************************************************/
+   private WatchVerdict asOwner(QRecord row, Instant now)
+   {
+      String owner = row.getValueString("owner_user_id");
+      QSession session;
+      try
+      {
+         session = config.getOwnerSessionSupplier() == null
+            ? new QAuthenticationModuleDispatcher().getQModule(QContext.getQInstance().getAuthentication()).createAutomatedSessionForUser(QContext.getQInstance(), owner)
+            : config.getOwnerSessionSupplier().apply(owner);
+      }
+      catch(Exception e)
+      {
+         return (WatchVerdict.skipped("Owner session unavailable"));
+      }
+      if(owner == null || session == null || session.getUser() == null || !Objects.equals(owner, session.getUser().getIdReference())
+         || session.getSecurityKeyValues() == null || !session.getSecurityKeyValues().getOrDefault(DataAnalysisTables.USER_KEY, List.of()).contains(owner))
+      {
+         return (WatchVerdict.skipped("Owner session unavailable"));
+      }
+      WatchVerdict[] verdict = new WatchVerdict[1];
+      try
+      {
+         QContext.withTemporaryContext(new CapturedContext(QContext.getQInstance(), session), () ->
+         {
+            QRecord owned = WatchRecords.get(DataAnalysisTables.WATCH, row.getValue("id"));
+            verdict[0] = service.watchPreview(owned, now).verdict();
+         });
+         return (verdict[0]);
+      }
+      catch(QPermissionDeniedException e)
+      {
+         return (WatchVerdict.skipped("Owner lacks READ or Analyze access to the watched table."));
+      }
+      catch(com.kingsrook.qqq.backend.core.exceptions.QBadRequestException e)
+      {
+         return (WatchVerdict.skipped("Watch evaluation unavailable: " + e.getMessage()));
+      }
+      catch(Exception e)
+      {
+         return (WatchVerdict.skipped("Watch evaluation unavailable; check the saved analysis and source configuration."));
+      }
+   }
+
+   /*******************************************************************************
+    ** Persist first; only successful evaluated transitions notify the alerts SPI.
+    *******************************************************************************/
+   private void store(QRecord row, WatchVerdict verdict, Instant now) throws QException
+   {
+      String key = "data_analysis_watch:" + row.getValue("id");
+      boolean wasFiring = "FIRING".equals(row.getValueString("state"));
+      QRecord patch = new QRecord().withValue("id", row.getValue("id")).withValue("last_evaluated_at", now)
+         .withValue("last_value", verdict.value()).withValue("last_message", verdict.message());
+      if(verdict.evaluated())
+      {
+         patch.withValue("state", verdict.firing() ? "FIRING" : "OK").withValue("alert_key", key);
+      }
+      update(patch);
+      if(!verdict.evaluated() || wasFiring == verdict.firing())
+      {
+         return;
+      }
+      if(verdict.firing())
+      {
+         AlertSignal signal = new AlertSignal(ALERT_TYPE).withDedupKey(key).withSeverity(AlertSeverity.valueOf(row.getValueString("severity")))
+            .withTitle("Data anomaly").withSummary(verdict.message()).withRecordRef(AlertRecordRef.of(DataAnalysisTables.WATCH, row.getValue("id")));
+         if(verdict.drill() != null)
+         {
+            signal.withLink(AlertLink.toTable(verdict.drill().getTableName(), verdict.drill().getFilter(), "View matching records"));
+         }
+         QAlerts.raise(signal);
+      }
+      else
+      {
+         QAlerts.resolve(ALERT_TYPE, key, verdict.message());
+      }
+   }
+
+   /*******************************************************************************
+    ** Enable and disable selected owner-locked rows through standard core updates.
+    *******************************************************************************/
+   public void setEnabled(Serializable id, Boolean enabled) throws QException
+   {
+      requireEdit();
+      WatchRecords.get(DataAnalysisTables.WATCH, id);
+      update(new QRecord().withValue("id", id).withValue("enabled", enabled));
+   }
+
+   /*******************************************************************************
+    ** Never count an unsuccessful core update as an evaluated watch.
+    *******************************************************************************/
+   private static void update(QRecord record) throws QException
+   {
+      UpdateInput input = new UpdateInput(DataAnalysisTables.WATCH).withRecords(List.of(record));
+      input.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+      QRecord stored = new UpdateAction().execute(input).getRecords().getFirst();
+      if(stored.getErrors() != null && !stored.getErrors().isEmpty())
+      {
+         throw new QException("Watch could not be updated: " + stored.getErrors().getFirst().getMessage());
+      }
+   }
+
+   /*******************************************************************************
+    ** Public mutation entry points enforce table permission before source queries.
+    *******************************************************************************/
+   private static void requireEdit() throws QPermissionDeniedException
+   {
+      if(!PermissionsHelper.hasTablePermission(new UpdateInput(DataAnalysisTables.WATCH), DataAnalysisTables.WATCH, TablePermissionSubType.EDIT))
+      {
+         throw new QPermissionDeniedException("Watch editing is unavailable.");
+      }
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchPreview.java

@@ -0,0 +1,31 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.util.List;
+import com.kingsrook.qbits.dataanalysis.result.Chart;
+
+/*******************************************************************************
+ ** Read-only preview with a semantic time series and detector explanation.
+ *******************************************************************************/
+public record WatchPreview(List<WatchBucket> series, Chart chart, WatchVerdict verdict)
+{
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchRecords.java

@@ -0,0 +1,141 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.io.Serializable;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.dataanalysis.engine.DrillFilters;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionEnforcement;
+import com.kingsrook.qqq.backend.core.actions.tables.QueryAction;
+import com.kingsrook.qqq.backend.core.exceptions.QBadRequestException;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.exceptions.QNotFoundException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QueryInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.utils.JsonUtils;
+
+/*******************************************************************************
+ ** Watch and saved-definition reads always use the active user's core permissions.
+ *******************************************************************************/
+public final class WatchRecords
+{
+   /*******************************************************************************
+    ** Utility class.
+    *******************************************************************************/
+   private WatchRecords()
+   {
+   }
+
+   /*******************************************************************************
+    ** Fetch heavy JSON with record locks; missing and inaccessible ids look alike.
+    *******************************************************************************/
+   public static QRecord get(String table, Serializable id) throws QException
+   {
+      QueryInput input = new QueryInput(table).withShouldFetchHeavyFields(true)
+         .withFilter(DrillFilters.condition("id", QCriteriaOperator.EQUALS, id).withLimit(1));
+      input.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+      List<QRecord> records = new QueryAction().execute(input).getRecords();
+      if(records.isEmpty())
+      {
+         throw new QNotFoundException("Watch or saved analysis is unavailable.");
+      }
+      return (records.getFirst());
+   }
+
+   /*******************************************************************************
+    ** Saved pivots provide one measure; explicit watch metrics can also be used.
+    *******************************************************************************/
+   public static QRecord resolve(QRecord row) throws QException
+   {
+      boolean saved = row.getValue("saved_id") != null;
+      WatchDefinition.require(saved != (row.getValueString("table_name") != null && !row.getValueString("table_name").isBlank()), "Choose either saved_id or table_name.");
+      QRecord result = new QRecord(row);
+      if(saved)
+      {
+         WatchDefinition.require(row.getValueString("filter_json") == null || row.getValueString("filter_json").isBlank(), "Saved watches use the saved analysis filter; omit filter_json.");
+         QRecord analysis = get(DataAnalysisTables.SAVED, row.getValue("saved_id"));
+         result.withValue("table_name", analysis.getValue("table_name")).withValue("filter_json", analysis.getValue("filter_json"));
+         Map<String, Object> definition = WatchDefinition.object(analysis.getValueString("definition_json"), "Saved definition");
+         if(result.getValueString("metric_json") == null || result.getValueString("metric_json").isBlank())
+         {
+            Object metric = definition.get("metric");
+            if(metric == null && definition.get("values") instanceof List<?> values && values.size() == 1)
+            {
+               metric = values.getFirst();
+            }
+            WatchDefinition.require(metric != null, "Saved analysis needs one COUNT, SUM or AVG measure, or an explicit watch metric_json.");
+            Map<String, Object> selected = new LinkedHashMap<>(WatchDefinition.object(JsonUtils.toJson(metric), "Saved metric"));
+            if(selected.get("percentile") == null)
+            {
+               selected.remove("percentile");
+            }
+            result.withValue("metric_json", JsonUtils.toJson(selected));
+         }
+         if(result.getValueString("filter_json") == null && definition.get("filter") != null)
+         {
+            result.withValue("filter_json", JsonUtils.toJson(definition.get("filter")));
+         }
+      }
+      filter(result);
+      return (result);
+   }
+
+   /*******************************************************************************
+    ** Reject invalid filter JSON rather than silently broadening the watch.
+    *******************************************************************************/
+   public static QQueryFilter filter(QRecord row) throws QBadRequestException
+   {
+      String json = row.getValueString("filter_json");
+      if(json == null || json.isBlank())
+      {
+         return (new QQueryFilter());
+      }
+      WatchDefinition.object(json, "filter_json");
+      try
+      {
+         QQueryFilter filter = JsonUtils.toObject(json, QQueryFilter.class);
+         validateFilter(filter);
+         return (filter);
+      }
+      catch(Exception e)
+      {
+         throw new QBadRequestException("filter_json must contain a valid core query filter.");
+      }
+   }
+
+   /*******************************************************************************
+    ** Invalid operators and null criteria are never interpreted as empty filters.
+    *******************************************************************************/
+   private static void validateFilter(QQueryFilter filter)
+   {
+      if(filter == null || filter.getCriteria() == null || filter.getSubFilters() == null
+         || filter.getCriteria().stream().anyMatch(c -> c == null || c.getOperator() == null || c.getFieldName() == null))
+      {
+         throw new IllegalArgumentException("Invalid filter.");
+      }
+      filter.getSubFilters().forEach(WatchRecords::validateFilter);
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchSeries.java

@@ -0,0 +1,278 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import java.io.Serializable;
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.ZonedDateTime;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.TemporalAdjusters;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import com.kingsrook.qbits.dataanalysis.engine.AnalysisSecurity;
+import com.kingsrook.qbits.dataanalysis.engine.DrillFilters;
+import com.kingsrook.qbits.dataanalysis.engine.EffectiveConfig;
+import com.kingsrook.qbits.dataanalysis.engine.RelationshipSupport;
+import com.kingsrook.qbits.dataanalysis.result.Axis;
+import com.kingsrook.qbits.dataanalysis.result.Chart;
+import com.kingsrook.qbits.dataanalysis.result.ChartKind;
+import com.kingsrook.qbits.dataanalysis.result.Point;
+import com.kingsrook.qbits.dataanalysis.result.Series;
+import com.kingsrook.qbits.dataanalysis.result.ValueType;
+import com.kingsrook.qqq.backend.core.exceptions.QException;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.Aggregate;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.AggregateResult;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.DateBucket;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.DateBucketUnit;
+import com.kingsrook.qqq.backend.core.model.actions.tables.aggregate.GroupBy;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.utils.ValueUtils;
+
+/*******************************************************************************
+ ** Exact core date-bucket aggregates under the active user's FULL-field policy.
+ *******************************************************************************/
+public class WatchSeries extends RelationshipSupport
+{
+   /*******************************************************************************
+    ** Bind request-local policy and core streamed limits.
+    *******************************************************************************/
+   public WatchSeries(AnalysisSecurity security, EffectiveConfig config, String table)
+   {
+      super(security, config, table);
+   }
+
+   /*******************************************************************************
+    ** Evaluate only complete buckets; reject partial or privacy-suppressed data.
+    *******************************************************************************/
+   public WatchPreview preview(WatchDefinition definition, QQueryFilter filter, Instant now) throws QException
+   {
+      var date = full(definition.dateField());
+      WatchDefinition.require(date.getType() == QFieldType.DATE || date.getType() == QFieldType.DATE_TIME, "Choose a DATE or DATE_TIME field.");
+      WatchDefinition.require(!"HOUR".equals(definition.bucket()) || date.getType() == QFieldType.DATE_TIME, "HOUR requires a DATE_TIME field.");
+      if(definition.field() != null)
+      {
+         full(definition.field());
+      }
+      if(List.of("SUM", "AVG").contains(definition.metric()))
+      {
+         requireNumeric(definition.field());
+      }
+      validateFilter(filter);
+      ZonedDateTime end = boundary(now, definition.bucket());
+      ZonedDateTime start = move(end, definition.bucket(), -definition.k() - 1);
+      QQueryFilter window = DrillFilters.drill(table, filter, range(definition.dateField(), date.getType(), start, end)).getFilter();
+      GroupBy dateGroup = new GroupBy(date).withBucket(new DateBucket(DateBucketUnit.valueOf(definition.bucket())));
+      Aggregate count = measure(primaryKey(table), AggregateOperator.COUNT);
+      Aggregate metric = switch(definition.metric())
+      {
+         case "SUM" -> measure(definition.field(), AggregateOperator.SUM);
+         case "AVG" -> measure(definition.field(), AggregateOperator.AVG);
+         case "NULL_SHARE" -> measure(definition.field(), AggregateOperator.COUNT_NULL);
+         default -> count;
+      };
+      Aggregate nonNull = List.of("SUM", "AVG").contains(definition.metric()) ? measure(definition.field(), AggregateOperator.COUNT) : count;
+      AggregateInput input = new AggregateInput(table).withFilter(window).withGroupBy(dateGroup).withAggregate(count);
+      if(metric != count)
+      {
+         input.withAggregate(metric);
+      }
+      if(nonNull != count)
+      {
+         input.withAggregate(nonNull);
+      }
+      GroupBy category = "CATEGORY_SET".equals(definition.metric()) ? new GroupBy(full(definition.field())) : null;
+      if(category != null)
+      {
+         input.withGroupBy(category);
+      }
+      Map<String, List<AggregateResult>> grouped = new LinkedHashMap<>();
+      for(AggregateResult row : aggregate(input))
+      {
+         grouped.computeIfAbsent(key(row.getGroupByValues().get(dateGroup), definition.bucket()), ignored -> new ArrayList<>()).add(row);
+      }
+      List<WatchBucket> buckets = new ArrayList<>();
+      boolean seen = false;
+      for(ZonedDateTime at = start; at.isBefore(end); at = move(at, definition.bucket(), 1))
+      {
+         Serializable value = "HOUR".equals(definition.bucket()) ? at.toInstant() : at.toLocalDate();
+         List<AggregateResult> rows = grouped.getOrDefault(key(value, definition.bucket()), List.of());
+         seen |= !rows.isEmpty();
+         if(!seen)
+         {
+            continue;
+         }
+         long size = rows.stream().mapToLong(row -> ((Number) row.getAggregateValues().get(count)).longValue()).sum();
+         if(size < config.getMinGroupSize())
+         {
+            return (skipped("Privacy guard: a bucket is below the minimum group size."));
+         }
+         Set<String> categories = new LinkedHashSet<>();
+         if(category != null)
+         {
+            for(AggregateResult row : rows)
+            {
+               if(((Number) row.getAggregateValues().get(count)).longValue() < config.getMinGroupSize())
+               {
+                  return (skipped("Privacy guard: a category is below the minimum group size."));
+               }
+               Serializable member = row.getGroupByValues().get(category);
+               if(member != null)
+               {
+                  categories.add(member.toString());
+               }
+            }
+         }
+         if(nonNull != count && !rows.isEmpty() && ((Number) rows.getFirst().getAggregateValues().get(nonNull)).longValue() < config.getMinGroupSize())
+         {
+            return (skipped("Privacy guard: too few non-null values."));
+         }
+         Object raw = rows.isEmpty() ? null : rows.getFirst().getAggregateValues().get(metric);
+         if("NULL_SHARE".equals(definition.metric()) && raw != null)
+         {
+            long nulls = ((Number) raw).longValue();
+            if(nulls > 0 && nulls < config.getMinGroupSize() || size - nulls > 0 && size - nulls < config.getMinGroupSize())
+            {
+               return (skipped("Privacy guard: too few null or non-null values."));
+            }
+         }
+         Double number = switch(definition.metric())
+         {
+            case "COUNT" -> (double) size;
+            case "CATEGORY_SET" -> (double) categories.size();
+            case "NULL_SHARE" -> size == 0 ? null : ((Number) raw).doubleValue() / size;
+            case "SUM" -> rows.isEmpty() ? 0.0 : raw == null ? null : ((Number) raw).doubleValue();
+            default -> raw == null ? null : ((Number) raw).doubleValue();
+         };
+         buckets.add(new WatchBucket(value, number, categories, DrillFilters.drill(table, filter,
+            range(definition.dateField(), date.getType(), at, move(at, definition.bucket(), 1)))));
+      }
+      String label = switch(definition.metric())
+      {
+         case "COUNT" -> security.table(table).getLabel();
+         case "NULL_SHARE" -> "Null share of " + full(definition.field()).getLabel();
+         case "CATEGORY_SET" -> full(definition.field()).getLabel();
+         default -> ("SUM".equals(definition.metric()) ? "Sum of " : "Average ") + full(definition.field()).getLabel();
+      };
+      label += " per " + definition.bucket().toLowerCase(java.util.Locale.ROOT);
+      WatchVerdict verdict = WatchDetectors.evaluate(definition, buckets, label);
+      List<Point> points = buckets.stream().map(bucket -> new Point().withX(bucket.start())
+         .withY(bucket.value() == null ? null : BigDecimal.valueOf(bucket.value())).withDrill(bucket.drill())).toList();
+      List<String> annotations = new ArrayList<>();
+      if(verdict.baselineLow() != null)
+      {
+         annotations.add("Baseline band: " + verdict.baselineLow() + " to " + verdict.baselineHigh() + ".");
+      }
+      annotations.add(verdict.message());
+      Chart chart = new Chart().withKind(ChartKind.TIME_SERIES)
+         .withX(new Axis().withField(definition.dateField()).withLabel(date.getLabel()).withValueType(date.getType() == QFieldType.DATE ? ValueType.DATE : ValueType.DATE_TIME))
+         .withY(new Axis().withLabel(label).withValueType("NULL_SHARE".equals(definition.metric()) ? ValueType.PERCENT : ValueType.DECIMAL))
+         .withSeries(List.of(new Series().withName(label).withPoints(points))).withAnnotations(annotations);
+      return (new WatchPreview(buckets, chart, verdict));
+   }
+
+   /*******************************************************************************
+    ** No suppressed values, categories or drill predicates leave this computation.
+    *******************************************************************************/
+   public static WatchPreview skipped(String message)
+   {
+      return (new WatchPreview(List.of(), new Chart().withKind(ChartKind.TIME_SERIES).withAnnotations(List.of(message)), WatchVerdict.skipped(message)));
+   }
+
+   /*******************************************************************************
+    ** User-authored watch filters must also use FULL fields only.
+    *******************************************************************************/
+   private void validateFilter(QQueryFilter filter) throws QException
+   {
+      if(filter == null)
+      {
+         return;
+      }
+      for(var criterion : filter.getCriteria())
+      {
+         if(criterion.getFieldName() != null)
+         {
+            full(criterion.getFieldName());
+         }
+         if(criterion.getFieldFunction() != null)
+         {
+            full(criterion.getFieldFunction().getFieldName());
+         }
+      }
+      for(QQueryFilter child : filter.getSubFilters())
+      {
+         validateFilter(child);
+      }
+   }
+
+   /*******************************************************************************
+    ** Calendar boundaries follow the owner's time zone, including DST changes.
+    *******************************************************************************/
+   private static ZonedDateTime boundary(Instant now, String bucket)
+   {
+      ZonedDateTime local = now.atZone(ValueUtils.getSessionOrInstanceZoneId());
+      return (switch(bucket)
+      {
+         case "HOUR" -> local.truncatedTo(ChronoUnit.HOURS);
+         case "WEEK" -> local.toLocalDate().with(TemporalAdjusters.previousOrSame(java.time.DayOfWeek.MONDAY)).atStartOfDay(local.getZone());
+         case "MONTH" -> local.toLocalDate().withDayOfMonth(1).atStartOfDay(local.getZone());
+         default -> local.toLocalDate().atStartOfDay(local.getZone());
+      });
+   }
+
+   /*******************************************************************************
+    ** Advance on the calendar, rather than treating every day as 24 hours.
+    *******************************************************************************/
+   private static ZonedDateTime move(ZonedDateTime at, String bucket, int count)
+   {
+      return (switch(bucket)
+      {
+         case "HOUR" -> at.plusHours(count);
+         case "WEEK" -> at.plusWeeks(count);
+         case "MONTH" -> at.plusMonths(count);
+         default -> at.plusDays(count);
+      });
+   }
+
+   /*******************************************************************************
+    ** Keep DATE filters as dates and DATE_TIME filters as instants.
+    *******************************************************************************/
+   private static QQueryFilter range(String field, QFieldType type, ZonedDateTime start, ZonedDateTime end)
+   {
+      return (DrillFilters.range(field, type == QFieldType.DATE ? start.toLocalDate() : start.toInstant(),
+         type == QFieldType.DATE ? end.toLocalDate() : end.toInstant(), false));
+   }
+
+   /*******************************************************************************
+    ** Normalize core date-bucket keys independently of JDBC value representation.
+    *******************************************************************************/
+   private static String key(Serializable value, String bucket)
+   {
+      return ("HOUR".equals(bucket) ? ValueUtils.getValueAsInstant(value).toString() : ValueUtils.getValueAsLocalDate(value).toString());
+   }
+}

src/main/java/com/kingsrook/qbits/dataanalysis/watch/WatchVerdict.java

@@ -0,0 +1,38 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis.watch;
+
+import com.kingsrook.qbits.dataanalysis.result.Drill;
+
+/*******************************************************************************
+ ** An unevaluated verdict must never resolve an existing alert.
+ *******************************************************************************/
+public record WatchVerdict(Boolean evaluated, Boolean firing, Double value, Double baseline,
+   Double baselineLow, Double baselineHigh, String message, Drill drill)
+{
+   /*******************************************************************************
+    ** Fail closed without retaining data or a drill.
+    *******************************************************************************/
+   public static WatchVerdict skipped(String message)
+   {
+      return (new WatchVerdict(false, false, null, null, null, null, message, null));
+   }
+}

src/main/resources/openapi/qqq-v1/data-analysis.yaml

@@ -1503,6 +1503,60 @@ components:
             \ null when suppressed."
           nullable: true
       type: "object"
+    DataAnalysisWatchBucketV1:
+      properties:
+        start:
+          description: "Calendar date or hour-start instant in the user's time zone."
+          type: "string"
+        value:
+          nullable: true
+          type: "number"
+        categories:
+          items:
+            type: "string"
+          type: "array"
+        drill:
+          $ref: "#/components/schemas/DataAnalysisDrillV1"
+          nullable: true
+      type: "object"
+    DataAnalysisWatchPreviewV1:
+      properties:
+        series:
+          items:
+            $ref: "#/components/schemas/DataAnalysisWatchBucketV1"
+          type: "array"
+        chart:
+          $ref: "#/components/schemas/DataAnalysisChartV1"
+          description: "TIME_SERIES with baseline band annotations."
+        verdict:
+          $ref: "#/components/schemas/DataAnalysisWatchVerdictV1"
+      type: "object"
+    DataAnalysisWatchVerdictV1:
+      properties:
+        evaluated:
+          description: "False for insufficient history or privacy suppression; must\
+            \ not resolve an alert."
+          type: "boolean"
+        firing:
+          type: "boolean"
+        value:
+          nullable: true
+          type: "number"
+        baseline:
+          nullable: true
+          type: "number"
+        baselineLow:
+          nullable: true
+          type: "number"
+        baselineHigh:
+          nullable: true
+          type: "number"
+        message:
+          type: "string"
+        drill:
+          $ref: "#/components/schemas/DataAnalysisDrillV1"
+          nullable: true
+      type: "object"
     FilterCriteria:
       properties:
         fieldName:
@@ -1586,9 +1640,9 @@ info:
   contact:
     email: "contact@kingsrook.com"
   description: "Profiles, pivots, comparisons, distribution shifts, correlations,\
-    \ scatter samples and crosstabs as the signed-in user."
+    \ scatter samples, crosstabs and anomaly watch previews as the signed-in user."
   title: "QQQ Middleware API extension: data-analysis"
-  version: "1.2.0"
+  version: "1.3.0"
 openapi: "3.0.3"
 paths:
   /qqq/v1/qbit/data-analysis/tables/{table}/capabilities:
@@ -1818,6 +1872,29 @@ paths:
       - sessionUuidCookie:
         - "N/A"
       summary: "Cross-tabulate two fields"
+  /qqq/v1/qbit/data-analysis/watches/{id}/preview:
+    post:
+      description: "Compute complete time buckets and a detector verdict under the\
+        \ signed-in user's READ, Analyze and FULL-field permissions. Includes a baseline\
+        \ band annotation. Does not update the watch or raise alerts."
+      parameters:
+      - description: "Owner-scoped analysis run or watch id."
+        in: "path"
+        name: "id"
+        required: true
+        schema:
+          type: "string"
+      responses:
+        200:
+          content:
+            application/json:
+              schema:
+                $ref: "#/components/schemas/DataAnalysisWatchPreviewV1"
+          description: "Successful Response"
+      security:
+      - sessionUuidCookie:
+        - "N/A"
+      summary: "Preview an anomaly watch"
   /qqq/v1/qbit/data-analysis/tables/{table}/pivot:
     post:
       description: "Group records by up to 3 row and 2 column fields (with date or\
@@ -1931,7 +2008,7 @@ paths:
         \ result of an owner-scoped run. Completed results are rechecked against current\
         \ permissions."
       parameters:
-      - description: "Owner-scoped analysis run id."
+      - description: "Owner-scoped analysis run or watch id."
         in: "path"
         name: "runId"
         required: true

src/test/java/com/kingsrook/qbits/dataanalysis/DataAnalysisApiTest.java

@@ -73,6 +73,10 @@ class DataAnalysisApiTest extends EngineFixture
             session.withPermission("orders.analyze");
          }
          session.withSecurityKeyValue("data_analysis_user", session.getUser().getIdReference());
+         if(session.getUuid().startsWith("watch-"))
+         {
+            session.getSecurityKeyValues().remove("data_analysis.admin");
+         }
       }
    }
 
@@ -169,7 +173,7 @@ class DataAnalysisApiTest extends EngineFixture
          assertThat(denied.getStatus()).isEqualTo(403);
          assertThat(run.path(operation.equals("COLUMN_PROFILE") ? "columnProfile" : "tableProfile").isObject()).isTrue();
       }
-      assertThat(invoker.listOperations().stream().filter(o -> o.pathTemplate().startsWith(DataAnalysisApi.PREFIX)).toList()).hasSize(13);
+      assertThat(invoker.listOperations().stream().filter(o -> o.pathTemplate().startsWith(DataAnalysisApi.PREFIX)).toList()).hasSize(14);
       var client = java.net.http.HttpClient.newHttpClient();
       var response = client.send(java.net.http.HttpRequest.newBuilder(java.net.URI.create("http://127.0.0.1:" + server.getBoundPort() + "/qqq/v1/openapi.json")).build(), java.net.http.HttpResponse.BodyHandlers.ofString());
       assertThat(response.statusCode()).isEqualTo(200);
@@ -269,7 +273,7 @@ class DataAnalysisApiTest extends EngineFixture
    }
 
    /*******************************************************************************
-    ** Generated fragment is committed verbatim and has all thirteen operations.
+    ** Generated fragment is committed verbatim and has all fourteen operations.
     *******************************************************************************/
    @Test
    void contract() throws Exception
@@ -351,7 +355,7 @@ class DataAnalysisApiTest extends EngineFixture
       }
       relationshipCountDrills(call(HttpMethod.POST, "/tables/{table}/crosstab", Map.of("rowField", "status", "columnField", "customer"), 200));
       // All new schemas are discoverable from the same served OpenAPI catalog.
-      assertThat(com.kingsrook.qbits.dataanalysis.api.DataAnalysisApi.extension().getSemanticVersion()).isEqualTo("1.2.0");
+      assertThat(com.kingsrook.qbits.dataanalysis.api.DataAnalysisApi.extension().getSemanticVersion()).isEqualTo("1.3.0");
    }
 
    /*******************************************************************************
@@ -415,4 +419,37 @@ class DataAnalysisApiTest extends EngineFixture
          }
       }
    }
+
+   @Test
+   void watchPreviewOverHttpIsReadOnlyAndOwnerLocked() throws Exception
+   {
+      caller = "watch-owner";
+      String owner = "User:" + Math.floorMod(caller.hashCode(), 10_000);
+      java.time.ZoneId zone = java.time.ZoneId.of("America/Chicago");
+      java.time.LocalDate today = java.time.LocalDate.now(zone);
+      for(int i = 0; i < 5; i++)
+      {
+         TestSupport.insert("orders", new com.kingsrook.qqq.backend.core.model.data.QRecord().withValue("id", 100 + i).withValue("customer", 1)
+            .withValue("created", today.minusDays(i < 2 ? 3 - i : 1).atTime(12, 0).atZone(zone).toInstant()));
+      }
+      var row = TestSupport.insert(com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables.WATCH,
+         WatchDetectorsTest.definition("SPIKE", "COUNT", "{\"k\":2}").withValue("owner_user_id", owner));
+      start();
+      var request = new UiApiRequest().withMethod(HttpMethod.POST).withPath(DataAnalysisApi.PREFIX + "/watches/{id}/preview")
+         .withPathParam("id", row.getValueString("id")).withSessionUUID(caller).withUserTimezone("America/Chicago");
+      var response = invoker.invoke(request);
+      assertThat(response.getStatus()).as(response.getBody()).isEqualTo(200);
+      JsonNode json = response.getJson();
+      assertThat(json.path("series").size()).isEqualTo(3);
+      assertThat(json.path("verdict").path("firing").asBoolean()).isTrue();
+      assertThat(json.path("verdict").path("value").asDouble()).isEqualTo(3);
+      assertThat(json.path("chart").path("kind").asText()).isEqualTo("TIME_SERIES");
+      assertThat(json.path("chart").path("annotations").toString()).contains("Baseline band");
+      checkDrills(json);
+      assertThat(com.kingsrook.qbits.dataanalysis.watch.WatchRecords.get(com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables.WATCH, row.getValue("id")).getValue("last_evaluated_at")).isNull();
+      assertThat(invoker.invoke(request.withSessionUUID("watch-outsider")).getStatus()).isEqualTo(404);
+      assertThat(invoker.invoke(request.withSessionUUID("denied")).getStatus()).isEqualTo(403);
+      assertThat(invoker.invoke(request.withSessionUUID(caller).withPathParam("id", "99999")).getStatus()).isEqualTo(404);
+      assertThat(invoker.invoke(request.withPathParam("id", "invalid")).getStatus()).isEqualTo(400);
+   }
 }

src/test/java/com/kingsrook/qbits/dataanalysis/DataAnalysisHomeTest.java

@@ -90,9 +90,9 @@ class DataAnalysisHomeTest
       TestSupport.insert(DataAnalysisTables.SAVED_STAR, new QRecord().withValue("saved_id", mine.getValue("id")).withValue("user_id", "alice"));
       TestSupport.insert(DataAnalysisTables.SAVED_STAR, new QRecord().withValue("saved_id", bob.getValue("id")).withValue("user_id", "alice"));
       TestSupport.insert(DataAnalysisTables.SAVED_STAR, new QRecord().withValue("saved_id", mine.getValue("id")).withValue("user_id", "bob"));
-      TestSupport.insert(DataAnalysisTables.WATCH, new QRecord().withValue("detector", "SPIKE").withValue("enabled", true).withValue("state", "FIRING"));
-      TestSupport.insert(DataAnalysisTables.WATCH, new QRecord().withValue("detector", "DROP").withValue("enabled", true));
-      TestSupport.insert(DataAnalysisTables.WATCH, new QRecord().withValue("detector", "DROP").withValue("enabled", false));
+      TestSupport.insert(DataAnalysisTables.WATCH, WatchDetectorsTest.definition("SPIKE", "COUNT", null).withValue("enabled", true).withValue("state", "FIRING"));
+      TestSupport.insert(DataAnalysisTables.WATCH, WatchDetectorsTest.definition("DROP", "COUNT", null).withValue("enabled", true));
+      TestSupport.insert(DataAnalysisTables.WATCH, WatchDetectorsTest.definition("DROP", "COUNT", null).withValue("enabled", false));
       run("alice", "COMPLETE", now);
       run("alice", "COMPLETE", now.minus(7, ChronoUnit.DAYS));
       run("alice", "FAILED", now.minus(1, ChronoUnit.DAYS));

src/test/java/com/kingsrook/qbits/dataanalysis/DataAnalysisMetadataTest.java

@@ -67,7 +67,7 @@ class DataAnalysisMetadataTest
       QInstance instance = TestSupport.instance();
       new QInstanceValidator().validate(instance);
       assertThat(instance.getTables()).containsOnlyKeys(DataAnalysisTables.ALL);
-      assertThat(instance.getPossibleValueSources().keySet().stream().filter(name -> name.startsWith("data_analysis_"))).hasSize(13);
+      assertThat(instance.getPossibleValueSources().keySet().stream().filter(name -> name.startsWith("data_analysis_"))).hasSize(16);
       assertThat(instance.getJoins()).hasSize(3);
       assertThat(instance.getWidgets()).hasSize(4).containsKey(DataAnalysisHome.HOME_WIDGET);
       assertThat(instance.getApp("dataAnalysis").getSections().getFirst().getProcesses()).isNullOrEmpty();
@@ -204,7 +204,7 @@ class DataAnalysisMetadataTest
       }
       assertThat(ddl).contains("id INTEGER GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY", "create_date TIMESTAMP WITHOUT TIME ZONE",
          "modify_date TIMESTAMP WITHOUT TIME ZONE", "rows_scanned BIGINT", "result_json TEXT", "shared BOOLEAN", "day INTEGER",
-         "owner_user_id VARCHAR(250) NOT NULL", "CREATE UNIQUE INDEX IF NOT EXISTS uq_data_analysis_table_config_1 ON data_analysis_table_config (table_name)",
+         "owner_user_id VARCHAR(250) NOT NULL", "date_field VARCHAR(250) NOT NULL", "last_value NUMERIC", "last_message TEXT", "CREATE UNIQUE INDEX IF NOT EXISTS uq_data_analysis_table_config_1 ON data_analysis_table_config (table_name)",
          "CREATE UNIQUE INDEX IF NOT EXISTS uq_data_analysis_saved_star_1 ON data_analysis_saved_star (saved_id, user_id)");
       assertThat(ddl).endsWith(";\n");
    }

src/test/java/com/kingsrook/qbits/dataanalysis/DataAnalysisPossibleValuesTest.java

@@ -0,0 +1,167 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+
+package com.kingsrook.qbits.dataanalysis;
+
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Map;
+import com.kingsrook.qbits.dataanalysis.engine.EffectiveConfigResolver;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisFieldPossibleValueProvider;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qqq.backend.core.actions.values.SearchPossibleValueSourceAction;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.actions.values.SearchPossibleValueSourceInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.PossibleValueSearchFilterUseCase;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValue;
+import com.kingsrook.qqq.backend.core.model.metadata.possiblevalues.QPossibleValueSourceType;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Dependent possible values through the same search inputs used by form requests.
+ *******************************************************************************/
+class DataAnalysisPossibleValuesTest
+{
+   /*******************************************************************************
+    ** Release the request context.
+    *******************************************************************************/
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+
+
+
+   /*******************************************************************************
+    ** The watch table selection filters saved analyses without widening owner locks.
+    *******************************************************************************/
+   @Test
+   void watchSavedAnalysisDependsOnTable() throws Exception
+   {
+      QInstance instance = TestSupport.instance();
+      QRecord orders = TestSupport.insert(DataAnalysisTables.SAVED, TestSupport.saved("alice", false));
+      QRecord customers = TestSupport.insert(DataAnalysisTables.SAVED, TestSupport.saved("alice", false).withValue("table_name", "customers"));
+      TestSupport.insert(DataAnalysisTables.SAVED, TestSupport.saved("bob", true));
+      QContext.init(instance, TestSupport.user("alice").withPermissions("data_analysis.saved.read", "data_analysis.watch.edit"));
+      QFieldMetaData field = instance.getTable(DataAnalysisTables.WATCH).getField("saved_id");
+      assertThat(search(field, Map.of("table_name", "orders"))).extracting(value -> (Object) value.getId()).containsExactly(orders.getValueInteger("id"));
+      assertThat(search(field, Map.of("table_name", "customers"))).extracting(value -> (Object) value.getId()).containsExactly(customers.getValueInteger("id"));
+      assertThat(search(field, Map.of())).isEmpty();
+      assertThat(search(field, Map.of("table_name", "missing"))).isEmpty();
+   }
+
+
+
+   /*******************************************************************************
+    ** All four table references use the core source, including host-owned sources.
+    *******************************************************************************/
+   @Test
+   void tableNamesUseCoreTablesSource() throws Exception
+   {
+      QInstance instance = TestSupport.instance();
+      instance.addTable(new QTableMetaData().withName("orders").withLabel("Orders"));
+      for(String tableName : List.of(DataAnalysisTables.TABLE_CONFIG, DataAnalysisTables.SAVED, DataAnalysisTables.WATCH, DataAnalysisTables.RUN))
+      {
+         QFieldMetaData field = instance.getTable(tableName).getField("table_name");
+         assertThat(field.getPossibleValueSourceName()).isEqualTo("tables");
+         assertThat(search(field, Map.of())).extracting(value -> value.getId().toString()).containsExactlyInAnyOrderElementsOf(instance.getTables().keySet());
+      }
+      var hostSource = instance.getPossibleValueSource("tables").withLabel("Host tables");
+      QInstance host = new QInstance();
+      host.addBackend(instance.getBackend("memory"));
+      host.addPossibleValueSource(hostSource);
+      new DataAnalysisQBitProducer().withQBitConfig(new DataAnalysisQBitConfig().withBackendName("memory")).produce(host);
+      assertThat(host.getPossibleValueSource("tables")).isSameAs(hostSource);
+      assertThat(host.getPossibleValueSource("tables").getLabel()).isEqualTo("Host tables");
+   }
+
+
+
+   /*******************************************************************************
+    ** Both CSV settings offer only selected-table fields and retain CSV parsing.
+    *******************************************************************************/
+   @Test
+   void fieldPickersDependOnTableAndKeepCsvStorage() throws Exception
+   {
+      QInstance instance = TestSupport.instance();
+      instance.addTable(new QTableMetaData().withName("orders").withLabel("Orders")
+         .withField(new QFieldMetaData("amount", QFieldType.DECIMAL).withLabel("Order total"))
+         .withField(new QFieldMetaData("status", QFieldType.STRING)));
+      instance.addTable(new QTableMetaData().withName("customers").withLabel("Customers")
+         .withField(new QFieldMetaData("email", QFieldType.STRING).withLabel("Email")));
+      for(String fieldName : List.of("analyzable_fields", "excluded_fields"))
+      {
+         QFieldMetaData field = instance.getTable(DataAnalysisTables.TABLE_CONFIG).getField(fieldName);
+         assertThat(field.getType()).isEqualTo(QFieldType.TEXT);
+         assertThat(instance.getPossibleValueSource(field.getPossibleValueSourceName()).getType()).isEqualTo(QPossibleValueSourceType.CUSTOM);
+         assertThat(search(field, Map.of("table_name", "orders"))).extracting(value -> (Object) value.getId()).containsExactly("amount", "status");
+         assertThat(search(field, Map.of("table_name", "orders"))).extracting(QPossibleValue::getLabel).containsExactly("Order total (amount)", "status (status)");
+         assertThat(search(field, Map.of("table_name", "customers"))).extracting(value -> (Object) value.getId()).containsExactly("email");
+         assertThat(search(field, Map.of())).isEmpty();
+         assertThat(search(field, Map.of("table_name", ""))).isEmpty();
+         assertThat(search(field, Map.of("table_name", "missing"))).isEmpty();
+         assertThat(search(field, null)).isEmpty();
+      }
+      SearchPossibleValueSourceInput input = new SearchPossibleValueSourceInput().withPossibleValueSourceName(DataAnalysisFieldPossibleValueProvider.NAME)
+         .withOtherValues(Map.of("table_name", "orders")).withSearchTerm("TOTAL");
+      assertThat(new SearchPossibleValueSourceAction().execute(input).getResults()).extracting(value -> (Object) value.getId()).containsExactly("amount");
+      input.withSearchTerm("AMOUNT");
+      assertThat(new SearchPossibleValueSourceAction().execute(input).getResults()).extracting(value -> (Object) value.getId()).containsExactly("amount");
+      input.withIdList(List.of("status", "email"));
+      assertThat(new SearchPossibleValueSourceAction().execute(input).getResults()).extracting(value -> (Object) value.getId()).containsExactly("status");
+      DataAnalysisFieldPossibleValueProvider provider = new DataAnalysisFieldPossibleValueProvider();
+      assertThat(provider.getPossibleValue(null)).isNull();
+      assertThat(provider.getPossibleValue("amount,status").getId()).isEqualTo("amount,status");
+      TestSupport.insert(DataAnalysisTables.TABLE_CONFIG, new QRecord().withValue("table_name", "orders")
+         .withValue("analyzable_fields", "amount, status").withValue("excluded_fields", "status, amount"));
+      var config = new EffectiveConfigResolver(new DataAnalysisQBitConfig().withBackendName("memory")).resolve("orders");
+      assertThat(config.getAnalyzableFields()).containsExactlyInAnyOrder("amount", "status");
+      assertThat(config.getExcludedFields()).containsExactlyInAnyOrder("amount", "status");
+   }
+
+
+
+   /*******************************************************************************
+    ** Mirror TablePossibleValuesSpecV1: clone and interpret the field filter as FORM.
+    *******************************************************************************/
+   private List<QPossibleValue<?>> search(QFieldMetaData field, Map<String, Serializable> otherValues) throws Exception
+   {
+      SearchPossibleValueSourceInput input = new SearchPossibleValueSourceInput()
+         .withPossibleValueSourceName(field.getPossibleValueSourceName()).withOtherValues(otherValues);
+      if(field.getPossibleValueSourceFilter() != null)
+      {
+         var filter = field.getPossibleValueSourceFilter().clone();
+         filter.interpretValues(Map.of("input", otherValues), PossibleValueSearchFilterUseCase.FORM);
+         input.withDefaultQueryFilter(filter);
+      }
+      return (new SearchPossibleValueSourceAction().execute(input).getResults());
+   }
+}

src/test/java/com/kingsrook/qbits/dataanalysis/PivotContractTest.java

@@ -347,4 +347,22 @@ abstract class PivotContractTest extends EngineFixture
       equalNumber(result.getGrandTotals().getFirst(), "0");
       countDrills(result);
    }
+
+   @Test
+   void booleanHeadersAndHumanMeasureLabels() throws Exception
+   {
+      var booleans = service.pivot(pivot(List.of("paid"), List.of("status")));
+      assertThat(booleans.getRowHeaders().stream().flatMap(List::stream).map(PivotHeader::getLabel)).containsExactlyInAnyOrder("Yes", "No", "(Blank)");
+      assertThat(booleans.getMeasures().getFirst().getLabel()).isEqualTo("Count");
+      var operators = List.of(PivotMeasure.Operator.AVG, PivotMeasure.Operator.SUM, PivotMeasure.Operator.COUNT_DISTINCT,
+         PivotMeasure.Operator.MIN, PivotMeasure.Operator.MAX, PivotMeasure.Operator.PERCENTILE);
+      var measures = operators.stream().map(operator -> measure(operator, "amount")).toList();
+      measures.getLast().withPercentile(new BigDecimal("0.5"));
+      var result = service.pivot(pivot(List.of("paid"), List.of()).withValues(measures));
+      assertThat(result.getMeasures().stream().map(m -> m.getLabel())).containsExactly("Average Amount", "Sum of Amount", "Distinct Amount", "Minimum Amount", "Maximum Amount", "Median Amount");
+      measures.getLast().withPercentile(new BigDecimal("0.9"));
+      assertThat(service.pivot(pivot(List.of("paid"), List.of()).withValues(measures)).getMeasures().getLast().getLabel()).isEqualTo("90th percentile of Amount");
+      var cross = service.crosstab(new com.kingsrook.qbits.dataanalysis.result.CrosstabRequest().withTableName("orders").withRowField("paid").withColumnField("status"));
+      assertThat(cross.getRows().stream().map(PivotHeader::getLabel)).containsExactlyInAnyOrder("Yes", "No", "(Blank)");
+   }
 }

src/test/java/com/kingsrook/qbits/dataanalysis/WatchDetectorsTest.java

@@ -0,0 +1,174 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Set;
+import com.kingsrook.qbits.dataanalysis.result.Drill;
+import com.kingsrook.qbits.dataanalysis.watch.WatchBucket;
+import com.kingsrook.qbits.dataanalysis.watch.WatchDefinition;
+import com.kingsrook.qbits.dataanalysis.watch.WatchDetectors;
+import com.kingsrook.qqq.backend.core.exceptions.QBadRequestException;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/*******************************************************************************
+ ** Hand-computed detector cases isolate statistical behavior from the database.
+ *******************************************************************************/
+class WatchDetectorsTest
+{
+   static QRecord definition(String detector, String metric, String parameters)
+   {
+      return (new QRecord().withValue("table_name", "orders").withValue("owner_user_id", "alice").withValue("detector", detector)
+         .withValue("metric_json", "{\"operator\":\"" + metric + "\",\"field\":\"amount\"}").withValue("parameters_json", parameters)
+         .withValue("date_field", "created").withValue("bucket", "DAY").withValue("severity", "WARNING").withValue("enabled", true));
+   }
+
+   private WatchDefinition parsed(String detector, String metric, String parameters) throws Exception
+   {
+      return (WatchDefinition.parse(definition(detector, metric, parameters)));
+   }
+
+   private List<WatchBucket> series(Double... values)
+   {
+      List<WatchBucket> result = new ArrayList<>();
+      for(Double value : values)
+      {
+         result.add(new WatchBucket("day-" + result.size(), value, Set.of(), new Drill().withTableName("orders")));
+      }
+      return (result);
+   }
+
+   @Test
+   void robustZScoreHasDirectionAndStrictThreshold() throws Exception
+   {
+      var spike = parsed("SPIKE", "COUNT", "{\"k\":3}");
+      assertThat(WatchDetectors.evaluate(spike, series(9.0, 10.0, 11.0, 20.0), "Orders per day").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(spike, series(9.0, 10.0, 11.0, 1.0), "Orders").firing()).isFalse();
+      var drop = parsed("DROP", "COUNT", "{\"k\":3}");
+      var verdict = WatchDetectors.evaluate(drop, series(40.0, 45.0, 55.0, 12.0), "Orders per day");
+      assertThat(verdict.firing()).isTrue();
+      assertThat(verdict.baseline()).isEqualTo(45.0);
+      assertThat(verdict.message()).contains("Orders per day dropped to 12", "usual 40");
+      assertThat(verdict.drill().getTableName()).isEqualTo("orders");
+      assertThat(WatchDetectors.evaluate(drop, series(40.0, 45.0, 55.0, 100.0), "Orders").firing()).isFalse();
+      var exact = parsed("SPIKE", "COUNT", "{\"k\":3,\"threshold\":0.6745}");
+      assertThat(WatchDetectors.evaluate(exact, series(9.0, 10.0, 11.0, 11.0), "Orders").firing()).isFalse();
+   }
+
+   @Test
+   void zeroMadUsesRelativeChangeAndHandlesZeroAndNegativeBaselines() throws Exception
+   {
+      var spike = parsed("SPIKE", "COUNT", "{\"k\":2}");
+      for(double x : new double[] {10, 15})
+      {
+         assertThat(WatchDetectors.evaluate(spike, series(10.0, 10.0, x), "Orders").firing()).isFalse();
+      }
+      assertThat(WatchDetectors.evaluate(spike, series(10.0, 10.0, 16.0), "Orders").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(spike, series(0.0, 0.0, 0.0), "Orders").firing()).isFalse();
+      assertThat(WatchDetectors.evaluate(spike, series(0.0, 0.0, 1.0), "Orders").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(spike, series(-10.0, -10.0, -4.0), "Balance").firing()).isTrue();
+      var drop = parsed("DROP", "COUNT", "{\"k\":2}");
+      assertThat(WatchDetectors.evaluate(drop, series(10.0, 10.0, 4.0), "Orders").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(drop, series(0.0, 0.0, -1.0), "Balance").firing()).isTrue();
+   }
+
+   @Test
+   void rangeIncludesBothBoundsAndNeedsOnlyLatestValue() throws Exception
+   {
+      var range = parsed("OUT_OF_RANGE", "SUM", "{\"min\":10,\"max\":20}");
+      assertThat(WatchDetectors.evaluate(range, series(10.0), "Amount").firing()).isFalse();
+      assertThat(WatchDetectors.evaluate(range, series(20.0), "Amount").firing()).isFalse();
+      assertThat(WatchDetectors.evaluate(range, series(9.0), "Amount").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(range, series(21.0), "Amount").firing()).isTrue();
+   }
+
+   @Test
+   void categoriesCompareWithEntireTrailingWindow() throws Exception
+   {
+      var definition = parsed("NEW_CATEGORY", "CATEGORY_SET", "{\"k\":2}");
+      var baseline = List.of(new WatchBucket("1", 1.0, Set.of("a"), null), new WatchBucket("2", 1.0, Set.of("b"), null));
+      List<WatchBucket> values = new ArrayList<>(baseline);
+      values.add(new WatchBucket("3", 2.0, Set.of("a", "b"), null));
+      assertThat(WatchDetectors.evaluate(definition, values, "Status").firing()).isFalse();
+      values.set(2, new WatchBucket("3", 2.0, Set.of("a", "new"), null));
+      assertThat(WatchDetectors.evaluate(definition, values, "Status").message()).contains("1 new category");
+      values.set(2, new WatchBucket("3", 0.0, Set.of(), null));
+      assertThat(WatchDetectors.evaluate(definition, values, "Status").firing()).isFalse();
+   }
+
+   @Test
+   void nullRateDetectsDeltaOrConsecutiveRises() throws Exception
+   {
+      var definition = parsed("NULL_RATE_CREEP", "NULL_SHARE", "{\"k\":3}");
+      assertThat(WatchDetectors.evaluate(definition, series(0.1, 0.1, 0.1, 0.16), "Null share").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(definition, series(0.1, 0.11, 0.12, 0.13), "Null share").firing()).isTrue();
+      assertThat(WatchDetectors.evaluate(definition, series(0.1, 0.12, 0.11, 0.13), "Null share").firing()).isFalse();
+      assertThat(WatchDetectors.evaluate(definition, series(0.1, 0.11, 0.11, 0.12), "Null share").firing()).isFalse();
+   }
+
+   @ParameterizedTest
+   @ValueSource(strings = {"SPIKE", "DROP", "NEW_CATEGORY", "NULL_RATE_CREEP"})
+   void insufficientHistoryNeverBecomesRecovery(String detector) throws Exception
+   {
+      String metric = detector.equals("NEW_CATEGORY") ? "CATEGORY_SET" : detector.equals("NULL_RATE_CREEP") ? "NULL_SHARE" : "COUNT";
+      var result = WatchDetectors.evaluate(parsed(detector, metric, null), series(1.0), "Orders");
+      assertThat(result.evaluated()).isFalse();
+      assertThat(result.message()).contains("Not enough history");
+      assertThat(result.value()).isNull();
+      assertThat(result.drill()).isNull();
+   }
+
+   @Test
+   void emptyAndMissingNumericBucketsAreNotZero() throws Exception
+   {
+      var range = parsed("OUT_OF_RANGE", "AVG", "{\"min\":0,\"max\":1}");
+      assertThat(WatchDetectors.evaluate(range, List.of(), "Orders").evaluated()).isFalse();
+      assertThat(WatchDetectors.evaluate(range, series((Double) null), "Orders").evaluated()).isFalse();
+      assertThat(WatchDetectors.evaluate(range, series(Double.NaN), "Orders").evaluated()).isFalse();
+   }
+
+   @ParameterizedTest
+   @ValueSource(strings = {"[]", "{", "null", "{\"k\":0}", "{\"k\":1001}", "{\"k\":1.5}", "{\"k\":\"14\"}", "{\"threshold\":0}", "{\"threshold\":null}", "{\"relativeChange\":-1}", "{\"delta\":0.1}"})
+   void rejectsInvalidSpikeParameters(String json)
+   {
+      assertThatThrownBy(() -> parsed("SPIKE", "COUNT", json)).isInstanceOf(QBadRequestException.class);
+   }
+
+   @Test
+   void validatesMetricDetectorAndRangeCompatibility() throws Exception
+   {
+      assertThatThrownBy(() -> parsed("SPIKE", "CATEGORY_SET", null)).hasMessageContaining("NEW_CATEGORY");
+      assertThatThrownBy(() -> parsed("NEW_CATEGORY", "COUNT", null)).hasMessageContaining("CATEGORY_SET");
+      assertThatThrownBy(() -> parsed("NULL_RATE_CREEP", "SUM", null)).hasMessageContaining("NULL_SHARE");
+      assertThatThrownBy(() -> parsed("OUT_OF_RANGE", "COUNT", "{\"min\":2,\"max\":1}")).hasMessageContaining("min <= max");
+      assertThatThrownBy(() -> parsed("OUT_OF_RANGE", "COUNT", null)).hasMessageContaining("requires min and max");
+      assertThatThrownBy(() -> parsed("NULL_RATE_CREEP", "NULL_SHARE", "{\"delta\":2}")).hasMessageContaining("delta");
+      assertThatThrownBy(() -> parsed("NULL_RATE_CREEP", "NULL_SHARE", "{\"n\":0}")).hasMessageContaining("n must");
+      assertThat(parsed("SPIKE", "COUNT", null).k()).isEqualTo(14);
+      assertThat(parsed("SPIKE", "COUNT", null).threshold()).isEqualTo(3.5);
+   }
+}

src/test/java/com/kingsrook/qbits/dataanalysis/WatchesTest.java

@@ -0,0 +1,460 @@
+/*
+ * QQQ - Low-code Application Framework for Engineers.
+ * Copyright (C) 2021-2025.  Kingsrook, LLC
+ * 651 N Broad St Ste 205 # 6917 | Middletown DE 19709 | United States
+ * contact@kingsrook.com
+ * https://github.com/Kingsrook/
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.kingsrook.qbits.dataanalysis;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.ZoneOffset;
+import java.util.ArrayList;
+import java.util.List;
+import com.kingsrook.qbits.dataanalysis.app.DataAnalysisHome;
+import com.kingsrook.qbits.dataanalysis.engine.DrillFilters;
+import com.kingsrook.qbits.dataanalysis.model.DataAnalysisTables;
+import com.kingsrook.qbits.dataanalysis.processes.WatchActions;
+import com.kingsrook.qbits.dataanalysis.watch.WatchEvaluator;
+import com.kingsrook.qbits.dataanalysis.watch.WatchRecords;
+import com.kingsrook.qqq.backend.core.actions.permissions.PermissionEnforcement;
+import com.kingsrook.qqq.backend.core.actions.tables.InsertAction;
+import com.kingsrook.qqq.backend.core.actions.tables.UpdateAction;
+import com.kingsrook.qqq.backend.core.alerts.AlertResolution;
+import com.kingsrook.qqq.backend.core.alerts.AlertSignal;
+import com.kingsrook.qqq.backend.core.alerts.AlertSinkInterface;
+import com.kingsrook.qqq.backend.core.alerts.AlertsInstanceMetaData;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.instances.QInstanceEnricher;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepInput;
+import com.kingsrook.qqq.backend.core.model.actions.processes.RunBackendStepOutput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.insert.InsertInput;
+import com.kingsrook.qqq.backend.core.model.actions.tables.query.QCriteriaOperator;
+import com.kingsrook.qqq.backend.core.model.actions.tables.update.UpdateInput;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.fields.QFieldType;
+import com.kingsrook.qqq.backend.core.model.metadata.security.RecordSecurityLock;
+import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/*******************************************************************************
+ ** Real core memory rows, owner sessions, alerts SPI and record-action summaries.
+ *******************************************************************************/
+class WatchesTest
+{
+   static final LocalDate TODAY = LocalDate.now(ZoneOffset.UTC);
+   static final Instant NOW = TODAY.atTime(12, 0).toInstant(ZoneOffset.UTC);
+   private QInstance instance;
+   private DataAnalysisQBitConfig config;
+   private WatchEvaluator evaluator;
+   private final List<Integer> latestIds = new ArrayList<>();
+
+   public static class Sink implements AlertSinkInterface
+   {
+      static final List<AlertSignal> raised = new ArrayList<>();
+      static final List<AlertResolution> resolved = new ArrayList<>();
+
+      @Override
+      public void raise(AlertSignal signal)
+      {
+         raised.add(signal);
+      }
+
+      @Override
+      public void resolve(AlertResolution resolution)
+      {
+         resolved.add(resolution);
+      }
+
+      @Override
+      public void heartbeat(String key, Instant at)
+      {
+      }
+   }
+
+   static QSession owner(String user)
+   {
+      return (TestSupport.user(user).withPermissions("orders.read", "orders.analyze", "data_analysis.watch.read", "data_analysis.watch.edit", "data_analysis.saved.read")
+         .withValue(QSession.VALUE_KEY_USER_TIMEZONE, "UTC"));
+   }
+
+   @BeforeEach
+   void setup() throws Exception
+   {
+      instance = TestSupport.instance();
+      instance.addTable(new QTableMetaData().withName("orders").withLabel("Orders").withBackendName("memory").withPrimaryKeyField("id")
+         .withPermissionRules(DataAnalysisTables.permissionRules("orders"))
+         .withField(new QFieldMetaData("id", QFieldType.INTEGER)).withField(new QFieldMetaData("amount", QFieldType.DECIMAL))
+         .withField(new QFieldMetaData("created", QFieldType.DATE_TIME)).withField(new QFieldMetaData("day", QFieldType.DATE))
+         .withField(new QFieldMetaData("category", QFieldType.STRING)).withField(new QFieldMetaData("owner", QFieldType.STRING))
+         .withRecordSecurityLock(new RecordSecurityLock().withFieldName("owner").withSecurityKeyType(DataAnalysisTables.USER_KEY)));
+      new QInstanceEnricher(instance).enrich();
+      QContext.init(instance, TestSupport.admin().withPermissions("orders.read", "orders.insert", "orders.edit", "orders.analyze")
+         .withValue(QSession.VALUE_KEY_USER_TIMEZONE, "UTC"));
+      int[] counts = {10, 11, 9, 60};
+      for(int day = 0; day < 4; day++)
+      {
+         for(int i = 0; i < counts[day]; i++)
+         {
+            QRecord row = order(TODAY.minusDays(4 - day), "alice", i % 2 == 0 ? null : BigDecimal.TEN, day == 3 ? "new" : "old");
+            if(day == 3)
+            {
+               latestIds.add(row.getValueInteger("id"));
+            }
+         }
+      }
+      // Another owner and the incomplete current bucket must never affect Alice.
+      order(TODAY.minusDays(1), "bob", new BigDecimal("1000000"), "private");
+      order(TODAY, "alice", new BigDecimal("1000000"), "incomplete");
+      config = new DataAnalysisQBitConfig().withBackendName("memory").withOwnerSessionSupplier(WatchesTest::owner);
+      new DataAnalysisRuntime(config).addSelfToInstance(instance);
+      evaluator = DataAnalysisRuntime.watches();
+      Sink.raised.clear();
+      Sink.resolved.clear();
+      AlertsInstanceMetaData.ofOrWithNew(instance).withSink(new QCodeReference(Sink.class));
+   }
+
+   private QRecord order(LocalDate day, String owner, BigDecimal amount, String category) throws Exception
+   {
+      return (TestSupport.insert("orders", new QRecord().withValue("created", day.atTime(12, 0).toInstant(ZoneOffset.UTC))
+         .withValue("day", day).withValue("owner", owner).withValue("amount", amount).withValue("category", category)));
+   }
+
+   private QRecord watch(String detector, String metric, String parameters) throws Exception
+   {
+      return (TestSupport.insert(DataAnalysisTables.WATCH, WatchDetectorsTest.definition(detector, metric, parameters)));
+   }
+
+   private QRecord stored(QRecord watch) throws Exception
+   {
+      return (WatchRecords.get(DataAnalysisTables.WATCH, watch.getValue("id")));
+   }
+
+   @AfterEach
+   void cleanup()
+   {
+      QContext.clear();
+   }
+
+   @Test
+   void raisesAndResolvesOnceUsingOwnerRows() throws Exception
+   {
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      QSession caller = QContext.getQSession();
+      var verdict = evaluator.evaluate(watch.getValue("id"), NOW);
+      assertThat(verdict.firing()).isTrue();
+      assertThat(verdict.value()).isEqualTo(60.0);
+      assertThat(QContext.getQSession()).isSameAs(caller);
+      assertThat(stored(watch).getValueString("state")).isEqualTo("FIRING");
+      assertThat(stored(watch).getValueInstant("last_evaluated_at")).isEqualTo(NOW);
+      assertThat(Sink.raised).hasSize(1);
+      AlertSignal signal = Sink.raised.getFirst();
+      assertThat(signal.getAlertTypeName()).isEqualTo(WatchEvaluator.ALERT_TYPE);
+      assertThat(signal.getDedupKey()).isEqualTo("data_analysis_watch:" + watch.getValue("id"));
+      assertThat(signal.getRecordRefs().getFirst().getTableName()).isEqualTo(DataAnalysisTables.WATCH);
+      assertThat(signal.getLinks().getFirst().getTableName()).isEqualTo("orders");
+      assertThat(signal.getLinks().getFirst().getUrl()).isNull();
+      assertThat(signal.getLinks().getFirst().getFilter()).isNotNull();
+      evaluator.evaluate(watch.getValue("id"), NOW);
+      assertThat(Sink.raised).hasSize(1);
+      for(int id : latestIds.subList(10, latestIds.size()))
+      {
+         UpdateInput update = new UpdateInput("orders").withRecords(List.of(new QRecord().withValue("id", id)
+            .withValue("created", TODAY.plusDays(1).atTime(12, 0).toInstant(ZoneOffset.UTC))));
+         update.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+         assertThat(new UpdateAction().execute(update).getRecords().getFirst().getErrors()).isNullOrEmpty();
+      }
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).firing()).isFalse();
+      assertThat(stored(watch).getValueString("state")).isEqualTo("OK");
+      assertThat(Sink.resolved).hasSize(1);
+      assertThat(Sink.resolved.getFirst().getDedupKey()).isEqualTo(signal.getDedupKey());
+      evaluator.evaluate(watch.getValue("id"), NOW);
+      assertThat(Sink.resolved).hasSize(1);
+   }
+
+   @Test
+   void missingOrMismatchedOwnerSessionsNeverReadAsCaller() throws Exception
+   {
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      config.withOwnerSessionSupplier(id -> null);
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).message()).isEqualTo("Owner session unavailable");
+      assertThat(stored(watch).getValue("last_value")).isNull();
+      config.withOwnerSessionSupplier(id -> owner("bob"));
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).message()).isEqualTo("Owner session unavailable");
+      config.withOwnerSessionSupplier(null);
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).message()).isEqualTo("Owner session unavailable");
+      assertThat(Sink.raised).isEmpty();
+   }
+
+   @Test
+   void revokedReadOrAnalyzeSkipsWithoutResolving() throws Exception
+   {
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      evaluator.evaluate(watch.getValue("id"), NOW);
+      for(String revoked : List.of("orders.read", "orders.analyze"))
+      {
+         config.withOwnerSessionSupplier(id ->
+         {
+            QSession session = owner(id);
+            session.getPermissions().remove(revoked);
+            return (session);
+         });
+         var verdict = evaluator.evaluate(watch.getValue("id"), NOW);
+         assertThat(verdict.evaluated()).isFalse();
+         assertThat(verdict.message()).contains("Owner lacks READ or Analyze");
+         assertThat(verdict.value()).isNull();
+         assertThat(verdict.drill()).isNull();
+         assertThat(stored(watch).getValueString("state")).isEqualTo("FIRING");
+         assertThat(stored(watch).getValue("last_value")).isNull();
+      }
+      assertThat(Sink.resolved).isEmpty();
+      assertThat(Sink.raised).hasSize(1);
+   }
+
+   @Test
+   void previewCoversAllMetricsAndPrivacy() throws Exception
+   {
+      QContext.setQSession(owner("alice"));
+      for(String metric : List.of("COUNT", "SUM", "AVG", "NULL_SHARE", "CATEGORY_SET"))
+      {
+         String detector = metric.equals("CATEGORY_SET") ? "NEW_CATEGORY" : metric.equals("NULL_SHARE") ? "NULL_RATE_CREEP" : "SPIKE";
+         QRecord definition = WatchDetectorsTest.definition(detector, metric, "{\"k\":3}");
+         if(metric.equals("CATEGORY_SET"))
+         {
+            definition.withValue("metric_json", "{\"operator\":\"CATEGORY_SET\",\"field\":\"category\"}");
+         }
+         var preview = new DataAnalysisService(config).watchPreview(definition, NOW);
+         assertThat(preview.series()).hasSize(4);
+         assertThat(preview.chart().getKind().name()).isEqualTo("TIME_SERIES");
+         assertThat(preview.chart().getAnnotations()).isNotEmpty();
+         assertThat(preview.verdict().evaluated()).isTrue();
+         if(metric.equals("CATEGORY_SET"))
+         {
+            assertThat(preview.verdict().firing()).isTrue();
+            assertThat(preview.series().getLast().categories()).containsExactly("new");
+         }
+         if(metric.equals("AVG"))
+         {
+            assertThat(preview.verdict().value()).isEqualTo(10.0);
+         }
+         if(metric.equals("NULL_SHARE"))
+         {
+            assertThat(preview.verdict().value()).isEqualTo(0.5);
+         }
+      }
+      var guarded = new DataAnalysisService(new DataAnalysisQBitConfig().withBackendName("memory").withDefaultMinGroupSize(12))
+         .watchPreview(WatchDetectorsTest.definition("SPIKE", "COUNT", "{\"k\":3}"), NOW);
+      assertThat(guarded.series()).isEmpty();
+      assertThat(guarded.verdict().message()).contains("Privacy guard");
+      assertThat(guarded.verdict().drill()).isNull();
+   }
+
+   @Test
+   void fullFieldPolicyAndRowLimitAreEnforced() throws Exception
+   {
+      QContext.setQSession(owner("alice"));
+      var restricted = new DataAnalysisQBitConfig().withBackendName("memory").withExcludeTables(java.util.Set.of("orders"));
+      assertThatThrownBy(() -> new DataAnalysisService(restricted).watchPreview(WatchDetectorsTest.definition("SPIKE", "COUNT", "{\"k\":3}"), NOW))
+         .hasMessageContaining("unavailable");
+      config.withFieldAnalysisAccessResolver(() -> (table, field) -> field.getName().equals("amount")
+         ? com.kingsrook.qbits.dataanalysis.engine.FieldAnalysisAccess.NONE : com.kingsrook.qbits.dataanalysis.engine.FieldAnalysisAccess.FULL);
+      QRecord definition = WatchDetectorsTest.definition("SPIKE", "SUM", "{\"k\":3}");
+      assertThatThrownBy(() -> new DataAnalysisService(config).watchPreview(definition, NOW)).hasMessageContaining("Unknown analysis field");
+      definition.withValue("metric_json", "{\"operator\":\"COUNT\"}").withValue("filter_json", "{\"criteria\":[{\"fieldName\":\"amount\",\"operator\":\"IS_BLANK\"}]}");
+      assertThatThrownBy(() -> new DataAnalysisService(config).watchPreview(definition, NOW)).hasMessageContaining("Unknown analysis field");
+   }
+
+   @Test
+   void actionsHaveReviewAndResultLinesAndEnableDisable() throws Exception
+   {
+      QRecord spike = watch("SPIKE", "COUNT", "{\"k\":3}");
+      QRecord ok = watch("OUT_OF_RANGE", "COUNT", "{\"min\":0,\"max\":100}");
+      QRecord secondOk = watch("DROP", "COUNT", "{\"k\":3}");
+      RunBackendStepInput input = new RunBackendStepInput().withProcessName(WatchActions.EVALUATE).withRecords(List.of(spike, ok, secondOk));
+      RunBackendStepOutput output = new RunBackendStepOutput();
+      WatchActions.Review review = new WatchActions.Review();
+      review.runOnePage(input, output);
+      assertThat(review.getProcessSummary(output, false)).hasSize(3).allSatisfy(line -> assertThat(line.toString()).contains("will be evaluated"));
+      assertThat(Sink.raised).isEmpty();
+      WatchActions.Result result = new WatchActions.Result();
+      result.runOnePage(input, new RunBackendStepOutput());
+      assertThat(result.getProcessSummary(output, true).getFirst().toString()).contains("3 watches evaluated: 1 firing, 2 OK");
+      for(String action : List.of(WatchActions.DISABLE, WatchActions.ENABLE))
+      {
+         WatchActions.Result load = new WatchActions.Result();
+         load.runOnePage(new RunBackendStepInput().withProcessName(action).withRecords(List.of(spike)), new RunBackendStepOutput());
+         assertThat(stored(spike).getValueBoolean("enabled")).isEqualTo(action.equals(WatchActions.ENABLE));
+         assertThat(load.getProcessSummary(output, true).getFirst().toString()).contains(action.equals(WatchActions.ENABLE) ? "1 watch enabled" : "1 watch disabled");
+      }
+   }
+
+   @Test
+   void scheduleEvaluatesEnabledAndHomeLinksFiringFirst() throws Exception
+   {
+      QRecord spike = watch("SPIKE", "COUNT", "{\"k\":3}");
+      QRecord disabled = watch("DROP", "COUNT", "{\"k\":3}");
+      evaluator.setEnabled(disabled.getValue("id"), false);
+      assertThat(evaluator.evaluateEnabled(NOW)).isEqualTo(1);
+      assertThat(stored(disabled).getValue("last_evaluated_at")).isNull();
+      String html = DataAnalysisHome.render(NOW);
+      assertThat(html.indexOf("Firing &middot; Spike")).isLessThan(html.indexOf("Ok &middot; Drop"));
+      assertThat(html).contains("/" + spike.getValue("id") + "\"");
+      assertThat(instance.getProcess(WatchActions.SCHEDULED).getIsHidden()).isTrue();
+      assertThat(instance.getProcess(WatchActions.EVALUATE).getTableName()).isEqualTo(DataAnalysisTables.WATCH);
+   }
+
+   @Test
+   void standardCrudValidatesPartialUpdatesAndOwnerLocks() throws Exception
+   {
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      UpdateInput input = new UpdateInput(DataAnalysisTables.WATCH).withRecords(List.of(new QRecord().withValue("id", watch.getValue("id")).withValue("parameters_json", "{\"k\":0}")));
+      input.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+      assertThat(new UpdateAction().execute(input).getRecords().getFirst().getErrors()).isNotEmpty();
+      assertThat(stored(watch).getValueString("parameters_json")).contains("3");
+      InsertInput invalid = new InsertInput(DataAnalysisTables.WATCH).withRecord(WatchDetectorsTest.definition("OUT_OF_RANGE", "COUNT", "{}"));
+      invalid.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+      assertThat(new InsertAction().execute(invalid).getRecords().getFirst().getErrors()).isNotEmpty();
+      QContext.setQSession(owner("bob"));
+      assertThatThrownBy(() -> evaluator.preview(watch.getValue("id"), NOW)).hasMessageContaining("unavailable");
+      assertThatThrownBy(() -> evaluator.evaluate(watch.getValue("id"), NOW)).hasMessageContaining("unavailable");
+      QContext.setQSession(owner("alice"));
+      QContext.getQSession().getPermissions().remove("data_analysis.watch.edit");
+      assertThatThrownBy(() -> evaluator.evaluate(watch.getValue("id"), NOW)).hasMessageContaining("editing is unavailable");
+      assertThat(Sink.raised).isEmpty();
+   }
+
+   @Test
+   void savedAnalysisResolvesAsOwnerAndNoAlertsInstallationStillStoresState() throws Exception
+   {
+      QRecord saved = TestSupport.insert(DataAnalysisTables.SAVED, TestSupport.saved("alice", false)
+         .withValue("definition_json", "{\"values\":[{\"operator\":\"COUNT\"}]}"));
+      QRecord definition = WatchDetectorsTest.definition("SPIKE", "COUNT", "{\"k\":3}")
+         .withValue("saved_id", saved.getValue("id")).withValue("table_name", null).withValue("metric_json", null);
+      QRecord watch = TestSupport.insert(DataAnalysisTables.WATCH, definition);
+      AlertsInstanceMetaData.of(instance).withSink(null);
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).firing()).isTrue();
+      assertThat(stored(watch).getValueString("state")).isEqualTo("FIRING");
+      assertThat(Sink.raised).isEmpty();
+   }
+
+   public static class AutomatedSessions implements com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleCustomizerInterface
+   {
+      @Override
+      public void customizeAutomatedSessionForUser(QInstance instance, QSession session, java.io.Serializable userId)
+      {
+         QSession grants = owner(userId.toString());
+         session.setPermissions(grants.getPermissions());
+         session.setSecurityKeyValues(grants.getSecurityKeyValues());
+         session.withValue(QSession.VALUE_KEY_USER_TIMEZONE, "UTC");
+      }
+   }
+
+   @Test
+   void hostAuthenticationModuleSuppliesTheOwnerSession() throws Exception
+   {
+      config.withOwnerSessionSupplier(null);
+      instance.withInstanceDefaultAuthentication(new com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData()
+         .withName("owner-auth").withType(com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType.MOCK).withCustomizer(new QCodeReference(AutomatedSessions.class)));
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      assertThat(evaluator.evaluate(watch.getValue("id"), NOW).value()).isEqualTo(60.0);
+      assertThat(Sink.raised).hasSize(1);
+   }
+
+   @Test
+   void allCalendarUnitsExcludeCurrentBucketAndHaveExactDrills() throws Exception
+   {
+      for(String unit : List.of("HOUR", "DAY", "WEEK", "MONTH"))
+      {
+         java.time.ZonedDateTime end = NOW.atZone(ZoneOffset.UTC);
+         end = switch(unit)
+         {
+            case "HOUR" -> end.truncatedTo(java.time.temporal.ChronoUnit.HOURS);
+            case "WEEK" -> end.toLocalDate().with(java.time.temporal.TemporalAdjusters.previousOrSame(java.time.DayOfWeek.MONDAY)).atStartOfDay(ZoneOffset.UTC);
+            case "MONTH" -> end.toLocalDate().withDayOfMonth(1).atStartOfDay(ZoneOffset.UTC);
+            default -> end.toLocalDate().atStartOfDay(ZoneOffset.UTC);
+         };
+         java.time.temporal.ChronoUnit duration = switch(unit)
+         {
+            case "HOUR" -> java.time.temporal.ChronoUnit.HOURS;
+            case "WEEK" -> java.time.temporal.ChronoUnit.WEEKS;
+            case "MONTH" -> java.time.temporal.ChronoUnit.MONTHS;
+            default -> java.time.temporal.ChronoUnit.DAYS;
+         };
+         for(int ago : new int[] {2, 1, 1, 0})
+         {
+            TestSupport.insert("orders", new QRecord().withValue("owner", "alice").withValue("category", unit)
+               .withValue("created", end.minus(ago, duration).toInstant()).withValue("day", end.minus(ago, duration).toLocalDate()));
+         }
+         QRecord definition = WatchDetectorsTest.definition("SPIKE", "COUNT", "{\"k\":1}").withValue("bucket", unit)
+            .withValue("date_field", unit.equals("HOUR") ? "created" : "day")
+            .withValue("filter_json", com.kingsrook.qqq.backend.core.utils.JsonUtils.toJson(DrillFilters.condition("category", QCriteriaOperator.EQUALS, unit)));
+         QContext.setQSession(owner("alice"));
+         var preview = new DataAnalysisService(config).watchPreview(definition, NOW);
+         assertThat(preview.series()).as(unit).hasSize(2);
+         assertThat(preview.verdict().value()).as(unit).isEqualTo(2.0);
+         assertThat(preview.verdict().firing()).as(unit).isTrue();
+         var count = new com.kingsrook.qqq.backend.core.model.actions.tables.count.CountInput("orders").withFilter(preview.verdict().drill().getFilter());
+         count.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+         assertThat(new com.kingsrook.qqq.backend.core.actions.tables.CountAction().execute(count).getCount()).as(unit).isEqualTo(2);
+         QContext.setQSession(TestSupport.admin().withPermissions("orders.insert"));
+      }
+   }
+
+   @Test
+   void streamedProcessRunsSelectedRecordsThroughCore() throws Exception
+   {
+      QRecord watch = watch("SPIKE", "COUNT", "{\"k\":3}");
+      var input = new com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput().withProcessName(WatchActions.EVALUATE)
+         .withFrontendStepBehavior(com.kingsrook.qqq.backend.core.model.actions.processes.RunProcessInput.FrontendStepBehavior.SKIP)
+         .withCallback(new com.kingsrook.qqq.backend.core.actions.processes.QProcessCallback()
+         {
+            @Override
+            public com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter getQueryFilter()
+            {
+               return (DrillFilters.condition("id", QCriteriaOperator.EQUALS, watch.getValue("id")));
+            }
+         });
+      input.setPermissionEnforcement(PermissionEnforcement.ENFORCE);
+      var output = new com.kingsrook.qqq.backend.core.actions.processes.RunProcessAction().execute(input);
+      assertThat(output.getException()).isEmpty();
+      assertThat(output.getValue(com.kingsrook.qqq.backend.core.processes.implementations.etl.streamedwithfrontend.StreamedETLWithFrontendProcess.FIELD_PROCESS_SUMMARY).toString())
+         .contains("1 watch evaluated: 1 firing, 0 OK");
+      assertThat(stored(watch).getValueString("state")).isEqualTo("FIRING");
+   }
+
+   @Test
+   void configuredSchedulerUsesFifteenMinuteInterval() throws Exception
+   {
+      var configured = new QInstance();
+      configured.addBackend(new com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData().withName("memory")
+         .withBackendType(com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule.class));
+      new DataAnalysisQBitProducer().withQBitConfig(new DataAnalysisQBitConfig().withBackendName("memory").withSchedulerName("hostScheduler")).produce(configured);
+      var schedule = configured.getProcess(WatchActions.SCHEDULED).getSchedule();
+      assertThat(schedule.getSchedulerName()).isEqualTo("hostScheduler");
+      assertThat(schedule.getRepeatSeconds()).isEqualTo(900);
+      new WatchActions.Scheduled().run(new RunBackendStepInput(), new RunBackendStepOutput());
+   }
+}