diff --git a/.gencode_hash.txt b/.gencode_hash.txt
index f8c44acc75..9562f8a9f6 100644
--- a/.gencode_hash.txt
+++ b/.gencode_hash.txt
@@ -3,8 +3,8 @@
 fb876b8f7bfdccf156d751efddf2660a7a7ab5585e89be0f82c59f7a61e68d4c  gencode/docs/config.html
 22bfb3fd7c0e73dac4cbd830aa7983714917ec6da813f6bf5768ce3aa9786fb2  gencode/docs/config_mapping.html
 1ce30b7c554a0be09a9124707cb5602dce78753304cca17c06268ecb3337da51  gencode/docs/configuration_endpoint.html
-c2ad0293315ef1e8fd6a0326752b4ce4d7b0d54ae07cd1ebd07be351ac22f11f  gencode/docs/configuration_execution.html
-6a097f1f87ab7b9a82e4d3aa6f6cedf69499a95742863a7f53bb5fdf53fd6ff0  gencode/docs/configuration_pod.html
+136eeb98d4db5c2427498b0483e0fb8e73bc3dcb962f0ab942f7b98115b1b71f  gencode/docs/configuration_execution.html
+f4937abf377ef27e18e2dba77589fa08f2d955cf85a905f360ddaf4ad0bea0c3  gencode/docs/configuration_pod.html
 4d327975ae4d48a5dd920f298931fc5056a7239d677e61a9f4d129068316dd0a  gencode/docs/configuration_pubber.html
 f24d15e549f1b143b50d6011f2f7fd03286e1cd128a3e25591195b25f8efc472  gencode/docs/data_template.html
 45d39c61ce39afc0d14f3dc99a93947587d63999cd373cb44b63ce8f576c6e7e  gencode/docs/events.html
@@ -81,7 +81,7 @@ e256275d840d94c6e61316e756b0dad6d823db8900e35e785f6ef5001fdb5fc9  gencode/java/u
 a773f8271a99f96a3fb09ceb79bc4022a8b7ab941f5e6246e073703d33022b99  gencode/java/udmi/schema/GatewayConfig.java
 0f176bbd9748ac357ebcb7cf9ba66fca4f4768ce40a2a1436bb6d14093ee8101  gencode/java/udmi/schema/GatewayModel.java
 1f7f6809e143e9e3e13e6fa73776a893b62b0c05f8802087b1e276dc0ab83e19  gencode/java/udmi/schema/GatewayState.java
-d14fdea87e337eb8bbe44f6b1a435dd0bacaf71cf6384dfeaf28a0a121e07605  gencode/java/udmi/schema/IotAccess.java
+6ffcc0820c02c9f83787f3102389380b5d4da20c4d813219964151bdfebf335e  gencode/java/udmi/schema/IotAccess.java
 d0858bd7ba306176f4e3c19eeb69ef2467afec0832b6917b72d3ab87469e4974  gencode/java/udmi/schema/Jwt.java
 5a0cc53317592a868f7f23a67a7493bd571054bb25d9e28de51000ad6473dfeb  gencode/java/udmi/schema/Level.java
 6b4c29a682d0697d5420efb9e7788ec7358d94435da288938007205b8669e180  gencode/java/udmi/schema/LocalnetConfig.java
diff --git a/bin/pubber b/bin/pubber
index aa260bb5ba..1bda0b6854 100755
--- a/bin/pubber
+++ b/bin/pubber
@@ -13,7 +13,7 @@ function cleanup {
 }
 trap cleanup EXIT
 
-[[ $# == 0 ]] && usage SITE_{PATH/FILE} [PROJECT_ID] [DEVICE_ID] [SERIAL_NO] [options] ...
+[[ $# == 0 ]] && usage SITE_{PATH/FILE} [PROJECT_SPEC] [DEVICE_ID] [SERIAL_NO] [options] ...
 
 config_file=$(realpath $1)
 serial_no=NA
diff --git a/bin/test_combos b/bin/test_combos
index fc078d8359..013482cbac 100755
--- a/bin/test_combos
+++ b/bin/test_combos
@@ -4,8 +4,11 @@
 #
 
 tests=tests/combos/*.json
+echo Testing $tests
 
 validator/bin/build
+jarfile=$(realpath validator/build/libs/validator-1.0-SNAPSHOT-all.jar)
+javarun="java -cp $jarfile com.google.daq.mqtt.validator.SchemaTester"
 
 failures=
 for test in $tests; do
@@ -14,7 +17,10 @@ for test in $tests; do
     site_model=$(jq -r .site_model $test)
     validator_config=$(jq -r .validator $test)
     if [[ $validator_config != null ]]; then
-        validator/bin/validate -- schema trace $traces $site_model
+        cmd="$javarun -p -- -s $site_model -r $traces --"
+        echo $cmd
+        $cmd
+
         check=$(echo $validator_config | jq -r .check)
         echo $check
         code=0
diff --git a/bin/test_schema b/bin/test_schema
index b1282a95e4..cd88682d70 100755
--- a/bin/test_schema
+++ b/bin/test_schema
@@ -46,6 +46,7 @@ if [ "$build" == y ]; then
 fi
 
 jarfile=$(realpath validator/build/libs/validator-1.0-SNAPSHOT-all.jar)
+javarun="java -cp $jarfile com.google.daq.mqtt.validator.SchemaTester"
 
 if [[ -z $subsets ]]; then
     subsets=$(cd $testdir; ls -d *)
@@ -68,7 +69,7 @@ for subset in $subsets; do
         reldir=$(realpath --relative-to $schemadir $testdir)
         reltest=$reldir:$subset
     fi
-    (cd $schemadir; java -jar -Dnashorn.args=--no-deprecation-warning $jarfile -a $schemaname $upgrade -f $reltest) || true
+    (cd $schemadir; $javarun -a $schemaname $upgrade -f $reltest) || true
 
     for testfile in $testfiles; do
         outfile=${testfile%.json}.out
diff --git a/bin/test_trace b/bin/test_trace
index ae0a6ae3a0..a0ba0bc302 100755
--- a/bin/test_trace
+++ b/bin/test_trace
@@ -19,6 +19,9 @@ function redact() {
         -e 's/\\n.*\.java:[0-9]+\)/\\tREDACTED\\n/g'
 }
 
+jarfile=$(realpath validator/build/libs/validator-1.0-SNAPSHOT-all.jar)
+javarun="java -cp $jarfile com.google.daq.mqtt.validator.SchemaTester"
+validator/bin/build
 
 trace_in=tests/traces/${trace_name}
 trace_out=$trace_in/expected/
@@ -33,7 +36,9 @@ export UDMI_TOOLS=test_trace
 
 # Clean things out, including some potentially corrupt state.
 rm -rf $site_out $site_model/devices/XXX-1
-validator/bin/validate -- schema trace $trace_in $site_model
+cmd="$javarun -p -- -s $site_model -r $trace_in --"
+echo $cmd
+$cmd
 
 echo Redacting output files...
 for file in `find $site_out -type f`; do redact; done
diff --git a/bin/validator b/bin/validator
deleted file mode 100755
index 29bf78bd63..0000000000
--- a/bin/validator
+++ /dev/null
@@ -1,81 +0,0 @@
-#!/bin/bash -e
-
-UDMI_ROOT=$(realpath $(dirname $0)/..)
-source $UDMI_ROOT/etc/shell_common.sh
-
-OUT_DIR=$UDMI_ROOT/out
-mkdir -p $OUT_DIR
-echo $(date --utc +%Y%m%d-%H%M%S) $0 $@ >> $OUT_DIR/cmd_history.log
-
-[[ $# == 0 ]] && usage 'SITE_{PATH/FILE}' [PROJECT_ID] [SUBSCRIPTION_ID]
-
-site_path=$(realpath $1)
-shift
-
-project_id=
-if [[ -n ${1:-} ]]; then
-    project_id=$1
-    shift
-fi
-
-subscription=null
-if [[ -n ${1:-} && ! $1 =~ ^- ]]; then
-    subscription=\"$1\"
-    shift
-fi
-
-quoted_namespace=null
-quoted_bridge=null
-iot_provider=null
-if [[ $project_id =~ ^// ]]; then
-    no_prefix=${project_id#//}
-    iot_provider=\"${no_prefix%%/*}\"
-    no_provider=${no_prefix#*/}
-    project_id=${no_provider%/*}
-    udmi_namespace=${no_provider#$project_id}
-    [[ -n $udmi_namespace ]] && quoted_namespace=\"${udmi_namespace#/}\"
-elif [[ $project_id =~ @ ]]; then
-    iot_provider=\"jwt\"
-    quoted_bridge=\"${project_id#*@}\"
-    project_id=${project_id%@*}
-fi
-
-if [[ -n $project_id ]]; then
-    quoted_id=\"${project_id##*/}\"
-else
-    quoted_id=null
-fi
-
-if [[ -f $site_path ]]; then
-    site_file=$site_path
-    [[ -z $project_id ]] || fail project id specificed with config file
-    rm -f /tmp/validator_config.json
-elif [[ ! -d $site_path ]]; then
-    echo $site_path not found.
-    false
-elif [[ ! -f $site_path/cloud_iot_config.json ]]; then
-    echo $site_path/cloud_iot_config.json not found.
-    false
-elif [[ -n $project_id ]]; then
-    site_file=/tmp/validator_config.json
-    echo Writing config to $site_file
-    cat <<EOF > $site_file
-{
-  "iot_provider": $iot_provider,
-  "project_id": $quoted_id,
-  "bridge_host": $quoted_bridge,
-  "udmi_namespace": $quoted_namespace,
-  "site_model": "$site_path",
-  "feed_name": ${subscription}
-}
-EOF
-else
-    fail unknown configuration options
-fi
-
-echo Using site file $site_file:
-cat $site_file
-
-echo Running tools version $udmi_version
-
-$UDMI_ROOT/validator/bin/validate $site_file $* 2>&1 | tee $OUT_DIR/validator.log
diff --git a/bin/validator b/bin/validator
new file mode 120000
index 0000000000..ae1b6f072c
--- /dev/null
+++ b/bin/validator
@@ -0,0 +1 @@
+toolrun
\ No newline at end of file
diff --git a/common/src/main/java/com/google/udmi/util/Common.java b/common/src/main/java/com/google/udmi/util/Common.java
index 865f1bb8f8..4099fa5a70 100644
--- a/common/src/main/java/com/google/udmi/util/Common.java
+++ b/common/src/main/java/com/google/udmi/util/Common.java
@@ -51,16 +51,16 @@ public abstract class Common {
   public static final char DETAIL_SEPARATOR_CHAR = ';';
   public static final String DETAIL_SEPARATOR = DETAIL_SEPARATOR_CHAR + " ";
   public static final Joiner DETAIL_JOINER = Joiner.on(DETAIL_SEPARATOR);
-  public static final String CONFIG_CATEGORY = "config";
-  public static final String COMMANDS_CATEGORY = "commands";
   public static final String CATEGORY_PROPERTY_KEY = "category";
   public static final Pattern DEVICE_ID_ALLOWABLE = Pattern.compile("^[-_a-zA-Z0-9]+$");
   public static final Pattern POINT_NAME_ALLOWABLE = DEVICE_ID_ALLOWABLE;
   public static final int SEC_TO_MS = 1000;
   public static final String SOURCE_SEPARATOR = "+";
+  public static final String SOURCE_SEPARATOR_REGEX = "\\" + SOURCE_SEPARATOR;
 
   private static final String PREFIX_SEPARATOR = "~";
   private static final String UDMI_VERSION_ENV = "UDMI_TOOLS";
+  public static final int EXIT_CODE_ERROR = 1;
 
   /**
    * Remove the next item from the list in an exception-safe way.
diff --git a/etc/validator.out b/etc/validator.out
index c511f7348d..e53e227d31 100644
--- a/etc/validator.out
+++ b/etc/validator.out
@@ -60,7 +60,7 @@ sites/udmi_site_model/out/devices/AHU-1/events_pointset.out
   "sub_type" : "events",
   "status" : {
     "message" : "Multiple validation errors",
-    "detail" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold; While converting to json node: 1 schema violations found",
+    "detail" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold; While converting to json node: 1 schema violations found",
     "category" : "validation.device.multiple",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -70,8 +70,8 @@ sites/udmi_site_model/out/devices/AHU-1/events_pointset.out
     "extra" : [ ]
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "events_pointset: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "events_pointset: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -92,15 +92,15 @@ sites/udmi_site_model/out/devices/AHU-1/events_system.out
   "sub_folder" : "system",
   "sub_type" : "events",
   "status" : {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "events_system: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "events_system: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "events_system: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "events_system: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -115,15 +115,15 @@ sites/udmi_site_model/out/devices/AHU-1/state_localnet.out
   "sub_folder" : "localnet",
   "sub_type" : "state",
   "status" : {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_localnet: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_localnet: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_localnet: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_localnet: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -139,14 +139,14 @@ sites/udmi_site_model/out/devices/AHU-1/state.out
   "sub_type" : "state",
   "status" : {
     "message" : "Multiple validation errors",
-    "detail" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold; While converting to json node: 1 schema violations found",
+    "detail" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold; While converting to json node: 1 schema violations found",
     "category" : "validation.device.multiple",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_update: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_update: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -167,8 +167,8 @@ sites/udmi_site_model/out/devices/AHU-1/state_pointset.out
   "sub_folder" : "pointset",
   "sub_type" : "state",
   "status" : {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_pointset: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_pointset: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -178,8 +178,8 @@ sites/udmi_site_model/out/devices/AHU-1/state_pointset.out
     "extra" : [ ]
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_pointset: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_pointset: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -194,15 +194,15 @@ sites/udmi_site_model/out/devices/AHU-1/state_system.out
   "sub_folder" : "system",
   "sub_type" : "state",
   "status" : {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_system: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_system: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
   },
   "errors" : [ {
-    "message" : "Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
-    "detail" : "state_system: Timestamp jitter REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
+    "message" : "Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold",
+    "detail" : "state_system: Timestamp skew REDACTED_DURATION (REDACTED_TIMESTAMP to REDACTED_TIMESTAMP) exceeds REDACTED_DURATION threshold REDACTED_ERROR",
     "category" : "validation.device.schema",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
@@ -376,13 +376,19 @@ sites/udmi_site_model/out/devices/GAT-123/events_blobset.out
   "sub_folder" : "blobset",
   "sub_type" : "events",
   "status" : {
-    "message" : "Raw string message for GAT-123 events_blobset",
-    "detail" : "events_blobset: Raw string message for GAT-123 events_blobset REDACTED_ERROR",
-    "category" : "validation.device.schema",
+    "message" : "Multiple validation errors",
+    "detail" : "Unknown schema subFolder 'events_blobset' for GAT-123; Raw string message for GAT-123 events_blobset",
+    "category" : "validation.device.multiple",
     "timestamp" : "REDACTED_TIMESTAMP",
     "level" : 500
   },
   "errors" : [ {
+    "message" : "Unknown schema subFolder 'events_blobset' for GAT-123",
+    "detail" : "events_blobset: Unknown schema subFolder 'events_blobset' for GAT-123 REDACTED_ERROR",
+    "category" : "validation.device.schema",
+    "timestamp" : "REDACTED_TIMESTAMP",
+    "level" : 500
+  }, {
     "message" : "Raw string message for GAT-123 events_blobset",
     "detail" : "events_blobset: Raw string message for GAT-123 events_blobset REDACTED_ERROR",
     "category" : "validation.device.schema",
diff --git a/gencode/docs/configuration_execution.html b/gencode/docs/configuration_execution.html
index e43f3d5ab6..8ea23491e7 100644
--- a/gencode/docs/configuration_execution.html
+++ b/gencode/docs/configuration_execution.html
@@ -451,7 +451,7 @@ <h2 class="mb-0">
 
     <div class="enum-value" id="iot_provider_enum">
                 <h4>Must be one of:</h4>
-                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"gcp_native"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"clearblade_native"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
+                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"pref"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
                 </div>
             
 
diff --git a/gencode/docs/configuration_pod.html b/gencode/docs/configuration_pod.html
index c738eb1518..baa8f408df 100644
--- a/gencode/docs/configuration_pod.html
+++ b/gencode/docs/configuration_pod.html
@@ -6968,7 +6968,7 @@ <h2 class="mb-0">
 
     <div class="enum-value" id="iot_access_pattern1_provider_enum">
                 <h4>Must be one of:</h4>
-                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"gcp_native"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"clearblade_native"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
+                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"pref"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
                 </div>
             
 
@@ -7288,7 +7288,7 @@ <h2 class="mb-0">
 
     <div class="enum-value" id="iot_data_pattern1_provider_enum">
                 <h4>Must be one of:</h4>
-                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"gcp_native"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"clearblade_native"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
+                <ul class="list-group"><li class="list-group-item enum-item">"local"</li><li class="list-group-item enum-item">"dynamic"</li><li class="list-group-item enum-item">"implicit"</li><li class="list-group-item enum-item">"pubsub"</li><li class="list-group-item enum-item">"pref"</li><li class="list-group-item enum-item">"mqtt"</li><li class="list-group-item enum-item">"gbos"</li><li class="list-group-item enum-item">"etcd"</li><li class="list-group-item enum-item">"jwt"</li><li class="list-group-item enum-item">"clearblade"</li></ul>
                 </div>
             
 
diff --git a/gencode/java/udmi/schema/IotAccess.java b/gencode/java/udmi/schema/IotAccess.java
index 0749530044..a307741ddb 100644
--- a/gencode/java/udmi/schema/IotAccess.java
+++ b/gencode/java/udmi/schema/IotAccess.java
@@ -79,12 +79,11 @@ public enum IotProvider {
         DYNAMIC("dynamic"),
         IMPLICIT("implicit"),
         PUBSUB("pubsub"),
-        ETCD("etcd"),
+        PREF("pref"),
         MQTT("mqtt"),
-        JWT("jwt"),
-        GCP_NATIVE("gcp_native"),
         GBOS("gbos"),
-        CLEARBLADE_NATIVE("clearblade_native"),
+        ETCD("etcd"),
+        JWT("jwt"),
         CLEARBLADE("clearblade");
         private final String value;
         private final static Map<String, IotAccess.IotProvider> CONSTANTS = new HashMap<String, IotAccess.IotProvider>();
diff --git a/schema/common.json b/schema/common.json
index 47e886226c..a85cbea370 100644
--- a/schema/common.json
+++ b/schema/common.json
@@ -39,12 +39,11 @@
         "dynamic",
         "implicit",
         "pubsub",
-        "etcd",
+        "pref",
         "mqtt",
-        "jwt",
-        "gcp_native",
         "gbos",
-        "clearblade_native",
+        "etcd",
+        "jwt",
         "clearblade"
       ]
     },
diff --git a/tests/traces/lengthy/devices/AHU-1/001_events_pointset.json b/tests/traces/lengthy/devices/AHU-1/001_events_pointset.json
new file mode 100644
index 0000000000..bcde2868f8
--- /dev/null
+++ b/tests/traces/lengthy/devices/AHU-1/001_events_pointset.json
@@ -0,0 +1,15 @@
+{
+  "points" : {
+    "filter_alarm_pressure_status" : {
+      "present_value" : true
+    },
+    "filter_differential_pressure_setpoint" : {
+      "present_value" : 98
+    },
+    "filter_differential_pressure_sensor" : {
+      "present_value" : 11
+    }
+  },
+  "timestamp" : "2022-07-17T05:04:26Z",
+  "version" : "1.5.1"
+}
diff --git a/tests/traces/lengthy/devices/AHU-1/002_events_pointset.json b/tests/traces/lengthy/devices/AHU-1/002_events_pointset.json
new file mode 100644
index 0000000000..86561c0aca
--- /dev/null
+++ b/tests/traces/lengthy/devices/AHU-1/002_events_pointset.json
@@ -0,0 +1,15 @@
+{
+  "points" : {
+    "filter_alarm_pressure_status" : {
+      "present_value" : true
+    },
+    "filter_differential_pressure_setpoint" : {
+      "present_value" : 98
+    },
+    "filter_differential_pressure_sensor" : {
+      "present_value" : 11
+    }
+  },
+  "timestamp" : "2022-07-17T05:34:36+0000",
+  "version" : "1.5.1"
+}
diff --git a/tests/traces/lengthy/devices/AHU-1/003_events_pointset.json b/tests/traces/lengthy/devices/AHU-1/003_events_pointset.json
new file mode 100644
index 0000000000..0dadb70086
--- /dev/null
+++ b/tests/traces/lengthy/devices/AHU-1/003_events_pointset.json
@@ -0,0 +1,15 @@
+{
+  "points" : {
+    "filter_alarm_pressure_status" : {
+      "present_value" : true
+    },
+    "filter_differential_pressure_setpoint" : {
+      "present_value" : 98
+    },
+    "filter_differential_pressure_sensor" : {
+      "present_value" : 11
+    }
+  },
+  "timestamp" : "2022-07-17T06:07:36Z", // Over threshold from _initial_ event
+  "version" : "1.5.1"
+}
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.attr b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.attr
new file mode 100644
index 0000000000..a6ffd57222
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.attr
@@ -0,0 +1,10 @@
+{
+  "deviceNumId" : "00000062256946",
+  "publishTime" : "1999-10-20T01:02:03Z",
+  "subFolder" : "pointset",
+  "deviceRegistryId" : "ZZ-TRI-FECTA",
+  "subType" : "events",
+  "msgSource" : "003_events_pointset.json",
+  "deviceId" : "AHU-1",
+  "projectId" : "playback-project"
+}
\ No newline at end of file
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.json b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.json
new file mode 100644
index 0000000000..5615cf9118
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.json
@@ -0,0 +1,15 @@
+{
+  "points" : {
+    "filter_alarm_pressure_status" : {
+      "present_value" : true
+    },
+    "filter_differential_pressure_setpoint" : {
+      "present_value" : 98
+    },
+    "filter_differential_pressure_sensor" : {
+      "present_value" : 11
+    }
+  },
+  "timestamp" : "1999-10-20T01:02:03Z",
+  "version" : "1.5.1"
+}
\ No newline at end of file
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.out b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.out
new file mode 100644
index 0000000000..dc3cc3bfd9
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_pointset.out
@@ -0,0 +1,17 @@
+{
+  "timestamp" : "1999-10-20T01:02:03Z",
+  "version" : "1.5.1",
+  "sub_folder" : "pointset",
+  "sub_type" : "events",
+  "status" : {
+    "message" : "Successful validation",
+    "category" : "validation.device.receive",
+    "timestamp" : "1999-10-20T01:02:03Z",
+    "level" : 200
+  },
+  "pointset" : {
+    "missing" : [ ],
+    "extra" : [ ]
+  },
+  "errors" : [ ]
+}
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_system.attr b/tests/traces/lengthy/expected/devices/AHU-1/events_system.attr
new file mode 100644
index 0000000000..f21e8d3520
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_system.attr
@@ -0,0 +1,10 @@
+{
+  "deviceNumId" : "00000062256946",
+  "publishTime" : "2022-07-19T05:04:36+0000",
+  "subFolder" : "system",
+  "deviceRegistryId" : "ZZ-TRI-FECTA",
+  "subType" : "events",
+  "msgSource" : "002_events_system.json",
+  "deviceId" : "AHU-1",
+  "projectId" : "playback-project"
+}
\ No newline at end of file
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_system.json b/tests/traces/lengthy/expected/devices/AHU-1/events_system.json
new file mode 100644
index 0000000000..703d8551ce
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_system.json
@@ -0,0 +1,13 @@
+{
+  "logentries" : [ {
+    "message" : "1999-10-20T01:02:03Z sending test message #70",
+    "category" : "device.testing",
+    "timestamp" : "1999-10-20T01:02:03Z",
+    "level" : 200
+  } ],
+  "metrics" : {
+    "mem_total_mb" : 25164.88
+  },
+  "timestamp" : "2022-07-19T05:04:36+0000",
+  "version" : "1.5.1"
+}
\ No newline at end of file
diff --git a/tests/traces/lengthy/expected/devices/AHU-1/events_system.out b/tests/traces/lengthy/expected/devices/AHU-1/events_system.out
new file mode 100644
index 0000000000..a3237dc0c7
--- /dev/null
+++ b/tests/traces/lengthy/expected/devices/AHU-1/events_system.out
@@ -0,0 +1,13 @@
+{
+  "timestamp" : "1999-10-20T01:02:03Z",
+  "version" : "1.5.1",
+  "sub_folder" : "system",
+  "sub_type" : "events",
+  "status" : {
+    "message" : "Successful validation",
+    "category" : "validation.device.receive",
+    "timestamp" : "1999-10-20T01:02:03Z",
+    "level" : 200
+  },
+  "errors" : [ ]
+}
diff --git a/tests/traces/lengthy/expected/validation_report.json b/tests/traces/lengthy/expected/validation_report.json
new file mode 100644
index 0000000000..4f9f06b9d6
--- /dev/null
+++ b/tests/traces/lengthy/expected/validation_report.json
@@ -0,0 +1,56 @@
+{
+  "timestamp" : "1999-10-20T01:02:03Z",
+  "version" : "1.5.1",
+  "cloud_version" : {
+    "deployed_by" : "Default implementation for class com.google.daq.mqtt.validator.MessageReadingClient"
+  },
+  REDACTED,
+  "start_time" : "1999-10-20T01:02:03Z",
+  "summary" : {
+    "correct_devices" : [ "AHU-1" ],
+    "extra_devices" : [ ],
+    "missing_devices" : [ ],
+    "error_devices" : [ "AHU-22", "GAT-123", "SNS-4" ]
+  },
+  "devices" : {
+    "AHU-1" : {
+      "last_seen" : "1999-10-20T01:02:03Z",
+      "status" : {
+        "message" : "Successful validation",
+        "category" : "validation.device.receive",
+        "timestamp" : "1999-10-20T01:02:03Z",
+        "level" : 200
+      }
+    },
+    "AHU-22" : {
+      "last_seen" : "1999-10-20T01:02:03Z",
+      "status" : {
+        "message" : "While converting to json node: 2 schema violations found",
+        "detail" : "events_pointset: While converting to json node: 2 schema violations found @Validator.validateMessage(REDACTED); 2 schema violations found; object instance has properties which are not allowed by the schema: [\"yesvalue\"]; object has missing required properties ([\"version\"])",
+        "category" : "validation.device.schema",
+        "timestamp" : "1999-10-20T01:02:03Z",
+        "level" : 500
+      }
+    },
+    "GAT-123" : {
+      "last_seen" : "1999-10-20T01:02:03Z",
+      "status" : {
+        "message" : "missing pointset subblock",
+        "detail" : "state_update: missing pointset subblock",
+        "category" : "validation.device.schema",
+        "timestamp" : "1999-10-20T01:02:03Z",
+        "level" : 500
+      }
+    },
+    "SNS-4" : {
+      "last_seen" : "1999-10-20T01:02:03Z",
+      "status" : {
+        "message" : "Multiple validation errors",
+        "detail" : "While converting to json node: 1 schema violations found; Device has extra points: rocket_count, triangulating_axis_y; Unexpected character ('{' (code 123)): was expecting double-quote to start field name\n at [Source: (File), line: 1, column: 3]; While converting to json node: 1 schema violations found",
+        "category" : "validation.device.multiple",
+        "timestamp" : "1999-10-20T01:02:03Z",
+        "level" : 500
+      }
+    }
+  }
+}
diff --git a/udmis/src/main/java/com/google/bos/udmi/service/access/PubSubIotAccessProvider.java b/udmis/src/main/java/com/google/bos/udmi/service/access/PubSubIotAccessProvider.java
index 04bc3454ef..46fcfebe63 100644
--- a/udmis/src/main/java/com/google/bos/udmi/service/access/PubSubIotAccessProvider.java
+++ b/udmis/src/main/java/com/google/bos/udmi/service/access/PubSubIotAccessProvider.java
@@ -4,8 +4,6 @@
 import static com.google.bos.udmi.service.messaging.impl.PubSubPipe.SOURCE_KEY;
 import static com.google.bos.udmi.service.messaging.impl.PubSubPipe.getTransportChannelProvider;
 import static com.google.udmi.util.Common.CATEGORY_PROPERTY_KEY;
-import static com.google.udmi.util.Common.COMMANDS_CATEGORY;
-import static com.google.udmi.util.Common.CONFIG_CATEGORY;
 import static com.google.udmi.util.Common.DEVICE_ID_KEY;
 import static com.google.udmi.util.Common.REGISTRY_ID_PROPERTY_KEY;
 import static com.google.udmi.util.Common.SOURCE_SEPARATOR;
@@ -34,6 +32,7 @@
 import udmi.schema.CloudModel;
 import udmi.schema.Envelope;
 import udmi.schema.Envelope.SubFolder;
+import udmi.schema.Envelope.SubType;
 import udmi.schema.IotAccess;
 
 /**
@@ -41,6 +40,8 @@
  */
 public class PubSubIotAccessProvider extends IotAccessBase {
 
+  private static final String COMMAND_SUBTYPE = SubType.COMMANDS.value();
+  private static final String CONFIG_SUBTYPE = SubType.CONFIG.value();
   private final String projectId;
   private final String topic;
   private final Publisher publisher;
@@ -76,12 +77,12 @@ public boolean isEnabled() {
 
   @Override
   public void sendCommandBase(Envelope envelope, SubFolder folder, String message) {
-    publish(envelope, COMMANDS_CATEGORY, folder, message);
+    publish(envelope, COMMAND_SUBTYPE, folder, message);
   }
 
   @Override
   public String updateConfig(Envelope envelope, String config, Long version) {
-    publish(envelope, CONFIG_CATEGORY, null, config);
+    publish(envelope, CONFIG_SUBTYPE, null, config);
     return config;
   }
 
diff --git a/validator/.idea/runConfigurations/Validator_config.xml b/validator/.idea/runConfigurations/Validator_config.xml
index 9b669ddd3b..2b01c481a8 100644
--- a/validator/.idea/runConfigurations/Validator_config.xml
+++ b/validator/.idea/runConfigurations/Validator_config.xml
@@ -3,7 +3,7 @@
     <option name="ALTERNATIVE_JRE_PATH" value="17" />
     <option name="MAIN_CLASS_NAME" value="com.google.daq.mqtt.validator.Validator" />
     <module name="validator" />
-    <option name="PROGRAM_PARAMETERS" value="/tmp/validator_config.json" />
+    <option name="PROGRAM_PARAMETERS" value="out/validator_conf.json" />
     <option name="WORKING_DIRECTORY" value="$PROJECT_DIR$/.." />
     <method v="2">
       <option name="Make" enabled="true" />
diff --git a/validator/bin/build b/validator/bin/build
index 22766e12c9..31a4163a9b 100755
--- a/validator/bin/build
+++ b/validator/bin/build
@@ -16,9 +16,17 @@ while [[ $# -gt 0 ]]; do
 done
 
 ROOT=$(dirname $0)/..
-BASE=..
+UDMI_ROOT=..
 cd $ROOT
 
+jarfile=build/libs/validator-1.0-SNAPSHOT-all.jar
+
+newest=$(ls -rt `find src/ $UDMI_ROOT/gencode/java -type f` | tail -n 1)
+if [[ -z $check && -z $covg && -f $jarfile && $jarfile -nt $newest ]]; then
+    echo $jarfile up-to-date, skipping build.
+    exit 0
+fi
+
 echo Cleaning $ROOT/build/
 rm -rf build/
 mkdir build/
@@ -28,17 +36,9 @@ cp -a ../bin build/udmi_bin/
 cp -a ../etc build/udmi_etc/
 cp -a ../schema build/schema/
 
-jarfile=build/libs/validator-1.0-SNAPSHOT-all.jar
-
-newest=$(ls -rt `find src/ $BASE/gencode/java -type f` | tail -n 1)
-if [[ -z $check && -z $covg && -f $jarfile && $jarfile -nt $newest ]]; then
-    echo $jarfile up-to-date, skipping build.
-    exit 0
-fi
-
 if [[ -n $check ]]; then
-    (cd $BASE/sites/udmi_site_model/devices; rm -rf XXX-* ACME-*)
-    $BASE/bin/clone_model
+    (cd $UDMI_ROOT/sites/udmi_site_model/devices; rm -rf XXX-* ACME-*)
+    $UDMI_ROOT/bin/clone_model
 fi
 
 echo Java version $(java --version)
diff --git a/validator/bin/validate b/validator/bin/validate
deleted file mode 100755
index b072b6e109..0000000000
--- a/validator/bin/validate
+++ /dev/null
@@ -1,69 +0,0 @@
-#!/bin/bash -e
-
-if [[ -f $1 ]]; then
-    config_file=$(realpath $1)
-    schema=schema
-    target=config
-    shift 1
-    args=$config_file
-else
-    project=$1
-    schema=$2
-    target=$3
-    subscription=$4
-    sitepath=$(realpath $5)
-    shift 5
-    args="-p $project -s $sitepath"
-fi
-devices="$*"
-
-ROOT=$(dirname $0)/../..
-cd $ROOT
-
-if [[ "$schema" != schema ]]; then
-    echo Currently only default schema supported.
-    false
-fi
-
-jarfile=validator/build/libs/validator-1.0-SNAPSHOT-all.jar
-export UDMI_VERSION=`cd $ROOT; git describe`
-
-echo Building validator...
-validator/bin/build
-
-auth_type=`jq -r .type ~/.config/gcloud/application_default_credentials.json || true`
-echo Application default credentials auth type ${auth_type:-unknown}
-
-if [[ -n $sitepath ]]; then
-    echo Cleaning output directory $sitepath/out/
-    rm -rf $sitepath/out/
-fi
-
-echo Executing validator $schema $target...
-
-if [[ $target == reflect ]]; then
-    if [[ $subscription != -- ]]; then
-        echo Unsupported subsription $subscription with reflect
-        false
-    elif [[ -n $devices ]]; then
-        srcargs="-- $devices"
-    fi
-elif [[ $target == pubsub ]]; then
-    srcargs="-t $subscription"
-elif [[ $target == trace ]]; then
-    srcargs="-r $subscription -- $devices"
-elif [[ $target == config ]]; then
-    true
-else
-    echo Unsupported target $target
-    false
-fi
-
-echo Exeucting from $(pwd)
-echo java -jar $jarfile $args $srcargs
-
-error=0
-java -jar $jarfile $args $srcargs || error=$?
-
-echo Validation complete, exit $error
-exit $error
diff --git a/validator/src/main/java/com/google/bos/iot/core/proxy/IotReflectorClient.java b/validator/src/main/java/com/google/bos/iot/core/proxy/IotReflectorClient.java
index 594211d74b..3876ca4d6e 100644
--- a/validator/src/main/java/com/google/bos/iot/core/proxy/IotReflectorClient.java
+++ b/validator/src/main/java/com/google/bos/iot/core/proxy/IotReflectorClient.java
@@ -7,6 +7,8 @@
 import static com.google.udmi.util.Common.DEVICE_ID_KEY;
 import static com.google.udmi.util.Common.GATEWAY_ID_KEY;
 import static com.google.udmi.util.Common.PUBLISH_TIME_KEY;
+import static com.google.udmi.util.Common.SOURCE_KEY;
+import static com.google.udmi.util.Common.SOURCE_SEPARATOR_REGEX;
 import static com.google.udmi.util.Common.SUBFOLDER_PROPERTY_KEY;
 import static com.google.udmi.util.Common.SUBTYPE_PROPERTY_KEY;
 import static com.google.udmi.util.Common.TIMESTAMP_KEY;
@@ -16,6 +18,7 @@
 import static com.google.udmi.util.GeneralUtils.decodeBase64;
 import static com.google.udmi.util.GeneralUtils.getTimestamp;
 import static com.google.udmi.util.GeneralUtils.ifNotNullGet;
+import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
 import static com.google.udmi.util.GeneralUtils.isTrue;
 import static com.google.udmi.util.JsonUtil.asMap;
 import static com.google.udmi.util.JsonUtil.convertTo;
@@ -23,8 +26,10 @@
 import static com.google.udmi.util.JsonUtil.isoConvert;
 import static com.google.udmi.util.JsonUtil.mapCast;
 import static com.google.udmi.util.JsonUtil.stringify;
+import static com.google.udmi.util.JsonUtil.stringifyTerse;
 import static com.google.udmi.util.JsonUtil.toMap;
 import static com.google.udmi.util.JsonUtil.toObject;
+import static com.google.udmi.util.PubSubReflector.USER_NAME_DEFAULT;
 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.Optional.ofNullable;
@@ -51,6 +56,7 @@
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
 import org.jetbrains.annotations.NotNull;
 import udmi.schema.Credential;
 import udmi.schema.Envelope;
@@ -97,6 +103,8 @@ public class IotReflectorClient implements MessagePublisher {
   private final String updateTo;
   private final IotProvider iotProvider;
   private final boolean enforceUdmiVersion;
+  private final Function<Envelope, Boolean> messageFilter;
+  private final String userName;
   private Date reflectorStateTimestamp;
   private boolean isInstallValid;
   private boolean active;
@@ -111,28 +119,36 @@ public class IotReflectorClient implements MessagePublisher {
    * @param requiredVersion version of the functions that are required by the tools
    */
   public IotReflectorClient(ExecutionConfiguration iotConfig, int requiredVersion) {
+    this(iotConfig, requiredVersion, null);
+  }
+
+  /**
+   * Basic client that accepts a custom message filter.
+   */
+  public IotReflectorClient(ExecutionConfiguration iotConfig, int requiredVersion,
+      Function<Envelope, Boolean> messageFilter) {
     Preconditions.checkState(requiredVersion >= TOOLS_FUNCTIONS_VERSION,
         format("Min required version %s not satisfied by tools version %s", TOOLS_FUNCTIONS_VERSION,
             requiredVersion));
     this.requiredVersion = requiredVersion;
     this.enforceUdmiVersion = isTrue(iotConfig.enforce_version);
+    this.messageFilter = ofNullable(messageFilter).orElse(this::userMessageFilter);
     registryId = SiteModel.getRegistryActual(iotConfig);
     projectId = iotConfig.project_id;
     udmiVersion = ofNullable(iotConfig.udmi_version).orElseGet(Common::getUdmiVersion);
     updateTo = iotConfig.update_to;
-    String cloudRegion = ofNullable(iotConfig.reflect_region)
-        .orElse(iotConfig.cloud_region);
-    String prefix = getNamespacePrefix(iotConfig.udmi_namespace);
     iotProvider = ofNullable(iotConfig.iot_provider).orElse(IotProvider.GBOS);
+    userName = ofNullable(iotConfig.user_name).orElse(USER_NAME_DEFAULT);
     iotConfig.iot_provider = iotProvider;
-    subscriptionId = format("%s/%s/%s/%s%s/%s",
-        iotConfig.iot_provider, projectId, cloudRegion, prefix, UDMI_REFLECT, registryId);
-
+    String prefix = getNamespacePrefix(iotConfig.udmi_namespace);
+    String clientId = format("//%s/%s/%s %s", iotProvider, projectId, prefix, registryId);
     try {
+      System.err.println("Instantiating reflector client " + clientId);
       publisher = MessagePublisher.from(iotConfig, this::messageHandler, this::errorHandler);
-    } catch (Exception e1) {
-      throw new RuntimeException("While creating client " + subscriptionId, e1);
+    } catch (Exception e) {
+      throw new RuntimeException("While creating reflector client " + clientId, e);
     }
+    subscriptionId = publisher.getSubscriptionId();
     System.err.println("Subscribed to " + subscriptionId);
 
     try {
@@ -161,6 +177,10 @@ public IotReflectorClient(ExecutionConfiguration iotConfig, int requiredVersion)
     }
   }
 
+  private boolean userMessageFilter(Envelope envelope) {
+    return envelope.source == null || userName.equals(envelope.source);
+  }
+
   /**
    * Make an execution configuration that's used for reflector operations.
    *
@@ -254,15 +274,15 @@ private void messageHandler(String topic, String payload) {
     Map<String, Object> messageMap = asMap(
         new String(base64 ? Base64.decodeBase64(rawData) : rawData));
     try {
-      List<String> parts = parseMessageTopic(topic);
-      String category = parts.get(0);
-
-      if (Common.CONFIG_CATEGORY.equals(category)) {
+      Envelope envelope = parseMessageTopic(topic);
+      ifNotNullThen(envelope.source, source -> messageMap.put(SOURCE_KEY, source),
+          () -> messageMap.remove(SOURCE_KEY));
+      if (SubType.CONFIG == envelope.subType) {
         ensureCloudSync(messageMap);
-      } else if (Common.COMMANDS_CATEGORY.equals(category)) {
-        handleCommandEnvelope(messageMap);
+      } else if (SubType.COMMANDS == envelope.subType) {
+        handleEncapsulatedMessage(envelope, messageMap);
       } else {
-        throw new RuntimeException("Unknown message category " + category);
+        throw new RuntimeException("Unknown message category " + envelope.subType);
       }
     } catch (Exception e) {
       if (isInstallValid) {
@@ -274,8 +294,8 @@ private void messageHandler(String topic, String payload) {
     }
   }
 
-  private void handleCommandEnvelope(Map<String, Object> messageMap) {
-    if (!isInstallValid) {
+  private void handleEncapsulatedMessage(Envelope envelope, Map<String, Object> messageMap) {
+    if (!isInstallValid || !messageFilter.apply(envelope)) {
       return;
     }
     Map<String, String> attributes = extractAttributes(messageMap);
@@ -404,7 +424,7 @@ private boolean ensureCloudSync(Map<String, Object> message) {
     return false;
   }
 
-  private List<String> parseMessageTopic(String topic) {
+  private Envelope parseMessageTopic(String topic) {
     List<String> parts = new ArrayList<>(Arrays.asList(topic.substring(1).split("/")));
     String leader = parts.remove(0);
     if ("devices".equals(leader)) {
@@ -425,7 +445,23 @@ private List<String> parseMessageTopic(String topic) {
     } else {
       throw new RuntimeException("Unknown topic string " + topic);
     }
-    return parts;
+
+    Envelope envelope = new Envelope();
+    envelope.deviceRegistryId = registryId;
+
+    String[] bits1 = parts.remove(0).split(SOURCE_SEPARATOR_REGEX);
+    checkState(parts.isEmpty() || bits1.length == 1, "Malformed topic: " + topic);
+    envelope.subType = SubType.fromValue(bits1[0]);
+    if (parts.isEmpty()) {
+      envelope.source = bits1.length > 1 ? bits1[1] : null;
+    } else {
+      String[] bits2 = parts.remove(0).split(SOURCE_SEPARATOR_REGEX);
+      envelope.subFolder = SubFolder.fromValue(bits2[0]);
+      envelope.source = bits2.length > 1 ? bits2[1] : null;
+    }
+    checkState(parts.isEmpty());
+
+    return envelope;
   }
 
   private void errorHandler(Throwable throwable) {
diff --git a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
index f7ea54c3a1..dfa2147bf8 100644
--- a/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
+++ b/validator/src/main/java/com/google/daq/mqtt/registrar/Registrar.java
@@ -117,7 +117,6 @@ public class Registrar {
   private static final String MODEL_SUB_TYPE = "model";
   private static final boolean DEFAULT_BLOCK_UNKNOWN = false;
   private static final int EACH_ITEM_TIMEOUT_SEC = 60;
-  private static final int EXIT_CODE_ERROR = 1;
   private static final Map<String, Class<? extends Summarizer>> SUMMARIZERS = ImmutableMap.of(
       ".json", Summarizer.JsonSummarizer.class,
       ".csv", Summarizer.CsvSummarizer.class);
@@ -161,7 +160,7 @@ public static void main(String[] args) {
     } catch (Exception e) {
       System.err.println("Exception in main: " + friendlyStackTrace(e));
       e.printStackTrace();
-      System.exit(EXIT_CODE_ERROR);
+      System.exit(Common.EXIT_CODE_ERROR);
     }
 
     // Force exist because PubSub Subscriber in PubSubReflector does not shut down properly.
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
index c68e136bbc..5e02e655cb 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/CloudIotManager.java
@@ -14,8 +14,8 @@
 import static java.util.Objects.requireNonNull;
 import static java.util.Optional.ofNullable;
 import static udmi.schema.IotAccess.IotProvider.GBOS;
-import static udmi.schema.IotAccess.IotProvider.GCP_NATIVE;
 import static udmi.schema.IotAccess.IotProvider.MQTT;
+import static udmi.schema.IotAccess.IotProvider.PUBSUB;
 
 import com.google.common.collect.ImmutableList;
 import com.google.udmi.util.GeneralUtils;
@@ -67,7 +67,7 @@ public CloudIotManager(String projectId, File siteDir, String altRegistry,
     checkNotNull(projectId, "project id undefined");
     this.siteModel = checkNotNull(siteDir, "site directory undefined");
     checkState(siteDir.isDirectory(), "not a directory " + siteDir.getAbsolutePath());
-    this.useReflectClient = iotProvider != null && iotProvider != GCP_NATIVE;
+    this.useReflectClient = iotProvider != null && iotProvider != PUBSUB;
     this.projectId = projectId;
     File cloudConfig = new File(siteDir, CLOUD_IOT_CONFIG_JSON);
     try {
@@ -124,7 +124,7 @@ public CloudIotManager(ExecutionConfiguration config) {
 
   private static boolean shouldUseReflectorClient(ExecutionConfiguration config) {
     return (config.reflector_endpoint != null)
-        || ofNullable(config.iot_provider).orElse(GCP_NATIVE) != GCP_NATIVE;
+        || ofNullable(config.iot_provider).orElse(PUBSUB) != PUBSUB;
   }
 
   /**
@@ -192,7 +192,7 @@ private IotProvider makeIotProvider() {
       return new IotReflectorClient(executionConfiguration);
     }
 
-    if (executionConfiguration.iot_provider == GCP_NATIVE) {
+    if (executionConfiguration.iot_provider == PUBSUB) {
       return null;
     }
 
diff --git a/validator/src/main/java/com/google/daq/mqtt/util/MessagePublisher.java b/validator/src/main/java/com/google/daq/mqtt/util/MessagePublisher.java
index d62b642a45..46776ace21 100644
--- a/validator/src/main/java/com/google/daq/mqtt/util/MessagePublisher.java
+++ b/validator/src/main/java/com/google/daq/mqtt/util/MessagePublisher.java
@@ -4,7 +4,7 @@
 import static java.util.Optional.ofNullable;
 import static udmi.schema.IotAccess.IotProvider.IMPLICIT;
 import static udmi.schema.IotAccess.IotProvider.JWT;
-import static udmi.schema.IotAccess.IotProvider.PUBSUB;
+import static udmi.schema.IotAccess.IotProvider.PREF;
 
 import com.google.bos.iot.core.proxy.MqttPublisher;
 import com.google.daq.mqtt.validator.Validator.MessageBundle;
@@ -38,7 +38,7 @@ static MessagePublisher from(ExecutionConfiguration iotConfig,
     if (iotConfig.reflector_endpoint != null && iotProvider != IMPLICIT) {
       iotConfig.reflector_endpoint = null;
     }
-    if (PUBSUB == iotConfig.iot_provider) {
+    if (iotProvider == PREF) {
       return PubSubReflector.from(iotConfig, messageHandler, errorHandler);
     }
     return MqttPublisher.from(iotConfig, messageHandler, errorHandler);
diff --git a/validator/src/main/java/com/google/daq/mqtt/validator/LoggingHandler.java b/validator/src/main/java/com/google/daq/mqtt/validator/LoggingHandler.java
index 7dadc339ea..677461ed2a 100644
--- a/validator/src/main/java/com/google/daq/mqtt/validator/LoggingHandler.java
+++ b/validator/src/main/java/com/google/daq/mqtt/validator/LoggingHandler.java
@@ -19,8 +19,10 @@ public LoggingHandler(BiConsumer<Level, String> logger) {
   }
 
   private static void systemLogger(Level level, String message) {
-    PrintStream printStream = level.value() >= Level.WARNING.value() ? System.err : System.out;
-    printStream.println(message);
+    if (level.value() >= Level.DEBUG.value()) {
+      PrintStream printStream = level.value() >= Level.WARNING.value() ? System.err : System.out;
+      printStream.println(message);
+    }
   }
 
   void debug(String message) {
@@ -50,4 +52,12 @@ void error(String message) {
   void error(String format, Object... args) {
     error(format(format, args));
   }
+
+  void trace(String message) {
+    outputLogger.accept(Level.TRACE, message);
+  }
+
+  void trace(String format, Object... args) {
+    trace(format(format, args));
+  }
 }
diff --git a/validator/src/main/java/com/google/daq/mqtt/validator/ReportingDevice.java b/validator/src/main/java/com/google/daq/mqtt/validator/ReportingDevice.java
index 8aa22dac10..e8e50f56d1 100644
--- a/validator/src/main/java/com/google/daq/mqtt/validator/ReportingDevice.java
+++ b/validator/src/main/java/com/google/daq/mqtt/validator/ReportingDevice.java
@@ -14,7 +14,6 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.stream.Collectors;
 import udmi.schema.Category;
@@ -172,10 +171,9 @@ public boolean hasErrors() {
    * Validate a message against specific message-type expectations (outside of base schema).
    *
    * @param message    Message to validate
-   * @param timestamp  message timestamp string (rather than pull from typed object)
    * @param attributes message attributes
    */
-  public void validateMessageType(Object message, Date timestamp, Map<String, String> attributes) {
+  public void validateMessageType(Object message, Map<String, String> attributes) {
     if (reportingPointset == null) {
       return;
     }
@@ -321,9 +319,16 @@ private Date getThreshold(Instant now) {
     return Date.from(now.minusSeconds(THRESHOLD_SEC));
   }
 
-  public boolean markMessageType(String schemaName, Instant now) {
-    Date previous = messageMarks.put(schemaName, getTimestamp());
-    return previous == null || previous.before(getThreshold(now));
+  /**
+   * Check if a message schema should be processed, to filter out too frequent processing.
+   */
+  public boolean processMessageSchema(String schemaName, Instant now) {
+    Date previous = messageMarks.get(schemaName);
+    if (previous == null || previous.before(getThreshold(now))) {
+      messageMarks.put(schemaName, Date.from(now));
+      return true;
+    }
+    return false;
   }
 
   public Date getLastSeen() {
diff --git a/validator/src/main/java/com/google/daq/mqtt/validator/SchemaTester.java b/validator/src/main/java/com/google/daq/mqtt/validator/SchemaTester.java
new file mode 100644
index 0000000000..3ada106540
--- /dev/null
+++ b/validator/src/main/java/com/google/daq/mqtt/validator/SchemaTester.java
@@ -0,0 +1,27 @@
+package com.google.daq.mqtt.validator;
+
+import com.google.daq.mqtt.util.ExceptionMap;
+import java.util.Arrays;
+
+/**
+ * Variant that uses the validator core to test schema examples from files.
+ */
+public class SchemaTester {
+
+  /**
+   * Let's go.
+   */
+  public static void main(String[] args) {
+    try {
+      Validator validator = new Validator(Arrays.asList(args));
+      validator.messageLoop();
+    } catch (ExceptionMap processingException) {
+      System.exit(2);
+    } catch (Exception e) {
+      e.printStackTrace();
+      System.err.flush();
+      System.exit(-1);
+    }
+    System.exit(0);
+  }
+}
diff --git a/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java b/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
index dccb8c1f5e..60cef515f2 100644
--- a/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
+++ b/validator/src/main/java/com/google/daq/mqtt/validator/Validator.java
@@ -1,5 +1,6 @@
 package com.google.daq.mqtt.validator;
 
+import static com.google.api.client.util.Preconditions.checkState;
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.collect.ImmutableList.toImmutableList;
 import static com.google.daq.mqtt.registrar.Registrar.BASE_DIR;
@@ -10,6 +11,7 @@
 import static com.google.daq.mqtt.validator.ReportingDevice.typeFolderPairKey;
 import static com.google.udmi.util.Common.ERROR_KEY;
 import static com.google.udmi.util.Common.EXCEPTION_KEY;
+import static com.google.udmi.util.Common.EXIT_CODE_ERROR;
 import static com.google.udmi.util.Common.GCP_REFLECT_KEY_PKCS8;
 import static com.google.udmi.util.Common.MESSAGE_KEY;
 import static com.google.udmi.util.Common.NO_SITE;
@@ -30,6 +32,7 @@
 import static com.google.udmi.util.JsonUtil.getInstant;
 import static com.google.udmi.util.JsonUtil.isoConvert;
 import static com.google.udmi.util.JsonUtil.mapCast;
+import static com.google.udmi.util.JsonUtil.safeSleep;
 import static com.google.udmi.util.SiteModel.DEVICES_DIR;
 import static java.lang.String.format;
 import static java.util.Optional.ofNullable;
@@ -52,7 +55,6 @@
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.ImmutableSet;
 import com.google.daq.mqtt.util.CloudIotManager;
-import com.google.daq.mqtt.util.ConfigUtil;
 import com.google.daq.mqtt.util.ExceptionMap;
 import com.google.daq.mqtt.util.ExceptionMap.ErrorTree;
 import com.google.daq.mqtt.util.FileDataSink;
@@ -88,6 +90,7 @@
 import java.util.Set;
 import java.util.TreeMap;
 import java.util.TreeSet;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledFuture;
@@ -99,6 +102,7 @@
 import org.apache.commons.io.FileUtils;
 import udmi.schema.Category;
 import udmi.schema.DeviceValidationEvents;
+import udmi.schema.Envelope;
 import udmi.schema.Envelope.SubFolder;
 import udmi.schema.Envelope.SubType;
 import udmi.schema.ExecutionConfiguration;
@@ -142,6 +146,8 @@ public class Validator {
   private static final Set<String> INTERESTING_TYPES = ImmutableSet.of(
       SubType.EVENTS.value(),
       SubType.STATE.value());
+  private static final Set<String> IGNORE_FOLDERS = ImmutableSet.of(
+      SubFolder.ERROR.value());
   private static final Map<String, Class<?>> CONTENT_VALIDATORS = ImmutableMap.of(
       STATE_UPDATE_SCHEMA, State.class,
       EVENTS_POINTSET_SCHEMA, PointsetEvents.class,
@@ -158,6 +164,8 @@ public class Validator {
   private static final Date START_TIME = new Date();
   private static final int TIMESTAMP_JITTER_SEC = 60;
   private static final String UDMI_CONFIG_JSON_FILE = "udmi_config.json";
+  private static final String TOOL_NAME = "validator";
+  private static final long THREAD_JOIN_MS = 1000;
   private final Map<String, ReportingDevice> reportingDevices = new TreeMap<>();
   private final Set<String> extraDevices = new TreeSet<>();
   private final Set<String> processedDevices = new TreeSet<>();
@@ -166,7 +174,8 @@ public class Validator {
   private final ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
   private final Map<String, AtomicInteger> deviceMessageIndex = new HashMap<>();
   private final List<MessagePublisher> dataSinks = new ArrayList<>();
-  private final Set<String> targetDevices;
+  private final CountDownLatch messageLoopStarted = new CountDownLatch(1);
+  private Set<String> targetDevices;
   private final LoggingHandler outputLogger;
   private ImmutableSet<String> expectedDevices;
   private File outBaseDir;
@@ -179,6 +188,7 @@ public class Validator {
   private boolean simulatedMessages;
   private Instant mockNow = null;
   private boolean forceUpgrade;
+  private SiteModel siteModel;
 
   /**
    * Create a simplistic validator for encapsulated use.
@@ -200,32 +210,53 @@ public Validator(List<String> argList) {
     outputLogger = new LoggingHandler();
     List<String> listCopy = new ArrayList<>(argList);
     parseArgs(listCopy);
+    if (client == null) {
+      validateReflector();
+    }
+    targetDevices = Set.copyOf(listCopy);
+  }
 
-    if (schemaMap == null) {
-      setSchemaSpec("schema");
+  public Validator() {
+    outputLogger = new LoggingHandler();
+  }
+
+  Validator processArgs(List<String> argListRaw) {
+    List<String> argList = new ArrayList<>(argListRaw);
+    siteModel = new SiteModel(TOOL_NAME, argList);
+    processProfile(siteModel.getExecutionConfiguration());
+    postProcessArgs(argList);
+    targetDevices = Set.copyOf(argList);
+    return this;
+  }
+
+  void execute() {
+    if (!Strings.isNullOrEmpty(config.feed_name)) {
+      validatePubSub(config.feed_name);
     }
     if (client == null) {
       validateReflector();
     }
-    targetDevices = Set.copyOf(listCopy);
+    checkState(client != null, "no validator client specified");
+    messageLoop();
   }
 
   /**
-   * Let's go.
-   *
-   * @param args Arguments for program execution
-   */
+     * Let's go.
+     *
+     * @param args Arguments for program execution
+     */
   public static void main(String[] args) {
+    ArrayList<String> argList = new ArrayList<>(List.of(args));
     try {
-      Validator validator = new Validator(Arrays.asList(args));
-      validator.messageLoop();
-    } catch (ExceptionMap processingException) {
-      System.exit(2);
+      new Validator().processArgs(argList).execute();
     } catch (Exception e) {
+      System.err.println("Exception in main: " + friendlyStackTrace(e));
       e.printStackTrace();
-      System.err.flush();
-      System.exit(-1);
+      System.exit(EXIT_CODE_ERROR);
     }
+
+    // Force exist because PubSub Subscriber in PubSubReflector does not shut down properly.
+    safeSleep(2000);
     System.exit(0);
   }
 
@@ -265,48 +296,40 @@ private List<String> parseArgs(List<String> argList) {
     } else {
       config = new ExecutionConfiguration();
     }
-    while (!argList.isEmpty()) {
-      String option = removeNextArg(argList);
-      try {
-        switch (option) {
-          case "-p":
-            setProjectId(removeNextArg(argList));
-            break;
-          case "-s":
-            setSiteDir(removeNextArg(argList));
-            break;
-          case "-a":
-            setSchemaSpec(removeNextArg(argList));
-            break;
-          case "-t":
-            validatePubSub(removeNextArg(argList));
-            break;
-          case "-f":
-            validateFilesOutput(removeNextArg(argList));
-            break;
-          case "-u":
-            forceUpgrade = true;
-            break;
-          case "-r":
-            validateMessageTrace(removeNextArg(argList));
-            break;
-          case "-n":
-            client = new NullPublisher();
-            break;
-          case "-w":
-            setMessageTraceDir(removeNextArg(argList));
-            break;
-          case "--":
-            // All remaining arguments remain in the return list.
-            return argList;
-          default:
-            throw new RuntimeException("Unknown cmdline option " + option);
+    return postProcessArgs(argList);
+  }
+
+  private List<String> postProcessArgs(List<String> argList) {
+    try {
+      while (!argList.isEmpty()) {
+        String option = removeNextArg(argList);
+        try {
+          switch (option) {
+            case "-p" -> setProjectId(removeNextArg(argList));
+            case "-s" -> setSiteDir(removeNextArg(argList));
+            case "-a" -> setSchemaSpec(removeNextArg(argList));
+            case "-t" -> validatePubSub(removeNextArg(argList));
+            case "-f" -> validateFilesOutput(removeNextArg(argList));
+            case "-u" -> forceUpgrade = true;
+            case "-r" -> validateMessageTrace(removeNextArg(argList));
+            case "-n" -> client = new NullPublisher();
+            case "-w" -> setMessageTraceDir(removeNextArg(argList));
+            case "--" -> {
+              // All remaining arguments remain in the return list.
+              return argList;
+            }
+            default -> throw new RuntimeException("Unknown cmdline option " + option);
+          }
+        } catch (MissingFormatArgumentException e) {
+          throw new RuntimeException("For command line option " + option, e);
         }
-      } catch (MissingFormatArgumentException e) {
-        throw new RuntimeException("For command line option " + option, e);
+      }
+      return argList;
+    } finally {
+      if (schemaMap == null) {
+        setSchemaSpec("schema");
       }
     }
-    return argList;
   }
 
   private void setProjectId(String projectId) {
@@ -324,7 +347,7 @@ private void validatePubSub(String pubSubCombo) {
     Preconditions.checkArgument(parts.length <= 2, "Too many parts in pubsub path " + pubSubCombo);
     String instName = parts[0];
     CloudIotManager cloudIotManager = new CloudIotManager(config.project_id,
-        new File(config.site_model), null, config.registry_suffix, IotProvider.GCP_NATIVE);
+        new File(config.site_model), null, config.registry_suffix, IotProvider.PUBSUB);
     String registryId = getRegistryId();
     String updateTopic = parts.length > 1 ? parts[1] : cloudIotManager.getUpdateTopic();
     client = new PubSubClient(config.project_id, registryId, instName, updateTopic);
@@ -336,16 +359,18 @@ private void validatePubSub(String pubSubCombo) {
   }
 
   private void processProfile(File profilePath) {
-    config = ConfigUtil.readExeConfig(profilePath);
-    String siteModel = ofNullable(config.site_model).orElse(BASE_DIR.getName());
+    ExecutionConfiguration exeConfig = readExeConfig(profilePath);
+    String siteModel = ofNullable(exeConfig.site_model).orElse(BASE_DIR.getName());
     File model = new File(siteModel);
     File adjustedPath = model.isAbsolute() ? model :
         new File(profilePath.getParentFile(), siteModel);
-    config.site_model = adjustedPath.getAbsolutePath();
-    setSiteDir(adjustedPath.getAbsolutePath());
-    if (!Strings.isNullOrEmpty(config.feed_name)) {
-      validatePubSub(config.feed_name);
-    }
+    exeConfig.site_model = adjustedPath.getAbsolutePath();
+    processProfile(exeConfig);
+  }
+
+  private void processProfile(ExecutionConfiguration exeConfig) {
+    config = exeConfig;
+    setSiteDir(exeConfig.site_model);
   }
 
   MessageReadingClient getMessageReadingClient() {
@@ -477,10 +502,14 @@ private void validateReflector() {
     String keyFile = new File(config.site_model, GCP_REFLECT_KEY_PKCS8).getAbsolutePath();
     outputLogger.info("Loading reflector key file from " + keyFile);
     config.key_file = keyFile;
-    client = new IotReflectorClient(config, TOOLS_FUNCTIONS_VERSION);
+    client = new IotReflectorClient(config, TOOLS_FUNCTIONS_VERSION, this::messageFilter);
     dataSinks.add(client);
   }
 
+  private boolean messageFilter(Envelope envelope) {
+    return true;
+  }
+
   void messageLoop() {
     if (client == null || client.getSubscriptionId() == null) {
       return;
@@ -550,7 +579,7 @@ private void validateMessage(Object msgObject, Map<String, String> attributes) {
       processedDevices.add(deviceId);
     }
 
-    if (!shouldConsiderMessage(attributes)) {
+    if (!shouldProcessMessage(attributes)) {
       return;
     }
 
@@ -589,7 +618,7 @@ private Instant getMessageInstant(Object msgObject, Map<String, String> attribut
     return getInstant(attributes.get(PUBLISH_TIME_KEY));
   }
 
-  private ReportingDevice validateMessageCore(Object message, Map<String, String> attributes) {
+  private ReportingDevice validateMessageCore(Object messageObj, Map<String, String> attributes) {
 
     String deviceId = attributes.get("deviceId");
     if (deviceId == null) {
@@ -597,21 +626,52 @@ private ReportingDevice validateMessageCore(Object message, Map<String, String>
     }
 
     ReportingDevice device = reportingDevices.computeIfAbsent(deviceId, ReportingDevice::new);
+    device.clearMessageEntries();
 
     try {
       String schemaName = messageSchema(attributes);
-      if (!device.markMessageType(schemaName, getNow())) {
+      boolean isString = messageObj instanceof String;
+
+      Map<String, Object> message = isString ? null : mapCast(messageObj);
+      validateTimestamp(device, message, attributes);
+
+      if (!device.processMessageSchema(schemaName, getMessageInstant(messageObj, attributes))) {
+        outputLogger.trace("Ignoring device %s/%s (too soon)", deviceId, schemaName);
         return null;
       }
 
-      outputLogger.info("Processing device #%d/%d: %s/%s", processedDevices.size(),
-          reportingDevices.size(), deviceId, schemaName);
+      writeDeviceOutDir(messageObj, attributes, deviceId, schemaName);
+
+      String subFolder = attributes.get(SUBFOLDER_PROPERTY_KEY);
+      boolean processSchema = !IGNORE_FOLDERS.contains(subFolder);
+
+      try {
+        if (processSchema && !schemaMap.containsKey(schemaName)) {
+          throw new IllegalArgumentException(format(SCHEMA_SKIP_FORMAT, schemaName, deviceId));
+        }
+      } catch (Exception e) {
+        outputLogger.error("Missing schema entry " + schemaName);
+        device.addError(e, attributes, Category.VALIDATION_DEVICE_RECEIVE);
+      }
+
+      if (isString) {
+        String detail = format("Raw string message for %s %s", deviceId, schemaName);
+        outputLogger.error(detail);
+        IllegalArgumentException exception = new IllegalArgumentException(detail);
+        device.addError(exception, attributes, Category.VALIDATION_DEVICE_RECEIVE);
+        return device;
+      }
+
+      outputLogger.info("Processing device %s/%s as #%d/%d", deviceId, schemaName,
+          processedDevices.size(), reportingDevices.size());
 
       if ("true".equals(attributes.get("wasBase64"))) {
         base64Devices.add(deviceId);
       }
 
-      writeDeviceOutDir(message, attributes, deviceId, schemaName);
+      if (processExceptions(attributes, deviceId, device, message)) {
+        return device;
+      }
       validateDeviceMessage(device, message, attributes);
 
       if (!device.hasErrors()) {
@@ -624,28 +684,12 @@ private ReportingDevice validateMessageCore(Object message, Map<String, String>
     return device;
   }
 
-  /**
-   * Validate a device message against the core schema.
-   */
-  public void validateDeviceMessage(ReportingDevice device, Object baseMsg,
-      Map<String, String> attributes) {
-    String deviceId = attributes.get("deviceId");
-    device.clearMessageEntries();
-    String schemaName = messageSchema(attributes);
-
-    if (baseMsg instanceof String) {
-      String message = format("Raw string message for %s %s", deviceId, schemaName);
-      outputLogger.error(message);
-      IllegalArgumentException exception = new IllegalArgumentException(message);
-      device.addError(exception, attributes, Category.VALIDATION_DEVICE_RECEIVE);
-      return;
-    }
-
-    Map<String, Object> message = mapCast(baseMsg);
+  private boolean processExceptions(Map<String, String> attributes, String deviceId,
+      ReportingDevice device, Map<String, Object> message) {
     if (message.get(EXCEPTION_KEY) instanceof Exception exception) {
       outputLogger.error("Pipeline exception " + deviceId + ": " + getExceptionMessage(exception));
       device.addError(exception, attributes, Category.VALIDATION_DEVICE_RECEIVE);
-      return;
+      return true;
     }
 
     if (message.containsKey(ERROR_KEY)) {
@@ -654,56 +698,19 @@ public void validateDeviceMessage(ReportingDevice device, Object baseMsg,
       IllegalArgumentException exception = new IllegalArgumentException(
           "Error in message pipeline: " + error);
       device.addError(exception, attributes, Category.VALIDATION_DEVICE_RECEIVE);
-      return;
+      return true;
     }
+    return false;
+  }
 
+  /**
+   * Validate a device message against the core schema.
+   */
+  public void validateDeviceMessage(ReportingDevice device, Map<String, Object> message,
+      Map<String, String> attributes) {
+    String schemaName = messageSchema(attributes);
     upgradeMessage(schemaName, message);
 
-    String timestampRaw = (String) message.get("timestamp");
-    Instant timestamp = ifNotNullGet(timestampRaw, JsonUtil::getInstant);
-    String publishRaw = attributes.get(PUBLISH_TIME_KEY);
-    Instant publishTime = ifNotNullGet(publishRaw, JsonUtil::getInstant);
-    try {
-      // TODO: Validate message contests to make sure state sub-blocks don't also have timestamp.
-
-      String subTypeRaw = ofNullable(attributes.get(SUBTYPE_PROPERTY_KEY))
-          .orElse(UNKNOWN_TYPE_DEFAULT);
-      boolean lastSeenValid = LAST_SEEN_SUBTYPES.contains(SubType.fromValue(subTypeRaw));
-      if (lastSeenValid) {
-        if (publishTime != null) {
-          device.updateLastSeen(Date.from(publishTime));
-        }
-        if (timestamp == null) {
-          throw new RuntimeException("Missing message timestamp");
-        }
-        if (publishTime != null) {
-          if (!timestampRaw.endsWith(TIMESTAMP_ZULU_SUFFIX)
-              && !timestampRaw.endsWith(TIMESTAMP_UTC_SUFFIX_1)
-              && !timestampRaw.endsWith(TIMESTAMP_UTC_SUFFIX_2)) {
-            throw new RuntimeException("Invalid timestamp timezone " + timestampRaw);
-          }
-          long between = Duration.between(publishTime, timestamp).getSeconds();
-          if (between > TIMESTAMP_JITTER_SEC || between < -TIMESTAMP_JITTER_SEC) {
-            throw new RuntimeException(format(
-                "Timestamp jitter %ds (%s to %s) exceeds %ds threshold",
-                between, publishRaw, timestampRaw, TIMESTAMP_JITTER_SEC));
-          }
-        }
-      }
-    } catch (Exception e) {
-      outputLogger.error("Timestamp validation error: " + friendlyStackTrace(e));
-      device.addError(e, attributes, Category.VALIDATION_DEVICE_CONTENT);
-    }
-
-    try {
-      if (!schemaMap.containsKey(schemaName)) {
-        throw new IllegalArgumentException(format(SCHEMA_SKIP_FORMAT, schemaName, deviceId));
-      }
-    } catch (Exception e) {
-      outputLogger.error("Missing schema entry " + schemaName);
-      device.addError(e, attributes, Category.VALIDATION_DEVICE_RECEIVE);
-    }
-
     try {
       validateMessage(schemaMap.get(ENVELOPE_SCHEMA_ID), (Object) attributes);
     } catch (Exception e) {
@@ -720,13 +727,14 @@ public void validateDeviceMessage(ReportingDevice device, Object baseMsg,
       }
     }
 
+    String deviceId = attributes.get("deviceId");
     if (expectedDevices == null || expectedDevices.isEmpty()) {
       // No devices configured, so don't consider check metadata or consider extra.
     } else if (expectedDevices.contains(deviceId)) {
       try {
         ifNotNullThen(CONTENT_VALIDATORS.get(schemaName), targetClass -> {
           Object messageObject = OBJECT_MAPPER.convertValue(message, targetClass);
-          device.validateMessageType(messageObject, JsonUtil.getDate(publishRaw), attributes);
+          device.validateMessageType(messageObject, attributes);
         });
       } catch (Exception e) {
         outputLogger.error("Error validating contents: " + friendlyStackTrace(e));
@@ -737,6 +745,47 @@ public void validateDeviceMessage(ReportingDevice device, Object baseMsg,
     }
   }
 
+  private void validateTimestamp(ReportingDevice device, Map<String, Object> message,
+      Map<String, String> attributes) {
+    String timestampRaw = ifNotNullGet(message, m -> (String) m.get("timestamp"));
+    Instant timestamp = ifNotNullGet(timestampRaw, JsonUtil::getInstant);
+    String publishRaw = attributes.get(PUBLISH_TIME_KEY);
+    Instant publishTime = ifNotNullGet(publishRaw, JsonUtil::getInstant);
+    try {
+      // TODO: Validate message contests to make sure state sub-blocks don't also have timestamp.
+
+      String subTypeRaw = ofNullable(attributes.get(SUBTYPE_PROPERTY_KEY))
+          .orElse(UNKNOWN_TYPE_DEFAULT);
+      boolean lastSeenValid = LAST_SEEN_SUBTYPES.contains(SubType.fromValue(subTypeRaw));
+      if (lastSeenValid) {
+        if (publishTime != null) {
+          device.updateLastSeen(Date.from(publishTime));
+        }
+        if (message != null && timestamp == null) {
+          throw new RuntimeException("Missing message timestamp");
+        }
+        if (timestampRaw != null
+            && !timestampRaw.endsWith(TIMESTAMP_ZULU_SUFFIX)
+            && !timestampRaw.endsWith(TIMESTAMP_UTC_SUFFIX_1)
+            && !timestampRaw.endsWith(TIMESTAMP_UTC_SUFFIX_2)) {
+          throw new RuntimeException("Invalid timestamp timezone " + timestampRaw);
+        }
+        if (publishTime != null && timestamp != null) {
+          long between = Duration.between(publishTime, timestamp).getSeconds();
+          if (between > TIMESTAMP_JITTER_SEC || between < -TIMESTAMP_JITTER_SEC) {
+            throw new RuntimeException(format(
+                "Timestamp skew %ds (%s to %s) exceeds %ds threshold",
+                between, publishRaw, timestampRaw, TIMESTAMP_JITTER_SEC));
+          }
+        }
+      }
+    } catch (Exception e) {
+      outputLogger.error(format("Timestamp validation error for %s: %s", device.getDeviceId(),
+          friendlyStackTrace(e)));
+      device.addError(e, attributes, Category.VALIDATION_DEVICE_CONTENT);
+    }
+  }
+
   private void sendValidationResult(Map<String, String> origAttributes,
       ReportingDevice reportingDevice, Date now) {
     try {
@@ -747,9 +796,8 @@ private void sendValidationResult(Map<String, String> origAttributes,
       event.sub_folder = subFolder;
       String subType = origAttributes.get(SUBTYPE_PROPERTY_KEY);
       event.sub_type = ofNullable(subType).orElse(UNKNOWN_TYPE_DEFAULT);
-      event.status = ReportingDevice.getSummaryEntry(reportingDevice.getMessageEntries());
-      String prefix = format("%s:", typeFolderPairKey(subType, subFolder));
-      event.errors = reportingDevice.getErrors(now, prefix);
+      event.errors = reportingDevice.getMessageEntries();
+      event.status = ReportingDevice.getSummaryEntry(event.errors);
       if (POINTSET_SUBFOLDER.equals(subFolder)) {
         PointsetSummary pointsSummary = new PointsetSummary();
         pointsSummary.missing = arrayIfNotNull(reportingDevice.getMissingPoints());
@@ -799,7 +847,7 @@ private void writeMessageCapture(Object message, Map<String, String> attributes)
     }
   }
 
-  private boolean shouldConsiderMessage(Map<String, String> attributes) {
+  private boolean shouldProcessMessage(Map<String, String> attributes) {
     String registryId = attributes.get(DEVICE_REGISTRY_ID_KEY);
 
     if (!registryId.equals(getRegistryId())) {
@@ -821,10 +869,10 @@ private boolean shouldConsiderMessage(Map<String, String> attributes) {
     String subType = attributes.get(SUBTYPE_PROPERTY_KEY);
     String subFolder = attributes.get(SUBFOLDER_PROPERTY_KEY);
     String category = attributes.get("category");
-    boolean isInteresting = subType == null
+    boolean process = subType == null
         || INTERESTING_TYPES.contains(subType)
         || SubFolder.UPDATE.value().equals(subFolder);
-    return !CONFIG_CATEGORY.equals(category) && isInteresting;
+    return process && !CONFIG_CATEGORY.equals(category);
   }
 
   private void writeDeviceOutDir(Object message, Map<String, String> attributes, String deviceId,
diff --git a/validator/src/main/java/com/google/udmi/util/PubSubReflector.java b/validator/src/main/java/com/google/udmi/util/PubSubReflector.java
index 41443c395e..f6f0d9f660 100644
--- a/validator/src/main/java/com/google/udmi/util/PubSubReflector.java
+++ b/validator/src/main/java/com/google/udmi/util/PubSubReflector.java
@@ -7,9 +7,10 @@
 import static com.google.udmi.util.Common.PUBLISH_TIME_KEY;
 import static com.google.udmi.util.Common.SOURCE_KEY;
 import static com.google.udmi.util.Common.SOURCE_SEPARATOR;
+import static com.google.udmi.util.Common.SOURCE_SEPARATOR_REGEX;
 import static com.google.udmi.util.Common.SUBFOLDER_PROPERTY_KEY;
 import static com.google.udmi.util.Common.getNamespacePrefix;
-import static com.google.udmi.util.GeneralUtils.ifNullThen;
+import static com.google.udmi.util.GeneralUtils.ifNotNullThen;
 import static com.google.udmi.util.JsonUtil.isoConvert;
 import static com.google.udmi.util.JsonUtil.stringify;
 import static com.google.udmi.util.JsonUtil.stringifyTerse;
@@ -312,13 +313,21 @@ public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
         String topic = format("/devices/%s/%s%s", attributes.get(DEVICE_ID_KEY),
             attributes.get(CATEGORY_PROPERTY_KEY),
             suffix);
-        String messageSource = attributes.get(SOURCE_KEY);
-        String userMatch = SOURCE_SEPARATOR + userName;
-        if (!userMatch.equals(messageSource)) {
-          ifNullThen(messageSource, () -> System.err.println(
-              "Discarding message with null source: " + stringifyTerse(attributes)));
+        String messageSource = attributes.remove(SOURCE_KEY);
+        if (messageSource == null) {
+          System.err.println("Discarding message with null source: " + stringifyTerse(attributes));
           return;
         }
+        Object dstSource = messageBundle.message.remove(SOURCE_KEY);
+        String[] source = messageSource.split(SOURCE_SEPARATOR_REGEX, 3);
+        if (source.length == 1) {
+          attributes.put(SOURCE_KEY, source[0]);
+        } else if (source.length == 2 && source[0].isEmpty()) {
+          topic += messageSource;
+          ifNotNullThen(dstSource, () -> messageBundle.message.put(SOURCE_KEY, source[1]));
+        } else {
+          System.err.println("Discarding message with malformed source: " + messageSource);
+        }
         messageHandler.accept(topic, stringify(messageBundle.message));
       } catch (Exception e) {
         errorHandler.accept(e);
diff --git a/validator/src/test/java/com/google/daq/mqtt/validator/PlaybackTest.java b/validator/src/test/java/com/google/daq/mqtt/validator/PlaybackTest.java
index cd60609ca2..1f49bd7f73 100644
--- a/validator/src/test/java/com/google/daq/mqtt/validator/PlaybackTest.java
+++ b/validator/src/test/java/com/google/daq/mqtt/validator/PlaybackTest.java
@@ -24,6 +24,7 @@
 public class PlaybackTest extends TestBase {
 
   public static final String SIMPLE_TRACE_DIR = "../tests/traces/simple";
+  public static final String LENGTHY_TRACE_DIR = "../tests/traces/lengthy";
   private static final List<String> TRACE_DEVICES = List.of("--", "AHU-22", "SNS-4", "XXX", "YYY");
 
   @Test
@@ -72,6 +73,25 @@ private List<ValidationEvents> reports(List<OutputBundle> outputMessages, String
         .collect(Collectors.toList());
   }
 
+  @Test
+  public void notMissingDevice() {
+    MessageReadingClient client = validateTrace(LENGTHY_TRACE_DIR);
+    assertEquals("trace message count", 3, client.messageCount);
+    List<OutputBundle> outputMessages = client.getOutputMessages();
+    OutputBundle lastBundle = outputMessages.get(outputMessages.size() - 1);
+    ValidationState finalReport = asValidationState(lastBundle.message);
+    try {
+      assertEquals("correct devices", 1, finalReport.summary.correct_devices.size());
+      assertEquals("extra devices", 0, finalReport.summary.extra_devices.size());
+      assertEquals("missing devices", 3, finalReport.summary.missing_devices.size());
+      assertEquals("error devices", 0, finalReport.summary.error_devices.size());
+      assertEquals("device summaries", 1, finalReport.devices.size());
+    } catch (Throwable e) {
+      outputMessages.forEach(message -> System.err.println(JsonUtil.stringify(message)));
+      throw e;
+    }
+  }
+
   @Test
   public void deviceArgs() {
     MessageReadingClient client = validateTrace(SIMPLE_TRACE_DIR, TRACE_DEVICES);