ci: use the shared branch-aware CodeQL worker

Co-Authored-By: Codex <noreply@openai.com>

.circleci/config.yml

@@ -1,65 +1,14 @@
 version: 2.1
-
+setup: true
 orbs:
-  qqq-orb: kingsrook/qqq-orb@0.6.9
-
-commands:
-  configure-package-install:
-    steps:
-      - run:
-          name: Keep package installation unattended
-          command: |
-            # These build machines are discarded after the job.
-            sudo install -d /etc/needrestart/conf.d
-            printf '%s\n' '$nrconf{restart} = "l";' | sudo tee /etc/needrestart/conf.d/99-qqq-ci.conf >/dev/null
-
+  qqq-orb: kingsrook/qqq-orb@0.7.12
 workflows:
-
-  test_only:
+  setup:
     jobs:
-      - qqq-orb/mvn_test_only:
-          pre-steps:
-            - configure-package-install
-          context: [ qqq-maven-registry-credentials ]
-          filters:
-            branches:
-              ignore: /(develop|release/.*|integration.*)/
-            tags:
-              only: []
-
-  publish_snapshot:
-    jobs:
-      - qqq-orb/mvn_publish:
-          pre-steps:
-            - configure-package-install
-          context: [ qqq-maven-registry-credentials ]
-          branch_type: snapshot
-          filters:
-            branches:
-              only:
-                - develop
-
-  publish_release_candidate:
-    jobs:
-      - qqq-orb/mvn_publish:
-          pre-steps:
-            - configure-package-install
-          context: [ qqq-maven-registry-credentials ]
-          branch_type: release_candidate
-          filters:
-            branches:
-              only:
-                - /release\/.*/
-
-  publish_release:
-    jobs:
-      - qqq-orb/mvn_publish:
-          pre-steps:
-            - configure-package-install
-          context: [ qqq-maven-registry-credentials ]
-          branch_type: release
-          filters:
-            branches:
-              ignore: /.*/
-            tags:
-              only: [ /^v\d+\.\d+\.\d+$/ ]
+    - qqq-orb/generate_pipeline:
+        orb_version: 0.7.12
+        filters:
+          branches:
+            ignore: automation/qbit-rc-pending
+          tags:
+            only: /^v\d+\.\d+\.\d+$/

.github/workflows/codeql.yml

@@ -0,0 +1,23 @@
+name: CodeQL
+'on':
+  push:
+    branches:
+    - main
+    - develop
+    - release/**
+    - hotfix/**
+  pull_request: {}
+  schedule:
+  - cron: 41 16 * * 0
+  workflow_dispatch: {}
+permissions:
+  contents: read
+  actions: read
+  security-events: write
+  packages: read
+concurrency:
+  group: codeql-${{ github.event.pull_request.number || github.ref }}
+  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+jobs:
+  analyze:
+    uses: QRun-IO/qqq-orb/.github/workflows/qbit-codeql.yml@c288a8864fe9054f68410e1d78f5863aea4c4d22

.github/workflows/release-rc.yml

@@ -1,25 +1,74 @@
-# Cut the next release candidate (QRun-IO/qqq docs/QRUN-QBIT-REPO-STANDARD.md, C6):
-# run this workflow by hand, or merge a develop PR labelled `release:rc`. It merges develop into
-# the release line and the release/* push runs CircleCI publish_release_candidate.
 name: Release candidate
-
-on:
+'on':
   workflow_dispatch:
     inputs:
       dry_run:
-        description: Print the plan without merging or creating branches
+        description: Plan a cut without changing release state
         type: boolean
         default: false
+      finalize:
+        description: Retry finalization of an existing cut; do not cut another RC
+        type: boolean
+        default: false
+      recover:
+        description: Recover the same published RC after a deployment timeout; never
+          reupload
+        type: boolean
+        default: false
+      circle_build:
+        description: Original Circle publisher build number (recovery only)
+        type: string
+        default: ''
+      merge_sha:
+        description: Existing cut merge_sha (finalize retry only)
+        type: string
+        default: ''
+      bump_sha:
+        description: Existing cut bump_sha (finalize retry only)
+        type: string
+        default: ''
+      branch:
+        description: Existing cut branch (finalize retry only)
+        type: string
+        default: ''
+      version:
+        description: Existing cut version (finalize retry only)
+        type: string
+        default: ''
   pull_request:
-    types: [closed]
-    branches: [develop]
-
+    types:
+    - closed
+    branches:
+    - develop
+  repository_dispatch:
+    types:
+    - qbit-rc-published
 permissions:
   contents: write
-
+  actions: read
+  checks: read
 jobs:
   release-rc:
-    if: github.event_name == 'workflow_dispatch' || (github.event.pull_request.merged == true && contains(github.event.pull_request.labels.*.name, 'release:rc'))
-    uses: QRun-IO/.github/.github/workflows/qbit-release-rc.yml@main
+    if: github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch'
+      || (github.event_name == 'pull_request' && github.event.pull_request.merged
+      == true && github.event.pull_request.base.ref == 'develop' && contains(github.event.pull_request.labels.*.name,
+      'release:rc'))
+    uses: QRun-IO/qqq-orb/.github/workflows/qbit-release-rc.yml@cc30f599e63c8e41629c99d23395bac1adb8bc44
     with:
-      dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run }}
+      dry_run: ${{ github.event_name == 'workflow_dispatch' && inputs.dry_run || false
+        }}
+      finalize: ${{ github.event_name == 'repository_dispatch' || (github.event_name
+        == 'workflow_dispatch' && inputs.finalize) }}
+      recover: ${{ github.event_name == 'workflow_dispatch' && inputs.recover || false
+        }}
+      circle_build: ${{ github.event_name == 'workflow_dispatch' && inputs.circle_build
+        || '' }}
+      merge_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.merge_sha
+        || inputs.merge_sha || '' }}
+      bump_sha: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.bump_sha
+        || inputs.bump_sha || '' }}
+      branch: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.branch
+        || inputs.branch || '' }}
+      version: ${{ github.event_name == 'repository_dispatch' && github.event.client_payload.version
+        || inputs.version || '' }}
+    secrets: inherit

.qqq-ci.yml

@@ -0,0 +1,16 @@
+profile: qbit
+orb_version: 0.7.12
+default_branch: develop
+java_version: '21'
+browsers: true
+api_checks: false
+docs_only: false
+release_on_main: false
+release_tag_policy: stable
+test_branch_policy: non-release-no-integration
+publish:
+  snapshot: true
+  release_candidate: true
+  release: true
+  asciidoc: false
+release_completion: true

CHANGELOG.md

@@ -4,6 +4,16 @@ All notable changes are documented here, following [Keep a Changelog](https://ke
 
 ## [Unreleased]
 
+### Fixed
+
+- `worm-audit:13` failed with `permission denied for function worm_audit_purge` when the migrating role owns the function but had revoked its own EXECUTE (a runtime role that owns the schema, as on CNPG; PostgreSQL refuses `CREATE OR REPLACE` then). The changeset now grants EXECUTE back to the migrating role only if it lacks it, replaces the function, and revokes it again in the same transaction. Found by Website-Backend's CNPG-shaped changelog test.
+- `worm-audit:13` had a comment line starting with `-- changeset`, which Liquibase formatted-SQL parsing rejects as a malformed directive. The comment is reworded, and `CanonicalAndCryptoTest` now requires every `-- changeset` line to be a real directive.
+- Chain writer database load. Each 30-second tick re-read the whole late-detail lookback (24 h of `audit_detail`) and probed the GIN array index once per detail (18.4 s and 1.78M buffer hits per tick on a 120k-detail production database), and re-scanned every settled audit header for the unchained anti-join. Coverage now lives in `worm_audit_chain_detail` (`worm-audit:13`): one insert-only row per covered detail, written by the chain trigger in the same statement as its link and backfilled from the existing chain, so the late-detail check is a primary-key probe. Each tick also reads only ids above per-JVM sweep floors (rows settled since the last complete sweep); after a restart and once an hour a tick sweeps everything, as before. Link hash inputs, `detail_ids`, the verifier and exports are unchanged.
+- The chain trigger's overlap check (`EXISTS ... && ...`) ran as a cached generic plan that sequentially scanned all of `worm_audit_chain` for every inserted link (55k sequential scans, 1.2B tuples read in production, about 80 ms per link and growing). The trigger now inserts the coverage rows and lets their primary key reject a detail covered twice.
+- Verification counts detail coverage by aggregating the chain's own detail lists once instead of one index probe per detail, and breaks when the coverage table disagrees with the chain.
+
+- Startup on hosts whose default authentication requires a token (OAuth2, Auth0): `produce()` read the settings singleton under a plain tokenless `QSession`, which failed session validation (`JWTDecodeException: The token is null`) and stopped the host. The startup read now runs as the core `QSystemUserSession`, still honoring the persisted schedules. `TokenRequiredHostTest` runs production under a token-required test authentication module.
+
 ### Added
 
 - Foreign keys (`worm-audit:11-12`), `ON DELETE RESTRICT` only: chain link -> core audit header, sealed export -> its first and last checkpoints, export verification -> its export, with their indexes. Link references (ranges, break link, purge anchor) stay unconstrained because the retention purge deletes the links they outlive. Pre-existing orphans are copied to `qqq_fk_quarantine` (`KEPT`), never modified, and leave that constraint `NOT VALID`. Re-apply the rendered DDL (idempotent).

README.md

@@ -137,6 +137,12 @@ Open `/app/worm_audit` (or `/app/admin/worm_audit` with `parentAppName="admin"`)
 The full-width `wormAuditHome` HTML widget uses the standard `/qqq/v1/widget` API.
 Hosts can add `wormAuditHealthCard` to their own dashboard. Both widgets and the app
 require read permission on at least one WORM table, using the host's table permission rules.
+To embed WORM in an existing host app, set `withIncludeApp(false)` and configure
+`appName`, `appLabel` and optional `parentAppName` for that host app. The host owns
+its navigation, home widget placement and app help; the qbit does not create or replace
+the app or fill its help. Both widgets and all table help, permissions, guards,
+processes and schedules remain registered. The health card links to the configured
+host app. Standalone app creation remains enabled by default.
 The aggregate home shows only identifiers, counts, times, shortened hashes and reasons;
 it never renders audit messages or field values. Record links retain the target table's permissions.
 

UPGRADE.md

@@ -1,5 +1,9 @@
 # Upgrading qbit-worm-audit
 
+## Detail coverage table (0.1.0-RC.4)
+
+Apply changeset `worm-audit:13` from the rendered DDL (or your copy of it) before or with the new jar. It takes the chain advisory lock, creates `worm_audit_chain_detail` with its append-only guards and grants, backfills it from the existing chain, and replaces `worm_audit_reject()`, `worm_audit_detail_coverage()` and `worm_audit_purge()`. The jar requires the table: its chain writer reads it every tick. The GIN index `worm_audit_chain_detail_coverage` is no longer used but is kept so an RC.3 writer still running during a rolling deploy keeps an indexed probe; a later changeset drops it. Memory and H2 hosts need no migration beyond creating the new table (the writer fills it).
+
 ## Table help bundle (unreleased)
 
 Nothing to do. The qbit applies `help/qbit-worm-audit.yaml` at the end of `produce()`, filling only help slots the host has not set. No schema or seed changes: the settings row is still seeded by the database script (changeset `worm-audit:10`) and nothing else is seeded.

docs/DESIGN.md

@@ -192,7 +192,13 @@ chain would need the canonical form in SQL. The design instead uses **one chaine
   `kind = LATE_DETAIL`, covering exactly those detail ids. Verification flags any detail not
   covered by exactly one link.
 - Unique constraints back the lock: `UNIQUE (audit_id) WHERE kind = 'AUDIT'`, and the detail
-  coverage.
+  coverage: `worm_audit_chain_detail (detail_id PRIMARY KEY, chain_id)`, one append-only row per
+  covered detail, written by the chain insert trigger in the same statement as the link. The
+  writer finds uncovered details with a primary-key probe of it; verification still counts
+  coverage from the links' own `detail_ids` and breaks if the table disagrees.
+- Each tick reads only rows above per-JVM sweep floors: the highest audit and detail ids seen at
+  least (settle horizon + 5 minutes) before the last complete sweep. After a restart and once an
+  hour a tick ignores the floors, so a row whose transaction commits even later is still chained.
 
 Why not per-partition chains: they make each partition's chain shorter but multiply heads,
 checkpoints and verification state, and they still need a writer per partition. One chainer

docs/ddl/worm-audit-postgres.sql

@@ -385,3 +385,172 @@ BEGIN
    END LOOP;
 END;
 $$;
+
+-- changeset worm-audit:13 splitStatements:false
+-- comment: Normalized detail coverage. One insert-only row per audit_detail id covered by a chain link, written by the coverage trigger in the same statement as the link, so the chain writer probes a primary key instead of the GIN array index. Append-only like the chain, with no foreign keys (the owner cannot UPDATE WORM tables, so FK checks would be denied). The advisory lock keeps a running chainer from appending between the backfill and the new trigger body.
+DO $$
+BEGIN
+   PERFORM pg_advisory_xact_lock(86753090421);
+END;
+$$;
+CREATE TABLE IF NOT EXISTS public.worm_audit_chain_detail (
+   detail_id bigint PRIMARY KEY,
+   chain_id bigint NOT NULL,
+   CHECK (detail_id >= 0),
+   CHECK (chain_id > 0)
+);
+CREATE INDEX IF NOT EXISTS worm_audit_chain_detail_chain ON public.worm_audit_chain_detail(chain_id);
+-- The first link wins if a tampered chain covers a detail twice; the verifier still reports the duplicate from the chain itself.
+INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
+   SELECT x.id::bigint, c.id FROM public.worm_audit_chain c CROSS JOIN LATERAL unnest(string_to_array(c.detail_ids, ',')) AS x(id)
+   WHERE c.detail_ids <> '' ORDER BY c.id
+   ON CONFLICT (detail_id) DO NOTHING;
+
+CREATE OR REPLACE FUNCTION public.worm_audit_reject() RETURNS trigger
+LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
+DECLARE
+   old_time timestamptz;
+   function_owner name;
+   stack text;
+   row_identity text := 'all';
+BEGIN
+   IF TG_OP = 'DELETE' AND current_setting('worm_audit.purging', true) = 'on' THEN
+      SELECT pg_get_userbyid(proowner) INTO function_owner FROM pg_proc
+         WHERE oid = 'public.worm_audit_purge(timestamptz)'::regprocedure;
+      GET DIAGNOSTICS stack = PG_CONTEXT;
+      IF current_user = function_owner AND position('function worm_audit_purge(timestamp with time zone)' in stack) > 0 THEN
+         IF TG_RELID = '"audit"'::regclass THEN
+            old_time := OLD.timestamp;
+         ELSIF TG_RELID = '"audit_detail"'::regclass THEN
+            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
+         ELSIF TG_RELID = 'public.worm_audit_chain'::regclass THEN
+            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
+         ELSIF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
+            SELECT a.timestamp INTO old_time FROM public.worm_audit_chain c JOIN "audit" a ON a.id = c.audit_id WHERE c.id = OLD.chain_id;
+         END IF;
+         IF old_time < now() - interval '6 years' THEN
+            RETURN OLD;
+         END IF;
+      END IF;
+   END IF;
+   IF TG_LEVEL = 'ROW' THEN
+      -- Record fields resolve per statement, so each table's key is read in its own branch.
+      IF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
+         row_identity := OLD.detail_id::text;
+      ELSE
+         row_identity := OLD.id::text;
+      END IF;
+   END IF;
+   RAISE EXCEPTION USING ERRCODE = 'P0001',
+      MESSAGE = format('worm_audit: audit rows are write-once; %s of "%s" is not allowed (id=%s). Expired rows are removed only by worm_audit_purge().',
+         TG_OP, TG_TABLE_NAME, row_identity),
+      HINT = 'see qbit-worm-audit docs/DESIGN.md section 4';
+END;
+$fn$;
+
+CREATE OR REPLACE FUNCTION public.worm_audit_detail_coverage() RETURNS trigger
+LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp AS $fn$
+BEGIN
+   -- Serialize even direct SQL appends, then record coverage in the same statement as the link.
+   -- The primary key rejects a detail covered twice, within this link or by an earlier one. The
+   -- previous EXISTS overlap check ran as a cached generic plan that chose a sequential scan of
+   -- the whole chain for every inserted link.
+   PERFORM pg_advisory_xact_lock(86753090421);
+   IF NEW.detail_ids = '' THEN
+      RETURN NEW;
+   END IF;
+   BEGIN
+      INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
+         SELECT d, NEW.id FROM unnest(string_to_array(NEW.detail_ids, ',')::bigint[]) AS d;
+   EXCEPTION WHEN unique_violation THEN
+      RAISE EXCEPTION USING ERRCODE = '23505', MESSAGE = 'worm_audit: audit detail is already covered by the chain';
+   END;
+   RETURN NEW;
+END;
+$fn$;
+
+-- The array index worm_audit_chain_detail_coverage is no longer used by this release. It stays until
+-- a later migration drops it, so a previous-release writer still running during a rolling deploy keeps its indexed probe.
+
+DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_chain_detail;
+CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_chain_detail
+   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
+DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_chain_detail;
+CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_chain_detail
+   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
+REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_chain_detail FROM "app";
+GRANT SELECT, INSERT ON public.worm_audit_chain_detail TO "app";
+
+-- Same function as worm-audit:5, except that coverage rows are locked and purged before their links.
+-- The migrating role may have revoked its own EXECUTE (a runtime role that owns the schema, as on CNPG);
+-- PostgreSQL then refuses CREATE OR REPLACE. Grant it back for this statement only, then restore.
+DO $$
+BEGIN
+   IF NOT has_function_privilege(current_user, 'public.worm_audit_purge(timestamptz)', 'EXECUTE') THEN
+      EXECUTE format('GRANT EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) TO %I', current_user);
+      PERFORM set_config('worm_audit.purge_regrant', 'on', true);
+   END IF;
+END;
+$$;
+CREATE OR REPLACE FUNCTION public.worm_audit_purge(p_cutoff timestamptz) RETURNS void
+LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
+DECLARE
+   boundary bigint;
+   smaller bigint;
+   anchor text;
+   audits_count bigint;
+   details_count bigint;
+   links_count bigint;
+BEGIN
+   IF p_cutoff IS NULL OR p_cutoff > now() - interval '6 years' THEN
+      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
+         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'Cutoff is within retention or null');
+      RETURN;
+   END IF;
+   -- Same lock as the Java chainer; table locks also exclude concurrent audit inserts.
+   PERFORM pg_advisory_xact_lock(86753090421);
+   LOCK TABLE "audit", "audit_detail", public.worm_audit_chain, public.worm_audit_chain_detail IN SHARE ROW EXCLUSIVE MODE;
+   SELECT coalesce(min(c.id) - 1, (SELECT max(id) FROM public.worm_audit_chain)) INTO boundary
+      FROM public.worm_audit_chain c LEFT JOIN "audit" a ON a.id = c.audit_id
+      WHERE a.id IS NULL OR a.timestamp IS NULL OR a.timestamp >= p_cutoff;
+   -- Shrink to a prefix closed over every audit's links AND all its stored details.
+   LOOP
+      SELECT min(c.id) - 1 INTO smaller FROM public.worm_audit_chain c
+         WHERE c.id <= boundary AND (
+            EXISTS (SELECT FROM public.worm_audit_chain later WHERE later.audit_id = c.audit_id AND later.id > boundary)
+            OR EXISTS (SELECT FROM "audit_detail" d WHERE d.audit_id = c.audit_id AND NOT EXISTS (
+               SELECT FROM public.worm_audit_chain covered WHERE covered.audit_id = c.audit_id
+                  AND d.id::text = ANY(string_to_array(covered.detail_ids, ',')))));
+      EXIT WHEN smaller IS NULL;
+      boundary := smaller;
+   END LOOP;
+   SELECT link_hash INTO anchor FROM public.worm_audit_chain WHERE id = boundary;
+   IF anchor IS NULL THEN
+      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
+         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'No complete expired chain prefix');
+      RETURN;
+   END IF;
+   PERFORM set_config('worm_audit.purging', 'on', true);
+   DELETE FROM "audit_detail" d WHERE EXISTS
+      (SELECT FROM public.worm_audit_chain c WHERE c.audit_id = d.audit_id AND c.id <= boundary);
+   GET DIAGNOSTICS details_count = ROW_COUNT;
+   -- Coverage rows go before their links: the guard finds their age through the link.
+   DELETE FROM public.worm_audit_chain_detail WHERE chain_id <= boundary;
+   -- Keep ids locally because links must be deleted before their audit headers.
+   WITH removed AS (DELETE FROM public.worm_audit_chain WHERE id <= boundary RETURNING audit_id),
+      removed_audits AS (DELETE FROM "audit" WHERE id IN (SELECT audit_id FROM removed) RETURNING id)
+      SELECT (SELECT count(*) FROM removed), (SELECT count(*) FROM removed_audits) INTO links_count, audits_count;
+   PERFORM set_config('worm_audit.purging', 'off', true);
+   INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome,
+      audits_deleted, details_deleted, links_deleted, anchor_link_id, anchor_hash)
+      VALUES(now(), session_user, p_cutoff, '6 years', 'PURGED', audits_count, details_count, links_count, boundary, anchor);
+END;
+$fn$;
+DO $$
+BEGIN
+   IF current_setting('worm_audit.purge_regrant', true) = 'on' THEN
+      EXECUTE format('REVOKE EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) FROM %I', current_user);
+      PERFORM set_config('worm_audit.purge_regrant', 'off', true);
+   END IF;
+END;
+$$;

src/main/java/com/kingsrook/qbits/wormaudit/WormAuditQBitConfig.java

@@ -48,6 +48,7 @@ public class WormAuditQBitConfig implements QBitConfig
    private String exportLockMode = "GOVERNANCE";
    private int exportRetentionDays = 30;
    private final WormAuditSettings settings = new WormAuditSettings(this);
+   private boolean includeApp = true;
    private String appName = "worm_audit";
    private String appLabel = "WORM Audit";
    private String parentAppName;
@@ -131,7 +132,26 @@ public class WormAuditQBitConfig implements QBitConfig
 
 
    /*******************************************************************************
-    ** Read the app name.
+    ** Whether to create the standalone app. Widgets remain available when false.
+    *******************************************************************************/
+   public boolean getIncludeApp()
+   {
+      return (includeApp);
+   }
+
+
+   /*******************************************************************************
+    ** Disable standalone app creation when the host owns the Audit app and help.
+    *******************************************************************************/
+   public WormAuditQBitConfig withIncludeApp(boolean includeApp)
+   {
+      this.includeApp = includeApp;
+      return (this);
+   }
+
+
+   /*******************************************************************************
+    ** Read the app name (the host-owned destination when includeApp is false).
     *******************************************************************************/
    public String getAppName()
    {

src/main/java/com/kingsrook/qbits/wormaudit/WormAuditQBitProducer.java

@@ -44,6 +44,7 @@ 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.scheduleing.quartz.QuartzSchedulerMetaData;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.QTableMetaData;
+import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
 
 
 /*******************************************************************************
@@ -192,9 +193,14 @@ public class WormAuditQBitProducer implements QBitMetaDataProducer<WormAuditQBit
       var previousTransaction = QContext.getQBackendTransaction();
       try
       {
-         // Production is still assembling the host metadata. Do not validate it mid-build.
+         ////////////////////////////////////////////////////////////////////////////
+         // Production is still assembling the host metadata. Do not validate it   //
+         // mid-build. The startup settings read is system work: run it as the     //
+         // core system user. A plain tokenless QSession fails session validation  //
+         // under token-based host authentication (OAuth2, Auth0).                 //
+         ////////////////////////////////////////////////////////////////////////////
          QContext.setQInstance(qInstance);
-         QContext.setQSession(new com.kingsrook.qqq.backend.core.model.session.QSession());
+         QContext.setQSession(new QSystemUserSession(qInstance));
          QContext.setTransaction(null);
          addProcesses(output, qBit);
       }
@@ -208,7 +214,10 @@ public class WormAuditQBitProducer implements QBitMetaDataProducer<WormAuditQBit
       {
          qInstance.addScheduler(new QuartzSchedulerMetaData().withName(qBitConfig.getSchedulerName()));
       }
-      output.add(WormAuditApp.app(qBitConfig, producedTables));
+      if(qBitConfig.getIncludeApp())
+      {
+         output.add(WormAuditApp.app(qBitConfig, producedTables));
+      }
       for(QWidgetMetaData widget : WormAuditApp.widgets())
       {
          output.add(widget);

src/main/java/com/kingsrook/qbits/wormaudit/chain/AuditStore.java

@@ -68,9 +68,21 @@ import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSTableBackendDe
  *******************************************************************************/
 public class AuditStore
 {
+   /*******************************************************************************
+    ** Every (detail id, link id) pair the chain's own detail lists cover.
+    *******************************************************************************/
+   private static final String CHAIN_PAIRS = "SELECT x.id::bigint AS detail_id, c.id AS chain_id FROM worm_audit_chain c"
+      + " CROSS JOIN LATERAL unnest(string_to_array(c.detail_ids, ',')) AS x(id) WHERE c.detail_ids <> ''";
+
+   /*******************************************************************************
+    ** Chain coverage counted per detail id, aggregated in one pass.
+    *******************************************************************************/
+   private static final String CHAIN_COVERAGE = "SELECT detail_id AS id, count(*) AS n FROM (" + CHAIN_PAIRS + ") pairs GROUP BY detail_id";
+
    private final QBackendTransaction transaction;
    private final Map<String, String> names = new LinkedHashMap<>(16, 0.75F, true);
    private Boolean timestampWithZone;
+   private Boolean postgres;
 
 
    /*******************************************************************************
@@ -268,12 +280,23 @@ public class AuditStore
     ** memory uses QQQ queries and the same ordering and cutoff semantics.
     *******************************************************************************/
    public List<QRecord> unchained(Instant cutoff, int limit) throws QException
+   {
+      return (unchained(cutoff, limit, 0L));
+   }
+
+
+   /*******************************************************************************
+    ** Settled unchained headers above a swept audit id floor (RDBMS only); every
+    ** settled header at or below the floor is already known to be chained.
+    *******************************************************************************/
+   public List<QRecord> unchained(Instant cutoff, int limit, long floor) throws QException
    {
       if(isRdbms())
       {
-         String sql = "SELECT a." + column("audit", "id") + " FROM " + physical("audit") + " a WHERE a." + column("audit", "timestamp")
-            + " < ?" + unchainedPredicate() + " ORDER BY a." + column("audit", "timestamp") + ", a." + column("audit", "id") + " LIMIT " + limit;
-         Map<Long, QRecord> rows = index(in("audit", "id", sqlIds(sql, cutoff)));
+         String sql = "SELECT a." + column("audit", "id") + " FROM " + physical("audit") + " a WHERE a." + column("audit", "id") + " > ? AND a."
+            + column("audit", "timestamp") + " < ?" + unchainedPredicate() + " ORDER BY a." + column("audit", "timestamp") + ", a." + column("audit", "id")
+            + " LIMIT " + limit;
+         Map<Long, QRecord> rows = index(in("audit", "id", sqlIds(sql, floor, cutoff)));
          return (rows.values().stream().sorted(Comparator.comparing((QRecord r) -> r.getValueInstant("timestamp"))
             .thenComparing(r -> r.getValueLong("id"))).toList());
       }
@@ -380,15 +403,26 @@ public class AuditStore
     ** Poll uncovered details in a bounded time window; memory retains its simple fallback.
     *******************************************************************************/
    public List<QRecord> lateDetails(Instant cutoff, int hours, int limit) throws QException
+   {
+      return (lateDetails(cutoff, hours, limit, 0L));
+   }
+
+
+   /*******************************************************************************
+    ** Uncovered details above a swept detail id floor (RDBMS only). Coverage is a
+    ** primary-key probe of the normalized coverage table, never an array scan.
+    *******************************************************************************/
+   public List<QRecord> lateDetails(Instant cutoff, int hours, int limit, long floor) throws QException
    {
       if(isRdbms())
       {
-         String sql = "SELECT d." + column("auditDetail", "id") + " FROM " + physical("auditDetail") + " d JOIN " + physical("audit")
+         String detailId = "d." + column("auditDetail", "id");
+         String sql = "SELECT " + detailId + " FROM " + physical("auditDetail") + " d JOIN " + physical("audit")
             + " a ON a." + column("audit", "id") + "=d." + column("auditDetail", "auditId")
-            + " WHERE a." + column("audit", "timestamp") + " < ? AND a." + column("audit", "timestamp") + " >= ?"
+            + " WHERE " + detailId + " > ? AND a." + column("audit", "timestamp") + " < ? AND a." + column("audit", "timestamp") + " >= ?"
             + " AND EXISTS (SELECT 1 FROM worm_audit_chain h WHERE h.kind='AUDIT' AND h.audit_id=a." + column("audit", "id") + ")"
-            + " AND NOT EXISTS (SELECT 1 FROM worm_audit_chain c WHERE " + containsDetail() + ") ORDER BY d." + column("auditDetail", "id") + " LIMIT " + limit;
-         return (in("auditDetail", "id", sqlIds(sql, cutoff, cutoff.minusSeconds(hours * 3600L))));
+            + " AND NOT EXISTS (SELECT 1 FROM " + WormAuditTables.CHAIN_DETAIL + " x WHERE x.detail_id=" + detailId + ") ORDER BY " + detailId + " LIMIT " + limit;
+         return (in("auditDetail", "id", sqlIds(sql, floor, cutoff, cutoff.minusSeconds(hours * 3600L))));
       }
       List<QRecord> links = all(WormAuditTables.CHAIN);
       Set<Long> headers = links.stream().filter(r -> "AUDIT".equals(r.getValueString("kind"))).map(r -> r.getValueLong("audit_id")).collect(Collectors.toSet());
@@ -409,11 +443,89 @@ public class AuditStore
       String time = "a." + column("audit", "timestamp");
       String headers = "SELECT " + auditId + " AS id, (SELECT count(*) FROM worm_audit_chain c WHERE c.kind='AUDIT' AND c.audit_id=" + auditId
          + ") AS coverage FROM " + physical("audit") + " a WHERE " + time + " < ? OR " + time + " IS NULL";
-      String details = "SELECT d." + column("auditDetail", "id") + " AS id, (SELECT count(*) FROM worm_audit_chain c WHERE " + containsDetail()
-         + ") AS coverage FROM " + physical("auditDetail") + " d LEFT JOIN " + physical("audit") + " a ON " + auditId + "=d."
-         + column("auditDetail", "auditId") + " WHERE " + time + " < ? OR " + time + " IS NULL";
+      String detailId = "d." + column("auditDetail", "id");
+      String scope = " FROM " + physical("auditDetail") + " d LEFT JOIN " + physical("audit") + " a ON " + auditId + "=d."
+         + column("auditDetail", "auditId");
+      String details;
+      if(isPostgres())
+      {
+         ///////////////////////////////////////////////////////////////////////
+         // Count coverage from the chain itself (the hashed source of truth), //
+         // aggregating every link's ids once rather than probing per detail.  //
+         ///////////////////////////////////////////////////////////////////////
+         details = "SELECT " + detailId + " AS id, coalesce(cv.n, 0) AS coverage" + scope + " LEFT JOIN (" + CHAIN_COVERAGE
+            + ") cv ON cv.id=" + detailId + " WHERE " + time + " < ? OR " + time + " IS NULL";
+      }
+      else
+      {
+         details = "SELECT " + detailId + " AS id, (SELECT count(*) FROM worm_audit_chain c WHERE POSITION(',' || CAST(" + detailId
+            + " AS varchar) || ',' IN ',' || c.detail_ids || ',') > 0) AS coverage" + scope + " WHERE " + time + " < ? OR " + time + " IS NULL";
+      }
       reportCoverage(headers, "audit", cutoff, result);
       reportCoverage(details, "audit_detail", cutoff, result);
+      coverageTable(result);
+   }
+
+
+   /*******************************************************************************
+    ** The coverage table must hold exactly the chain's own detail lists. A
+    ** mismatch would let the writer skip or re-chain a detail, so it is a break.
+    *******************************************************************************/
+   private void coverageTable(VerificationResult result) throws QException
+   {
+      String table = "SELECT detail_id, chain_id FROM " + WormAuditTables.CHAIN_DETAIL;
+      String sql = isPostgres()
+         ? "SELECT detail_id, count(*) OVER() FROM ((" + CHAIN_PAIRS + " EXCEPT " + table + ") UNION ALL (" + table + " EXCEPT " + CHAIN_PAIRS
+            + ")) mismatch ORDER BY detail_id LIMIT 100"
+         : "SELECT x.detail_id, count(*) OVER() FROM " + WormAuditTables.CHAIN_DETAIL + " x WHERE NOT EXISTS (SELECT 1 FROM worm_audit_chain c"
+            + " WHERE c.id=x.chain_id AND POSITION(',' || CAST(x.detail_id AS varchar) || ',' IN ',' || c.detail_ids || ',') > 0) ORDER BY x.detail_id LIMIT 100";
+      try(PreparedStatement statement = ((RDBMSTransaction) transaction).getConnection().prepareStatement(sql); ResultSet rows = statement.executeQuery())
+      {
+         long total = 0;
+         int reported = 0;
+         while(rows.next())
+         {
+            total = rows.getLong(2);
+            result.add(null, null, "audit_detail " + com.kingsrook.qbits.wormaudit.verify.WormAuditVerifier.number(rows.getLong(1))
+               + " has a detail coverage row that disagrees with the chain");
+            reported++;
+         }
+         result.omitted(total - reported);
+      }
+      catch(Exception e)
+      {
+         throw (AppendOnlyGuard.map(e));
+      }
+   }
+
+
+   /*******************************************************************************
+    ** Highest id in a core audit table, or 0 when it is empty (RDBMS only).
+    *******************************************************************************/
+   public long maxId(String table) throws QException
+   {
+      return (sqlIds("SELECT coalesce(max(" + column(table, "id") + "), 0) FROM " + physical(table)).getFirst());
+   }
+
+
+   /*******************************************************************************
+    ** PostgreSQL maintains detail coverage in its chain trigger; other backends
+    ** have no trigger, so the writer appends coverage rows itself.
+    *******************************************************************************/
+   public boolean isPostgres() throws QException
+   {
+      if(postgres == null)
+      {
+         try
+         {
+            postgres = isRdbms() && "PostgreSQL".equals(((RDBMSTransaction) transaction).getConnection().getMetaData().getDatabaseProductName());
+         }
+         catch(Exception e)
+         {
+            throw (AppendOnlyGuard.map(e));
+         }
+      }
+      return (postgres);
    }
 
 
@@ -453,36 +565,22 @@ public class AuditStore
 
 
    /*******************************************************************************
-    ** PostgreSQL's containment expression matches the GIN expression index.
+    ** Read ids from an anti-join, binding instants by column type and ids as longs.
     *******************************************************************************/
-   private String containsDetail() throws QException
-   {
-      try
-      {
-         String id = "d." + column("auditDetail", "id");
-         if("PostgreSQL".equals(((RDBMSTransaction) transaction).getConnection().getMetaData().getDatabaseProductName()))
-         {
-            return ("string_to_array(c.detail_ids, ',') @> ARRAY[" + id + "::text]");
-         }
-         return ("POSITION(',' || CAST(" + id + " AS varchar) || ',' IN ',' || c.detail_ids || ',') > 0");
-      }
-      catch(Exception e)
-      {
-         throw (AppendOnlyGuard.map(e));
-      }
-   }
-
-
-   /*******************************************************************************
-    ** Read ids from an anti-join, binding all instants according to column type.
-    *******************************************************************************/
-   private List<Long> sqlIds(String sql, Instant... cutoffs) throws QException
+   private List<Long> sqlIds(String sql, Object... parameters) throws QException
    {
       try(PreparedStatement statement = ((RDBMSTransaction) transaction).getConnection().prepareStatement(sql))
       {
-         for(int i = 0; i < cutoffs.length; i++)
+         for(int i = 0; i < parameters.length; i++)
          {
-            bindCutoff(statement, i + 1, cutoffs[i]);
+            if(parameters[i] instanceof Instant instant)
+            {
+               bindCutoff(statement, i + 1, instant);
+            }
+            else
+            {
+               statement.setLong(i + 1, (Long) parameters[i]);
+            }
          }
          List<Long> ids = new ArrayList<>();
          try(ResultSet rows = statement.executeQuery())

src/main/java/com/kingsrook/qbits/wormaudit/chain/SweepFloors.java

@@ -0,0 +1,159 @@
+/*
+ * 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.wormaudit.chain;
+
+
+import java.time.Duration;
+import java.time.Instant;
+import java.util.ArrayDeque;
+import java.util.Collections;
+import java.util.Deque;
+import java.util.Map;
+import java.util.WeakHashMap;
+import com.kingsrook.qbits.wormaudit.WormAuditQBitConfig;
+
+
+/*******************************************************************************
+ ** Per-JVM id floors for the chain writer's two sweeps, so a tick reads only
+ ** audit and detail rows newer than the last complete sweep instead of the whole
+ ** settle window or late-detail lookback.
+ **
+ ** Each tick records the highest audit and audit_detail ids it can see. After a
+ ** sweep that returned fewer rows than its limit, every row allocated before
+ ** (settle horizon + MARGIN) ago is chained, on the same assumption the settle
+ ** horizon already makes: a row whose id was allocated that long ago has
+ ** committed. The floor moves to the highest id observed at or before that time.
+ ** A row whose transaction commits later than that is still chained: once an
+ ** hour (and after every restart) a tick ignores the floors and sweeps
+ ** everything, exactly as before floors existed. Floors live in memory only; a
+ ** second replica keeps its own. The daily verification checks full coverage
+ ** independently of all of this.
+ *******************************************************************************/
+public final class SweepFloors
+{
+   public static final Duration MARGIN = Duration.ofMinutes(5);
+   public static final Duration FULL_SWEEP_EVERY = Duration.ofHours(1);
+
+   private static final Map<WormAuditQBitConfig, SweepFloors> BY_CONFIG = Collections.synchronizedMap(new WeakHashMap<>());
+
+   private final Deque<Observation> observations = new ArrayDeque<>();
+   private long audit;
+   private long detail;
+   private Instant lastFullSweep;
+
+   /*******************************************************************************
+    ** The highest ids visible at one tick.
+    *******************************************************************************/
+   public record Observation(Instant at, long audit, long detail)
+   {
+   }
+
+
+   /*******************************************************************************
+    ** Floors for one installed qbit (config identity), created on first use.
+    *******************************************************************************/
+   public static SweepFloors of(WormAuditQBitConfig config)
+   {
+      return (BY_CONFIG.computeIfAbsent(config, c -> new SweepFloors()));
+   }
+
+
+   /*******************************************************************************
+    ** Audit ids at or below this are known chained.
+    *******************************************************************************/
+   public synchronized long audit()
+   {
+      return (audit);
+   }
+
+
+   /*******************************************************************************
+    ** Detail ids at or below this are known covered (or outside the lookback).
+    *******************************************************************************/
+   public synchronized long detail()
+   {
+      return (detail);
+   }
+
+
+   /*******************************************************************************
+    ** Whether this tick must ignore the floors: first tick, an hour since the last
+    ** complete full sweep, or a clock that moved backwards.
+    *******************************************************************************/
+   public synchronized boolean fullSweepDue(Instant now)
+   {
+      return (lastFullSweep == null || now.isBefore(lastFullSweep) || !now.isBefore(lastFullSweep.plus(FULL_SWEEP_EVERY)));
+   }
+
+
+   /*******************************************************************************
+    ** Record the highest visible ids. Ids below a floor mean rows were removed
+    ** wholesale (a purge of everything, or a reset test database): start over.
+    *******************************************************************************/
+   public synchronized Observation observe(Instant now, long maxAudit, long maxDetail)
+   {
+      if(maxAudit < audit || maxDetail < detail)
+      {
+         observations.clear();
+         audit = 0;
+         detail = 0;
+      }
+      Observation observation = new Observation(now, maxAudit, maxDetail);
+      observations.addLast(observation);
+      return (observation);
+   }
+
+
+   /*******************************************************************************
+    ** After a tick, raise each completely swept floor to the newest observation
+    ** made at or before the horizon, then drop observations no longer needed. A
+    ** full sweep (fullSweepAt not null) counts only when both sweeps completed.
+    *******************************************************************************/
+   public synchronized void swept(Observation current, Instant horizon, boolean auditsComplete, boolean detailsComplete, Instant fullSweepAt)
+   {
+      if(fullSweepAt != null && auditsComplete && detailsComplete)
+      {
+         lastFullSweep = fullSweepAt;
+      }
+      Observation settled = null;
+      for(Observation observation : observations)
+      {
+         if(!observation.at().isAfter(horizon) && (settled == null || !observation.at().isBefore(settled.at())))
+         {
+            settled = observation;
+         }
+      }
+      if(settled == null)
+      {
+         return;
+      }
+      if(auditsComplete)
+      {
+         audit = Math.max(audit, Math.min(settled.audit(), current.audit()));
+      }
+      if(detailsComplete)
+      {
+         detail = Math.max(detail, Math.min(settled.detail(), current.detail()));
+      }
+      Observation keep = settled;
+      observations.removeIf(observation -> observation != keep && !observation.at().isAfter(horizon));
+   }
+}

src/main/java/com/kingsrook/qbits/wormaudit/chain/WormAuditChainer.java

@@ -128,7 +128,10 @@ public class WormAuditChainer
       String previous = tail == null ? anchor.getValueString("anchor_hash") : tail.getValueString("link_hash");
       List<QRecord> appended = new ArrayList<>();
       Instant cutoff = now.minusSeconds(settings.settle_horizon_seconds());
-      List<QRecord> audits = store.unchained(cutoff, config.getBatchSize());
+      SweepFloors floors = SweepFloors.of(config);
+      SweepFloors.Observation observed = store.isRdbms() ? floors.observe(now, store.maxId("audit"), store.maxId("auditDetail")) : null;
+      boolean full = floors.fullSweepDue(now);
+      List<QRecord> audits = store.unchained(cutoff, config.getBatchSize(), full ? 0L : floors.audit());
       Map<Long, QRecord> details = AuditStore.index(store.in("auditDetail", "auditId", audits.stream().map(r -> r.getValueLong("id")).toList()));
       Map<Long, List<QRecord>> grouped = details.values().stream().collect(Collectors.groupingBy(r -> r.getValueLong("auditId")));
       int lateDetails = 0;
@@ -150,7 +153,7 @@ public class WormAuditChainer
       insertLinks(store, appended);
       int linkCount = appended.size();
       appended.clear();
-      details = AuditStore.index(store.lateDetails(cutoff, settings.late_detail_lookback_hours(), config.getBatchSize()));
+      details = AuditStore.index(store.lateDetails(cutoff, settings.late_detail_lookback_hours(), config.getBatchSize(), full ? 0L : floors.detail()));
       grouped = details.values().stream().collect(Collectors.groupingBy(r -> r.getValueLong("auditId"), TreeMap::new, Collectors.toList()));
       Map<Long, QRecord> headers = AuditStore.index(store.in("audit", "id", grouped.keySet()));
       for(var entry : grouped.entrySet())
@@ -166,12 +169,19 @@ public class WormAuditChainer
          }
       }
       insertLinks(store, appended);
+      if(observed != null)
+      {
+         floors.swept(observed, now.minusSeconds(settings.settle_horizon_seconds()).minus(SweepFloors.MARGIN),
+            audits.size() < config.getBatchSize(), details.size() < config.getBatchSize(), full ? now : null);
+      }
       return (new Counts(audits.size(), lateDetails, linkCount + appended.size(), checkpoint(store, now, settings), false));
    }
 
 
    /*******************************************************************************
-    ** Persist bounded insert batches inside the writer transaction.
+    ** Persist bounded insert batches inside the writer transaction. PostgreSQL's
+    ** chain trigger writes the detail coverage rows; elsewhere the writer does,
+    ** in the same transaction, so coverage never diverges from the chain.
     *******************************************************************************/
    private void insertLinks(AuditStore store, List<QRecord> links) throws QException
    {
@@ -179,6 +189,18 @@ public class WormAuditChainer
       {
          store.insert(WormAuditTables.CHAIN, links.subList(start, Math.min(start + config.getBatchSize(), links.size())));
       }
+      if(!store.isPostgres())
+      {
+         List<QRecord> coverage = new ArrayList<>();
+         for(QRecord link : links)
+         {
+            AuditStore.detailIds(link).forEach(id -> coverage.add(new QRecord().withValue("detail_id", id).withValue("chain_id", link.getValueLong("id"))));
+         }
+         for(int start = 0; start < coverage.size(); start += config.getBatchSize())
+         {
+            store.insert(WormAuditTables.CHAIN_DETAIL, coverage.subList(start, Math.min(start + config.getBatchSize(), coverage.size())));
+         }
+      }
    }
 
 

src/main/java/com/kingsrook/qbits/wormaudit/model/WormAuditHelp.java

@@ -115,7 +115,7 @@ public final class WormAuditHelp
                table.withHelpContent(SLOT, content(entry.getValue()));
             }
          }
-         else if(key.startsWith("app:"))
+         else if(config.getIncludeApp() && key.startsWith("app:"))
          {
             QAppMetaData app = qInstance.getApp(config.getAppName());
             if(app != null && (app.getHelpContent() == null || !app.getHelpContent().containsKey(SLOT)))

src/main/java/com/kingsrook/qbits/wormaudit/model/WormAuditTables.java

@@ -52,13 +52,14 @@ import com.kingsrook.qqq.backend.module.rdbms.model.metadata.RDBMSTableBackendDe
 public final class WormAuditTables
 {
    public static final String CHAIN = "worm_audit_chain";
+   public static final String CHAIN_DETAIL = "worm_audit_chain_detail";
    public static final String CHECKPOINT = "worm_audit_checkpoint";
    public static final String EXPORT = "worm_audit_export";
    public static final String VERIFICATION = "worm_audit_verification";
    public static final String PURGE = "worm_audit_purge";
    public static final String SETTINGS = "worm_audit_settings";
-   public static final List<String> ALL = List.of(CHAIN, CHECKPOINT, EXPORT, VERIFICATION, PURGE, SETTINGS);
-   public static final List<String> NAV = List.of(VERIFICATION, EXPORT, CHECKPOINT, CHAIN, PURGE, SETTINGS);
+   public static final List<String> ALL = List.of(CHAIN, CHAIN_DETAIL, CHECKPOINT, EXPORT, VERIFICATION, PURGE, SETTINGS);
+   public static final List<String> NAV = List.of(VERIFICATION, EXPORT, CHECKPOINT, CHAIN, CHAIN_DETAIL, PURGE, SETTINGS);
 
    /*******************************************************************************
     ** Utility class.
@@ -75,6 +76,9 @@ public final class WormAuditTables
    {
       List<QTableMetaData> tables = new ArrayList<>();
       tables.add(table(config, CHAIN, "Audit Chain Link", "id:L kind:S audit_id:L detail_ids:T prev_hash:S link_hash:S chained_at:D"));
+      // Keyed by the covered audit_detail id: one row per detail, written with its link.
+      tables.add(table(config, CHAIN_DETAIL, "Detail Coverage", "detail_id:L chain_id:L").withPrimaryKeyField("detail_id")
+         .withRecordLabelFormatAndFields("Detail #%s (link %s)", "detail_id", "chain_id"));
       tables.add(table(config, CHECKPOINT, "Checkpoint", "id:L from_link_id:L to_link_id:L link_count:L head_hash:S merkle_root:S algorithm:S key_id:S public_key:T signature:T created_at:D"));
       tables.add(table(config, EXPORT, "Sealed Export", "id:L from_link_id:L to_link_id:L from_checkpoint_id:L to_checkpoint_id:L bucket_alias:S object_prefix:T object_versions:T manifest_key:T manifest_sha256:S prev_manifest_sha256:S object_count:L byte_count:L lock_mode:S retain_until:D sealed_at:D"));
       tables.add(table(config, VERIFICATION, "Verification", "id:L kind:S trigger:S started_at:D finished_at:D from_link_id:L to_link_id:L links_checked:L outcome:S break_count:L first_break_link_id:L first_break_audit_id:L first_break_reason:T export_id:L run_by:S"));
@@ -167,7 +171,7 @@ public final class WormAuditTables
          };
          table.withField(new QFieldMetaData(parts[0], type).withBackendName(parts[0]).withLabel(fieldLabel(parts[0])));
       }
-      for(String reference : List.of("from_link_id", "to_link_id", "first_break_link_id", "anchor_link_id"))
+      for(String reference : List.of("from_link_id", "to_link_id", "first_break_link_id", "anchor_link_id", "chain_id"))
       {
          if(table.getFields().containsKey(reference))
          {
@@ -239,6 +243,7 @@ public final class WormAuditTables
       {
          case SETTINGS -> "Settings";
          case CHAIN -> "Audit Chain";
+         case CHAIN_DETAIL -> "Detail Coverage";
          case CHECKPOINT -> "Checkpoints";
          case EXPORT -> "Sealed Exports";
          case VERIFICATION -> "Verifications";
@@ -258,6 +263,8 @@ public final class WormAuditTables
          case "kind" -> "Kind";
          case "audit_id" -> "Audit";
          case "detail_ids" -> "Detail ids";
+         case "detail_id" -> "Audit detail";
+         case "chain_id" -> "Link";
          case "prev_hash" -> "Previous hash";
          case "link_hash" -> "Link hash";
          case "chained_at" -> "Chained";

src/main/java/com/kingsrook/qbits/wormaudit/processes/VerificationRuns.java

@@ -24,6 +24,7 @@ package com.kingsrook.qbits.wormaudit.processes;
 import java.time.Instant;
 import java.util.List;
 import java.util.Set;
+import java.util.stream.Collectors;
 import com.kingsrook.qbits.wormaudit.WormAuditQBitConfig;
 import com.kingsrook.qbits.wormaudit.chain.AuditStore;
 import com.kingsrook.qbits.wormaudit.model.WormAuditTables;
@@ -43,6 +44,11 @@ import com.kingsrook.qqq.backend.core.model.data.QRecord;
  *******************************************************************************/
 public final class VerificationRuns
 {
+   /** How many break reasons the alert summary spells out; the title carries the total. */
+   private static final int ALERT_SUMMARY_REASONS = 3;
+
+
+
    /*******************************************************************************
     ** Utility class.
     *******************************************************************************/
@@ -79,8 +85,15 @@ public final class VerificationRuns
             }
             else
             {
+               long problems = result.getBreakCount();
+               String summary = result.getBreaks().stream().limit(ALERT_SUMMARY_REASONS).map(b -> b.reason()).collect(Collectors.joining("; "));
+               if(problems > ALERT_SUMMARY_REASONS)
+               {
+                  summary += "; and " + (problems - ALERT_SUMMARY_REASONS) + " more";
+               }
                QAlerts.raise(new AlertSignal("wormaudit.chain.break").withDedupKey("chain")
-                  .withTitle(result.getBreaks().getFirst().reason()).withRecordRef(new AlertRecordRef(WormAuditTables.VERIFICATION, row.getValue("id"))));
+                  .withTitle("Audit chain verification found " + problems + (problems == 1 ? " problem" : " problems")).withSummary(summary)
+                  .withRecordRef(new AlertRecordRef(WormAuditTables.VERIFICATION, row.getValue("id"))));
             }
          }
          return (result);

src/main/resources/com/kingsrook/qbits/wormaudit/ddl/postgres/worm-audit.sql

@@ -385,3 +385,172 @@ BEGIN
    END LOOP;
 END;
 $$;
+
+-- changeset worm-audit:13 splitStatements:false
+-- comment: Normalized detail coverage. One insert-only row per audit_detail id covered by a chain link, written by the coverage trigger in the same statement as the link, so the chain writer probes a primary key instead of the GIN array index. Append-only like the chain, with no foreign keys (the owner cannot UPDATE WORM tables, so FK checks would be denied). The advisory lock keeps a running chainer from appending between the backfill and the new trigger body.
+DO $$
+BEGIN
+   PERFORM pg_advisory_xact_lock(86753090421);
+END;
+$$;
+CREATE TABLE IF NOT EXISTS public.worm_audit_chain_detail (
+   detail_id bigint PRIMARY KEY,
+   chain_id bigint NOT NULL,
+   CHECK (detail_id >= 0),
+   CHECK (chain_id > 0)
+);
+CREATE INDEX IF NOT EXISTS worm_audit_chain_detail_chain ON public.worm_audit_chain_detail(chain_id);
+-- The first link wins if a tampered chain covers a detail twice; the verifier still reports the duplicate from the chain itself.
+INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
+   SELECT x.id::bigint, c.id FROM public.worm_audit_chain c CROSS JOIN LATERAL unnest(string_to_array(c.detail_ids, ',')) AS x(id)
+   WHERE c.detail_ids <> '' ORDER BY c.id
+   ON CONFLICT (detail_id) DO NOTHING;
+
+CREATE OR REPLACE FUNCTION public.worm_audit_reject() RETURNS trigger
+LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
+DECLARE
+   old_time timestamptz;
+   function_owner name;
+   stack text;
+   row_identity text := 'all';
+BEGIN
+   IF TG_OP = 'DELETE' AND current_setting('worm_audit.purging', true) = 'on' THEN
+      SELECT pg_get_userbyid(proowner) INTO function_owner FROM pg_proc
+         WHERE oid = 'public.worm_audit_purge(timestamptz)'::regprocedure;
+      GET DIAGNOSTICS stack = PG_CONTEXT;
+      IF current_user = function_owner AND position('function worm_audit_purge(timestamp with time zone)' in stack) > 0 THEN
+         IF TG_RELID = '${audit}'::regclass THEN
+            old_time := OLD.timestamp;
+         ELSIF TG_RELID = '${audit_detail}'::regclass THEN
+            SELECT timestamp INTO old_time FROM ${audit} WHERE id = OLD.audit_id;
+         ELSIF TG_RELID = 'public.worm_audit_chain'::regclass THEN
+            SELECT timestamp INTO old_time FROM ${audit} WHERE id = OLD.audit_id;
+         ELSIF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
+            SELECT a.timestamp INTO old_time FROM public.worm_audit_chain c JOIN ${audit} a ON a.id = c.audit_id WHERE c.id = OLD.chain_id;
+         END IF;
+         IF old_time < now() - interval '${retention}' THEN
+            RETURN OLD;
+         END IF;
+      END IF;
+   END IF;
+   IF TG_LEVEL = 'ROW' THEN
+      -- Record fields resolve per statement, so each table's key is read in its own branch.
+      IF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
+         row_identity := OLD.detail_id::text;
+      ELSE
+         row_identity := OLD.id::text;
+      END IF;
+   END IF;
+   RAISE EXCEPTION USING ERRCODE = 'P0001',
+      MESSAGE = format('worm_audit: audit rows are write-once; %s of "%s" is not allowed (id=%s). Expired rows are removed only by worm_audit_purge().',
+         TG_OP, TG_TABLE_NAME, row_identity),
+      HINT = 'see qbit-worm-audit docs/DESIGN.md section 4';
+END;
+$fn$;
+
+CREATE OR REPLACE FUNCTION public.worm_audit_detail_coverage() RETURNS trigger
+LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp AS $fn$
+BEGIN
+   -- Serialize even direct SQL appends, then record coverage in the same statement as the link.
+   -- The primary key rejects a detail covered twice, within this link or by an earlier one. The
+   -- previous EXISTS overlap check ran as a cached generic plan that chose a sequential scan of
+   -- the whole chain for every inserted link.
+   PERFORM pg_advisory_xact_lock(86753090421);
+   IF NEW.detail_ids = '' THEN
+      RETURN NEW;
+   END IF;
+   BEGIN
+      INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
+         SELECT d, NEW.id FROM unnest(string_to_array(NEW.detail_ids, ',')::bigint[]) AS d;
+   EXCEPTION WHEN unique_violation THEN
+      RAISE EXCEPTION USING ERRCODE = '23505', MESSAGE = 'worm_audit: audit detail is already covered by the chain';
+   END;
+   RETURN NEW;
+END;
+$fn$;
+
+-- The array index worm_audit_chain_detail_coverage is no longer used by this release. It stays until
+-- a later migration drops it, so a previous-release writer still running during a rolling deploy keeps its indexed probe.
+
+DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_chain_detail;
+CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_chain_detail
+   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
+DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_chain_detail;
+CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_chain_detail
+   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
+REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_chain_detail FROM ${app_role};
+GRANT SELECT, INSERT ON public.worm_audit_chain_detail TO ${app_role};
+
+-- Same function as worm-audit:5, except that coverage rows are locked and purged before their links.
+-- The migrating role may have revoked its own EXECUTE (a runtime role that owns the schema, as on CNPG);
+-- PostgreSQL then refuses CREATE OR REPLACE. Grant it back for this statement only, then restore.
+DO $$
+BEGIN
+   IF NOT has_function_privilege(current_user, 'public.worm_audit_purge(timestamptz)', 'EXECUTE') THEN
+      EXECUTE format('GRANT EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) TO %I', current_user);
+      PERFORM set_config('worm_audit.purge_regrant', 'on', true);
+   END IF;
+END;
+$$;
+CREATE OR REPLACE FUNCTION public.worm_audit_purge(p_cutoff timestamptz) RETURNS void
+LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
+DECLARE
+   boundary bigint;
+   smaller bigint;
+   anchor text;
+   audits_count bigint;
+   details_count bigint;
+   links_count bigint;
+BEGIN
+   IF p_cutoff IS NULL OR p_cutoff > now() - interval '${retention}' THEN
+      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
+         VALUES(now(), session_user, p_cutoff, '${retention}', 'REFUSED', 'Cutoff is within retention or null');
+      RETURN;
+   END IF;
+   -- Same lock as the Java chainer; table locks also exclude concurrent audit inserts.
+   PERFORM pg_advisory_xact_lock(86753090421);
+   LOCK TABLE ${audit}, ${audit_detail}, public.worm_audit_chain, public.worm_audit_chain_detail IN SHARE ROW EXCLUSIVE MODE;
+   SELECT coalesce(min(c.id) - 1, (SELECT max(id) FROM public.worm_audit_chain)) INTO boundary
+      FROM public.worm_audit_chain c LEFT JOIN ${audit} a ON a.id = c.audit_id
+      WHERE a.id IS NULL OR a.timestamp IS NULL OR a.timestamp >= p_cutoff;
+   -- Shrink to a prefix closed over every audit's links AND all its stored details.
+   LOOP
+      SELECT min(c.id) - 1 INTO smaller FROM public.worm_audit_chain c
+         WHERE c.id <= boundary AND (
+            EXISTS (SELECT FROM public.worm_audit_chain later WHERE later.audit_id = c.audit_id AND later.id > boundary)
+            OR EXISTS (SELECT FROM ${audit_detail} d WHERE d.audit_id = c.audit_id AND NOT EXISTS (
+               SELECT FROM public.worm_audit_chain covered WHERE covered.audit_id = c.audit_id
+                  AND d.id::text = ANY(string_to_array(covered.detail_ids, ',')))));
+      EXIT WHEN smaller IS NULL;
+      boundary := smaller;
+   END LOOP;
+   SELECT link_hash INTO anchor FROM public.worm_audit_chain WHERE id = boundary;
+   IF anchor IS NULL THEN
+      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
+         VALUES(now(), session_user, p_cutoff, '${retention}', 'REFUSED', 'No complete expired chain prefix');
+      RETURN;
+   END IF;
+   PERFORM set_config('worm_audit.purging', 'on', true);
+   DELETE FROM ${audit_detail} d WHERE EXISTS
+      (SELECT FROM public.worm_audit_chain c WHERE c.audit_id = d.audit_id AND c.id <= boundary);
+   GET DIAGNOSTICS details_count = ROW_COUNT;
+   -- Coverage rows go before their links: the guard finds their age through the link.
+   DELETE FROM public.worm_audit_chain_detail WHERE chain_id <= boundary;
+   -- Keep ids locally because links must be deleted before their audit headers.
+   WITH removed AS (DELETE FROM public.worm_audit_chain WHERE id <= boundary RETURNING audit_id),
+      removed_audits AS (DELETE FROM ${audit} WHERE id IN (SELECT audit_id FROM removed) RETURNING id)
+      SELECT (SELECT count(*) FROM removed), (SELECT count(*) FROM removed_audits) INTO links_count, audits_count;
+   PERFORM set_config('worm_audit.purging', 'off', true);
+   INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome,
+      audits_deleted, details_deleted, links_deleted, anchor_link_id, anchor_hash)
+      VALUES(now(), session_user, p_cutoff, '${retention}', 'PURGED', audits_count, details_count, links_count, boundary, anchor);
+END;
+$fn$;
+DO $$
+BEGIN
+   IF current_setting('worm_audit.purge_regrant', true) = 'on' THEN
+      EXECUTE format('REVOKE EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) FROM %I', current_user);
+      PERFORM set_config('worm_audit.purge_regrant', 'off', true);
+   END IF;
+END;
+$$;

src/main/resources/help/qbit-worm-audit.yaml

@@ -24,6 +24,11 @@ entries:
     content: |
       **How rows get created:** the Chain settled audit records job appends links for audit records older than the settle horizon (every 30 seconds by default, set in Settings). Nobody adds, edits or deletes a link; the database refuses. An empty table means no audit record has settled yet, or the chain job is not running.
 
+  - key: table:worm_audit_chain_detail;slot:help
+    summary: "Which chain link covers each audit detail: one row per detail, written with the link."
+    content: |
+      **How rows get created:** whenever the chain job appends a link, the database writes one row here for every audit detail the link covers, in the same transaction (the database script also fills it for links that existed before this table). It lets the chain job find details that are not chained yet without scanning the chain. Nobody adds, edits or deletes a row; the database refuses, and verification checks that it matches the chain exactly. An empty table means no audit detail has been chained yet.
+
   - key: table:worm_audit_checkpoint;slot:help
     summary: "Signed ranges of links (Ed25519), proving the chain's head at a point in time."
     content: |

src/test/java/com/kingsrook/qbits/wormaudit/CanonicalAndCryptoTest.java

@@ -167,6 +167,9 @@ class CanonicalAndCryptoTest
                   java.nio.file.Files.readString(
                         java.nio.file.Path.of("docs/ddl/worm-audit-postgres.sql")));
       assertThat(sql).doesNotContain("${");
+      // Liquibase parses every line starting with "-- changeset" as a directive, comments included.
+      assertThat(sql.lines().filter(line -> line.startsWith("-- changeset")))
+            .allMatch(line -> line.matches("-- changeset worm-audit:[0-9]+( splitStatements:false)?"));
       assertThat(sql).contains("GET DIAGNOSTICS stack = PG_CONTEXT", "function worm_audit_purge(timestamp with time zone)", "SET timezone = 'UTC'");
       // The documented migration contract deliberately uses physical snake_case columns.
       assertThat(sql).contains("OLD.audit_id", "OLD.timestamp", "NEW.name", "GRANT UPDATE (label, create_date, modify_date)");

src/test/java/com/kingsrook/qbits/wormaudit/PostgresAcceptanceTest.java

@@ -328,6 +328,11 @@ class PostgresAcceptanceTest
       assertThat(purge.getValueLong("details_deleted")).isEqualTo(2);
       assertThat(purge.getValueLong("links_deleted")).isEqualTo(2);
       assertThat(purge.getValueString("anchor_hash")).isEqualTo(anchor);
+      try(var c = owner())
+      {
+         assertThat(count(c, "SELECT count(*) FROM worm_audit_chain_detail WHERE chain_id <= 2")).as("coverage purged with its links").isZero();
+         assertThat(count(c, "SELECT count(*) FROM worm_audit_chain_detail")).isZero();
+      }
       assertThat(new WormAuditVerifier(config).verify().getBreaks()).isEmpty();
       detail(3, 3);
       new WormAuditChainer(config).chainOnce();
@@ -593,6 +598,69 @@ class PostgresAcceptanceTest
    }
 
 
+   /*******************************************************************************
+    ** The chain trigger writes one coverage row per detail with its link; the
+    ** table is write-once, backfilled from the chain by the DDL, purged with its
+    ** links, and verification breaks if it ever disagrees with the chain.
+    *******************************************************************************/
+   @Test
+   void detailCoverageTableIsWrittenByTheTriggerBackfilledAndWriteOnce() throws Exception
+   {
+      // The fixture's audit headers are years old; poll them as late-detail candidates.
+      setting("late_detail_lookback_hours", 24 * 365 * 10);
+      audit(1);
+      detail(1, 1);
+      detail(2, 1);
+      audit(2);
+      detail(3, 2);
+      var writer = new WormAuditChainer(config);
+      assertThat(writer.chainOnce().links()).isEqualTo(2);
+      detail(4, 1);
+      assertThat(writer.chainOnce().lateDetails()).isEqualTo(1);
+      assertThat(writer.chainOnce().links()).isZero();
+      String pairs = "SELECT string_agg(detail_id || '@' || chain_id, ',' ORDER BY detail_id) FROM worm_audit_chain_detail";
+      try(var c = owner(); var runtime = app(); var s = c.createStatement())
+      {
+         try(var r = s.executeQuery(pairs))
+         {
+            r.next();
+            assertThat(r.getString(1)).isEqualTo("1@1,2@1,3@2,4@3");
+         }
+         assertThatThrownBy(() -> sql(runtime, "INSERT INTO worm_audit_chain_detail(detail_id,chain_id) VALUES(4,1)"))
+            .isInstanceOf(SQLException.class).satisfies(e -> assertThat(((SQLException) e).getSQLState()).isEqualTo("23505"));
+         for(String privilege : List.of("UPDATE", "DELETE", "TRUNCATE"))
+         {
+            assertThat(count(c, "SELECT has_table_privilege('app','worm_audit_chain_detail','" + privilege + "')::int")).isZero();
+         }
+         sql(c, "GRANT UPDATE, DELETE, TRUNCATE ON worm_audit_chain_detail TO app");
+         for(String operation : List.of("UPDATE worm_audit_chain_detail SET chain_id=chain_id", "DELETE FROM worm_audit_chain_detail", "TRUNCATE worm_audit_chain_detail"))
+         {
+            assertThatThrownBy(() -> sql(runtime, operation)).isInstanceOf(SQLException.class).hasMessageContaining("worm_audit:")
+               .satisfies(e -> assertThat(((SQLException) e).getSQLState()).isEqualTo("P0001"));
+         }
+
+         ///////////////////////////////////////////////////////////////////////
+         // An install that predates the table: the DDL backfills it from the  //
+         // chain, and the writer then finds nothing left to chain.            //
+         ///////////////////////////////////////////////////////////////////////
+         sql(c, "DROP TABLE worm_audit_chain_detail");
+         sql(c, WormAuditDdl.postgres(config));
+         try(var r = s.executeQuery(pairs))
+         {
+            r.next();
+            assertThat(r.getString(1)).isEqualTo("1@1,2@1,3@2,4@3");
+         }
+         assertThat(writer.chainOnce().links()).isZero();
+         assertThat(new WormAuditVerifier(config).verify().getBreaks()).isEmpty();
+
+         sql(c, "ALTER TABLE worm_audit_chain_detail DISABLE TRIGGER ALL; UPDATE worm_audit_chain_detail SET chain_id=1 WHERE detail_id=3;"
+            + " ALTER TABLE worm_audit_chain_detail ENABLE TRIGGER ALL");
+      }
+      assertThat(new WormAuditVerifier(config).verify().getBreaks()).anyMatch(b -> b.reason().contains("audit_detail 3 has a detail coverage row that disagrees"));
+   }
+
+
+
    /*******************************************************************************
     ** Referential integrity guardrail: every table reference of the WORM tables
     ** (possible-value fields, and the chain's audit header) has a RESTRICT
@@ -611,7 +679,9 @@ class PostgresAcceptanceTest
          "worm_audit_verification.from_link_id", links,
          "worm_audit_verification.to_link_id", links,
          "worm_audit_verification.first_break_link_id", links + "; a break may name a link that is missing",
-         "worm_audit_purge.anchor_link_id", "the anchor is the last link the same purge deleted");
+         "worm_audit_purge.anchor_link_id", "the anchor is the last link the same purge deleted",
+         "worm_audit_chain_detail.chain_id", "the chain is WORM: the owner has no UPDATE, so a foreign key check into it is denied;"
+            + " the coverage trigger writes the row with its link and the purge deletes it first");
 
       java.util.Set<String> references = new java.util.TreeSet<>(List.of("worm_audit_chain.audit_id"));
       for(var table : instance.getTables().values())

src/test/java/com/kingsrook/qbits/wormaudit/ProcessesAndAlertsTest.java

@@ -142,7 +142,8 @@ class ProcessesAndAlertsTest
       assertThat(Sink.raised.getFirst().getDedupKey()).isEqualTo("chain");
       assertThat(Sink.raised.getFirst().getRecordRefs().getFirst().getTableName())
             .isEqualTo(WormAuditTables.VERIFICATION);
-      assertThat(Sink.raised.getFirst().getTitle()).contains("Break at link 1");
+      assertThat(Sink.raised.getFirst().getTitle()).isEqualTo("Audit chain verification found 1 problem");
+      assertThat(Sink.raised.getFirst().getSummary()).contains("Break at link 1");
       var first = STORE.all(WormAuditTables.VERIFICATION).getFirst();
       assertThat(first.getValueLong("first_break_link_id")).isEqualTo(1);
       assertThat(first.getValueLong("first_break_audit_id")).isEqualTo(1);

src/test/java/com/kingsrook/qbits/wormaudit/RdbmsChainTest.java

@@ -23,6 +23,7 @@ package com.kingsrook.qbits.wormaudit;
 import java.sql.DriverManager;
 import java.util.Map;
 import java.util.UUID;
+import com.kingsrook.qbits.wormaudit.chain.SweepFloors;
 import com.kingsrook.qbits.wormaudit.chain.WormAuditChainer;
 import com.kingsrook.qbits.wormaudit.model.WormAuditTables;
 import com.kingsrook.qbits.wormaudit.verify.WormAuditVerifier;
@@ -209,4 +210,55 @@ class RdbmsChainTest
       }
    }
 
+
+   /*******************************************************************************
+    ** Without a coverage trigger the writer records coverage itself, and the
+    ** sweep floors advance only past settled ids: a later header and a late
+    ** detail for an old, already chained header are both still chained.
+    *******************************************************************************/
+   @Test
+   void coverageRowsAndSweepFloors() throws Exception
+   {
+      String url = "jdbc:h2:mem:floors_" + UUID.randomUUID() + ";DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=TRUE";
+      var config = new WormAuditQBitConfig().withBatchSize(500);
+      var instance = RdbmsTestSupport.instance(url, "sa", "", config, false);
+      try(var connection = DriverManager.getConnection(url, "sa", ""); var statement = connection.createStatement())
+      {
+         RdbmsTestSupport.createTables(connection, instance, false);
+         QContext.init(instance, new QSession());
+         lookups();
+         statement.execute("INSERT INTO audit(id,audit_table_id,audit_user_id,timestamp) VALUES(1,1,1,CURRENT_TIMESTAMP - INTERVAL '2' MINUTE)");
+         detail(10, 1);
+         detail(20, 1);
+         var writer = new WormAuditChainer(config);
+         var floors = SweepFloors.of(config);
+         var now = java.time.Instant.now();
+         assertThat(writer.chainOnce(now).audits()).isEqualTo(1);
+         assertThat(STORE.query(WormAuditTables.CHAIN_DETAIL, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter())).extracting(r -> r.getValueLong("detail_id") + "@" + r.getValueLong("chain_id"))
+            .containsExactlyInAnyOrder("10@1", "20@1");
+         assertThat(floors.audit()).isZero();
+         assertThat(writer.chainOnce(now.plusSeconds(600)).links()).isZero();
+         assertThat(floors.audit()).isEqualTo(1);
+         assertThat(floors.detail()).isEqualTo(20);
+         detail(30, 1);
+         statement.execute("INSERT INTO audit(id,audit_table_id,audit_user_id,timestamp) VALUES(2,1,1,CURRENT_TIMESTAMP - INTERVAL '2' MINUTE)");
+         detail(40, 2);
+         var counts = writer.chainOnce(now.plusSeconds(1200));
+         assertThat(counts.audits()).isEqualTo(1);
+         assertThat(counts.lateDetails()).isEqualTo(1);
+         assertThat(STORE.query(WormAuditTables.CHAIN_DETAIL, new com.kingsrook.qqq.backend.core.model.actions.tables.query.QQueryFilter())).hasSize(4);
+         assertThat(writer.chainOnce(now.plusSeconds(1800)).links()).isZero();
+         assertThat(floors.audit()).isEqualTo(2);
+         assertThat(floors.detail()).isEqualTo(40);
+         // A detail whose id sits below the floor (its transaction committed very late)
+         // is skipped by floored ticks and chained by the next hourly full sweep.
+         statement.execute("INSERT INTO audit_detail(id,audit_id,field_name,new_value) VALUES(5,2,'title','late')");
+         assertThat(writer.chainOnce(now.plusSeconds(2400)).lateDetails()).isZero();
+         assertThat(writer.chainOnce(now.plus(SweepFloors.FULL_SWEEP_EVERY)).lateDetails()).isEqualTo(1);
+         assertThat(new WormAuditVerifier(config).verify().getBreaks()).isEmpty();
+         statement.execute("DELETE FROM worm_audit_chain_detail WHERE detail_id=30");
+         statement.execute("INSERT INTO worm_audit_chain_detail(detail_id,chain_id) VALUES(30,1)");
+         assertThat(new WormAuditVerifier(config).verify().getBreaks()).anyMatch(b -> b.reason().contains("audit_detail 30 has a detail coverage row that disagrees"));
+      }
+   }
 }

src/test/java/com/kingsrook/qbits/wormaudit/RdbmsTestSupport.java

@@ -127,10 +127,10 @@ class RdbmsTestSupport
                      case DATE_TIME -> "timestamp(6) with time zone";
                      default -> "text";
                   };
-            if(field.getName().equals("id"))
+            if(field.getName().equals(table.getPrimaryKeyField()))
             {
                type +=
-                        table.getName().equals("worm_audit_chain")
+                        table.getName().equals("worm_audit_chain") || table.getName().equals("worm_audit_chain_detail")
                               ? " PRIMARY KEY"
                               : " GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY";
             }

src/test/java/com/kingsrook/qbits/wormaudit/RecordIntegrationTest.java

@@ -53,7 +53,7 @@ class RecordIntegrationTest
    {
       var instance = init(new WormAuditQBitConfig());
       Map<String, String> formats = Map.of(
-         WormAuditTables.CHAIN, "Link #%s", WormAuditTables.CHECKPOINT, "Checkpoint #%s",
+         WormAuditTables.CHAIN, "Link #%s", WormAuditTables.CHAIN_DETAIL, "Detail #%s (link %s)", WormAuditTables.CHECKPOINT, "Checkpoint #%s",
          WormAuditTables.EXPORT, "Export #%s (links %s–%s)", WormAuditTables.VERIFICATION, "Verification #%s (%s)",
          WormAuditTables.PURGE, "Purge #%s (%s)", WormAuditTables.SETTINGS, "WORM Audit Settings");
       for(String name : WormAuditTables.ALL)

src/test/java/com/kingsrook/qbits/wormaudit/TokenRequiredAuthenticationModule.java

@@ -0,0 +1,72 @@
+/*
+ * 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.wormaudit;
+
+
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import com.kingsrook.qqq.backend.core.exceptions.QAuthenticationException;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.model.session.QSystemUserSession;
+import com.kingsrook.qqq.backend.core.modules.authentication.QAuthenticationModuleInterface;
+
+
+/*******************************************************************************
+ ** Test authentication module that behaves like the token-based core modules
+ ** (OAuth2, Auth0): a QSystemUserSession is always valid, a session it created
+ ** is valid, and any other session has no token and fails validation with the
+ ** same message the OAuth2 module produces (JWTDecodeException: The token is
+ ** null).
+ *******************************************************************************/
+public class TokenRequiredAuthenticationModule implements QAuthenticationModuleInterface
+{
+   private static final Set<String> ISSUED = ConcurrentHashMap.newKeySet();
+
+
+
+   /*******************************************************************************
+    ** Issue a session that carries a token.
+    *******************************************************************************/
+   @Override
+   public QSession createSession(QInstance qInstance, Map<String, String> context) throws QAuthenticationException
+   {
+      QSession session = new QSession();
+      ISSUED.add(session.getUuid());
+      return (session);
+   }
+
+
+
+   /*******************************************************************************
+    ** Accept the system user and issued sessions; reject tokenless sessions.
+    *******************************************************************************/
+   @Override
+   public boolean isSessionValid(QInstance instance, QSession session)
+   {
+      if(session instanceof QSystemUserSession || ISSUED.contains(session.getUuid()))
+      {
+         return (true);
+      }
+      throw (new IllegalStateException("The token is null"));
+   }
+}

src/test/java/com/kingsrook/qbits/wormaudit/TokenRequiredHostTest.java

@@ -0,0 +1,153 @@
+/*
+ * 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.wormaudit;
+
+
+import java.util.List;
+import com.kingsrook.qbits.wormaudit.chain.AuditStore;
+import com.kingsrook.qbits.wormaudit.model.WormAuditTables;
+import com.kingsrook.qbits.wormaudit.processes.ExportProcesses;
+import com.kingsrook.qbits.wormaudit.processes.WormAuditProcesses;
+import com.kingsrook.qqq.backend.core.context.QContext;
+import com.kingsrook.qqq.backend.core.model.data.QRecord;
+import com.kingsrook.qqq.backend.core.model.metadata.QAuthenticationType;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.authentication.QAuthenticationMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.code.QCodeReference;
+import com.kingsrook.qqq.backend.core.model.session.QSession;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryRecordStore;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+
+/*******************************************************************************
+ ** Regression: production must work on a host whose default authentication
+ ** rejects tokenless sessions (kof22 uses OAuth2). produce() reads the settings
+ ** singleton for the startup schedules; under a plain new QSession() that read
+ ** failed with "The token is null" and the host could not start.
+ *******************************************************************************/
+class TokenRequiredHostTest
+{
+   /*******************************************************************************
+    ** Clear the context and the memory store.
+    *******************************************************************************/
+   @AfterEach
+   void tearDown()
+   {
+      MemoryRecordStore.fullReset();
+      QContext.clear();
+   }
+
+
+
+   /*******************************************************************************
+    ** The test module really rejects a tokenless session, so the tests below
+    ** prove the qbit does not use one.
+    *******************************************************************************/
+   @Test
+   void moduleRejectsTokenlessSession() throws Exception
+   {
+      MemoryRecordStore.fullReset();
+      QInstance instance = tokenRequiredHost(new WormAuditQBitConfig());
+      QContext.init(instance, new QSession());
+      assertThatThrownBy(() -> new AuditStore(null).get(WormAuditTables.SETTINGS, 1))
+         .hasStackTraceContaining("The token is null");
+   }
+
+
+
+   /*******************************************************************************
+    ** produce() succeeds and still honors the persisted settings row for the
+    ** startup schedules, read as the system user.
+    *******************************************************************************/
+   @Test
+   void produceReadsPersistedSettingsAsSystemUser() throws Exception
+   {
+      ////////////////////////////////////////////////////////////////////
+      // first start: an anonymous host stores an edited settings row //
+      ////////////////////////////////////////////////////////////////////
+      MemoryRecordStore.fullReset();
+      WormAuditQBitConfig seedConfig = new WormAuditQBitConfig();
+      QInstance seedHost = WormAuditTestSupport.core(memory());
+      new WormAuditQBitProducer().withQBitConfig(seedConfig.withBackendName("memory")).produce(seedHost);
+      QContext.init(seedHost, new QSession());
+      QRecord row = WormAuditSettings.defaults(seedConfig)
+         .withValue("chain_interval_seconds", 45)
+         .withValue("verify_cron", "0 15 4 * * ?")
+         .withValue("export_cron", "0 45 1 * * ?");
+      new AuditStore(null).insert(WormAuditTables.SETTINGS, List.of(row));
+      QContext.clear();
+
+      //////////////////////////////////////////////////////////////////////
+      // restart on a token-required host, with no session in context, as //
+      // when a host builds its instance before any request arrives       //
+      //////////////////////////////////////////////////////////////////////
+      QInstance instance = tokenRequiredHost(new WormAuditQBitConfig());
+
+      assertThat(instance.getProcess(WormAuditProcesses.CHAIN).getSchedule().getRepeatSeconds()).isEqualTo(45);
+      assertThat(instance.getProcess(WormAuditProcesses.DAILY).getSchedule().getCronExpression()).isEqualTo("0 15 4 * * ?");
+      assertThat(instance.getProcess(ExportProcesses.DAILY).getSchedule().getCronExpression()).isEqualTo("0 45 1 * * ?");
+      assertThat(QContext.getQSession()).isNull();
+   }
+
+
+
+   /*******************************************************************************
+    ** With no settings row, production falls back to the configured values.
+    *******************************************************************************/
+   @Test
+   void produceFallsBackToConfigWithoutRow() throws Exception
+   {
+      MemoryRecordStore.fullReset();
+      QInstance instance = tokenRequiredHost(new WormAuditQBitConfig().withChainIntervalSeconds(17));
+      assertThat(instance.getProcess(WormAuditProcesses.CHAIN).getSchedule().getRepeatSeconds()).isEqualTo(17);
+   }
+
+
+
+   /*******************************************************************************
+    ** Build a host whose default authentication requires a token, and produce.
+    *******************************************************************************/
+   private static QInstance tokenRequiredHost(WormAuditQBitConfig config) throws Exception
+   {
+      QInstance instance = WormAuditTestSupport.core(memory());
+      instance.withInstanceDefaultAuthentication(new QAuthenticationMetaData()
+         .withName("token-required")
+         .withType(QAuthenticationType.MOCK)
+         .withModuleCodeReference(new QCodeReference(TokenRequiredAuthenticationModule.class)));
+      new WormAuditQBitProducer().withQBitConfig(config.withBackendName("memory")).produce(instance);
+      return (instance);
+   }
+
+
+
+   /*******************************************************************************
+    ** The memory backend.
+    *******************************************************************************/
+   private static QBackendMetaData memory()
+   {
+      return (new QBackendMetaData().withName("memory").withBackendType(MemoryBackendModule.class));
+   }
+}

src/test/java/com/kingsrook/qbits/wormaudit/WormAuditHomeTest.java

@@ -270,7 +270,7 @@ class WormAuditHomeTest
       assertThat(instance.getApp("evidence").getWidgets()).containsExactly(WormAuditApp.HOME_WIDGET);
       assertThat(instance.getApp("evidence").getChildren()).extracting(c -> c.getName()).containsExactlyElementsOf(WormAuditTables.NAV);
       assertThat(instance.getApp("evidence").getChildren()).extracting(c -> c.getLabel())
-         .containsExactly("Verifications", "Sealed Exports", "Checkpoints", "Audit Chain", "Purges", "Settings");
+         .containsExactly("Verifications", "Sealed Exports", "Checkpoints", "Audit Chain", "Detail Coverage", "Purges", "Settings");
       assertThat(instance.getWidget(WormAuditApp.HOME_WIDGET).getGridColumns()).isEqualTo(12);
       assertThat(instance.getWidget(WormAuditApp.HOME_WIDGET).getIsCard()).isFalse();
       assertThat(instance.getWidget(WormAuditApp.HEALTH_CARD_WIDGET).getIsCard()).isTrue();
@@ -278,6 +278,29 @@ class WormAuditHomeTest
    }
 
 
+   /*******************************************************************************
+    ** Embedded widgets retain direct authorization without a standalone app.
+    *******************************************************************************/
+   @Test
+   void embeddedWidgetsRequireReadPermission() throws Exception
+   {
+      var config = ChainVerificationTest.signed().withIncludeApp(false).withAppName("audit").withAppLabel("Audit");
+      var instance = init(config);
+      assertThat(instance.getApp("worm_audit")).isNull();
+      assertThat(instance.getApp("audit")).isNull();
+      var input = new RenderWidgetInput().withWidgetMetaData(instance.getWidget(WormAuditApp.HEALTH_CARD_WIDGET));
+      QContext.getQSession().withPermissions("admin.hasAccess");
+      assertThatThrownBy(() -> new WormAuditApp.HealthCardRenderer().render(input)).isInstanceOf(QPermissionDeniedException.class);
+      assertThatThrownBy(() -> new WormAuditApp.HomeRenderer().render(input)).isInstanceOf(QPermissionDeniedException.class);
+      QContext.getQSession().withPermissions(WormAuditTables.CHAIN + ".read");
+      String card = ((RawHTML) new WormAuditApp.HealthCardRenderer().render(input).getWidgetData()).getHtml();
+      assertThat(card).contains("href=\"/app/audit\"");
+      assertThat(new WormAuditApp.HomeRenderer().render(input).getWidgetData()).isInstanceOf(RawHTML.class);
+      QContext.setQSession(new QSession());
+      assertThatThrownBy(() -> new WormAuditApp.HealthCardRenderer().render(input)).isInstanceOf(QPermissionDeniedException.class);
+   }
+
+
    /*******************************************************************************
     ** Exercise the real SQL anti-join without Docker, including its exact boundary.
     *******************************************************************************/

src/test/java/com/kingsrook/qbits/wormaudit/WormAuditQBitProducerTest.java

@@ -21,10 +21,15 @@
 package com.kingsrook.qbits.wormaudit;
 
 import java.util.List;
+import com.kingsrook.qbits.wormaudit.app.WormAuditApp;
+import com.kingsrook.qbits.wormaudit.model.WormAuditHelp;
 import com.kingsrook.qbits.wormaudit.ddl.WormAuditDdl;
 import com.kingsrook.qbits.wormaudit.model.WormAuditTables;
 import com.kingsrook.qbits.wormaudit.processes.WormAuditProcesses;
 import com.kingsrook.qqq.backend.core.model.metadata.QInstance;
+import com.kingsrook.qqq.backend.core.model.metadata.QBackendMetaData;
+import com.kingsrook.qqq.backend.core.model.metadata.layout.QAppMetaData;
+import com.kingsrook.qqq.backend.core.modules.backend.implementations.memory.MemoryBackendModule;
 import com.kingsrook.qqq.backend.core.model.metadata.tables.Capability;
 import org.junit.jupiter.api.Test;
 import static org.assertj.core.api.Assertions.assertThat;
@@ -43,6 +48,7 @@ class WormAuditQBitProducerTest
    {
       var config = new WormAuditQBitConfig();
       QInstance instance = WormAuditTestSupport.init(config);
+      assertThat(config.getIncludeApp()).isTrue();
       assertThat(config.getRetentionYears()).isEqualTo(6);
       assertThat(config.getTableMetaDataCustomizer()).isNull();
       assertThat(config.getSigningKey()).isNull();
@@ -160,4 +166,36 @@ class WormAuditQBitProducerTest
                   () -> WormAuditDdl.resolveNames(new WormAuditQBitConfig(), new QInstance()))
             .isInstanceOf(IllegalArgumentException.class);
    }
+
+
+   /*******************************************************************************
+    ** Embedded production keeps the host app intact and registers all evidence metadata.
+    *******************************************************************************/
+   @Test
+   void embeddedMetadataPreservesHostApp() throws Exception
+   {
+      var config = new WormAuditQBitConfig().withBackendName("memory").withIncludeApp(false).withAppName("audit");
+      var instance = WormAuditTestSupport.core(new QBackendMetaData().withName("memory").withBackendType(MemoryBackendModule.class));
+      var host = new QAppMetaData().withName("audit").withLabel("Combined Audit").withWidgets(List.of("hostSummary"));
+      instance.addApp(host);
+      new WormAuditQBitProducer().withQBitConfig(config).produce(instance);
+      assertThat(config.getIncludeApp()).isFalse();
+      assertThat(instance.getApp("worm_audit")).isNull();
+      assertThat(instance.getApp("audit")).isSameAs(host);
+      assertThat(host.getLabel()).isEqualTo("Combined Audit");
+      assertThat(host.getWidgets()).containsExactly("hostSummary");
+      assertThat(host.getHelpContent()).isNullOrEmpty();
+      assertThat(instance.getWidget(WormAuditApp.HEALTH_CARD_WIDGET)).isNotNull();
+      assertThat(instance.getWidget(WormAuditApp.HOME_WIDGET)).isNotNull();
+      for(String name : WormAuditTables.ALL)
+      {
+         assertThat(instance.getTable(name).getHelpContent()).containsKey(WormAuditHelp.SLOT);
+         assertThat(instance.getTable(name).getPermissionRules().getPermissionBaseName()).isEqualTo(name);
+      }
+      assertThat(instance.getTable(WormAuditTables.CHAIN).getDisabledCapabilities()).contains(Capability.TABLE_UPDATE, Capability.TABLE_DELETE, Capability.TABLE_INSERT);
+      assertThat(instance.getTable("audit").getDisabledCapabilities()).contains(Capability.TABLE_UPDATE, Capability.TABLE_DELETE);
+      assertThat(instance.getProcess(WormAuditProcesses.VERIFY)).isNotNull();
+      assertThat(instance.getProcess(WormAuditProcesses.CHAIN).getSchedule().getRepeatSeconds()).isEqualTo(config.getChainIntervalSeconds());
+      assertThat(instance.getProcess(WormAuditProcesses.DAILY).getSchedule().getCronExpression()).isEqualTo(config.getVerifyCron());
+   }
 }

src/test/java/com/kingsrook/qbits/wormaudit/WormAuditSettingsSeedTest.java

@@ -107,7 +107,12 @@ class WormAuditSettingsSeedTest
       Matcher matcher = Pattern.compile("(?m)^INSERT INTO\\s+(\\S+)").matcher(ddl);
       while(matcher.find())
       {
-         assertThat(matcher.group(1)).as("every top-level INSERT in the script targets the settings table, not a log").isEqualTo("public.worm_audit_settings");
+         //////////////////////////////////////////////////////////////////////////
+         // worm-audit:13 backfills detail coverage from links that already exist: //
+         // derived from the chain, never seeded content.                         //
+         //////////////////////////////////////////////////////////////////////////
+         assertThat(matcher.group(1)).as("every top-level INSERT in the script targets the settings table, not a log")
+            .isIn("public.worm_audit_settings", "public.worm_audit_chain_detail(detail_id,");
       }
    }
 

src/test/java/com/kingsrook/qbits/wormaudit/WormAuditTablesGuardrailTest.java

@@ -62,6 +62,7 @@ class WormAuditTablesGuardrailTest
     *******************************************************************************/
    static final Map<String, String> STARTS_EMPTY = Map.of(
       WormAuditTables.CHAIN, "log: the chain job links each settled audit record",
+      WormAuditTables.CHAIN_DETAIL, "log: the chain trigger (or writer) records each detail a new link covers",
       WormAuditTables.CHECKPOINT, "log: the chain job signs a range every N links or minutes",
       WormAuditTables.EXPORT, "log: the daily export or Export now seals a range",
       WormAuditTables.VERIFICATION, "log: every verification run, manual or daily",

src/test/java/com/kingsrook/qbits/wormaudit/chain/SweepFloorsTest.java

@@ -0,0 +1,104 @@
+/*
+ * 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.wormaudit.chain;
+
+
+import java.time.Instant;
+import com.kingsrook.qbits.wormaudit.WormAuditQBitConfig;
+import org.junit.jupiter.api.Test;
+import static org.assertj.core.api.Assertions.assertThat;
+
+
+/*******************************************************************************
+ ** Floors move only after complete sweeps, only to ids observed before the
+ ** horizon, and reset when the tables shrink below them.
+ *******************************************************************************/
+class SweepFloorsTest
+{
+   private static final Instant T0 = Instant.parse("2026-10-15T00:00:00Z");
+
+
+
+   /*******************************************************************************
+    ** Floors follow the newest observation at or before the horizon.
+    *******************************************************************************/
+   @Test
+   void floorsAdvanceOnlyPastSettledObservationsOfCompleteSweeps()
+   {
+      SweepFloors floors = SweepFloors.of(new WormAuditQBitConfig());
+      var first = floors.observe(T0, 10, 100);
+      floors.swept(first, T0.minusSeconds(1), true, true, null);
+      assertThat(floors.audit()).as("nothing observed before the horizon yet").isZero();
+
+      var second = floors.observe(T0.plusSeconds(30), 12, 130);
+      floors.swept(second, T0.plusSeconds(10), false, true, null);
+      assertThat(floors.audit()).as("the header sweep hit its limit").isZero();
+      assertThat(floors.detail()).isEqualTo(100);
+
+      var third = floors.observe(T0.plusSeconds(60), 15, 160);
+      floors.swept(third, T0.plusSeconds(45), true, false, null);
+      assertThat(floors.audit()).isEqualTo(12);
+      assertThat(floors.detail()).as("the late-detail sweep hit its limit").isEqualTo(100);
+
+      var fourth = floors.observe(T0.plusSeconds(90), 15, 160);
+      floors.swept(fourth, T0.plusSeconds(90), true, true, null);
+      assertThat(floors.audit()).isEqualTo(15);
+      assertThat(floors.detail()).isEqualTo(160);
+
+      var shrunk = floors.observe(T0.plusSeconds(120), 3, 7);
+      assertThat(floors.audit()).as("tables emptied below the floor: start over").isZero();
+      assertThat(floors.detail()).isZero();
+      floors.swept(shrunk, T0.plusSeconds(100), true, true, null);
+      assertThat(floors.audit()).isZero();
+   }
+
+
+
+   /*******************************************************************************
+    ** A full sweep is due first, then hourly after each complete one.
+    *******************************************************************************/
+   @Test
+   void fullSweepIsDueFirstAndHourly()
+   {
+      SweepFloors floors = SweepFloors.of(new WormAuditQBitConfig());
+      assertThat(floors.fullSweepDue(T0)).isTrue();
+      var first = floors.observe(T0, 1, 1);
+      floors.swept(first, T0, true, false, T0);
+      assertThat(floors.fullSweepDue(T0.plusSeconds(30))).as("an incomplete full sweep repeats").isTrue();
+      var second = floors.observe(T0.plusSeconds(30), 1, 1);
+      floors.swept(second, T0, true, true, T0.plusSeconds(30));
+      assertThat(floors.fullSweepDue(T0.plusSeconds(60))).isFalse();
+      assertThat(floors.fullSweepDue(T0.plusSeconds(30).plus(SweepFloors.FULL_SWEEP_EVERY))).isTrue();
+      assertThat(floors.fullSweepDue(T0)).as("clock moved backwards").isTrue();
+   }
+
+
+
+   /*******************************************************************************
+    ** Each installed config has its own floors.
+    *******************************************************************************/
+   @Test
+   void floorsArePerConfig()
+   {
+      var config = new WormAuditQBitConfig();
+      assertThat(SweepFloors.of(config)).isSameAs(SweepFloors.of(config)).isNotSameAs(SweepFloors.of(new WormAuditQBitConfig()));
+   }
+}