From a799907866df94cf87400df01866c6890e27db64 Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Thu, 20 Aug 2026 14:04:30 +0000
Subject: [PATCH 1/6] feat(guard): detect NATS consumers that die while the pod
 stays Running
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A rolling node maintenance restarted NATS and left three Bento consumers
(latest-kv, alarms, entity-greptimedb) holding dead subscriptions. All three
reported Running 1/1 for hours; the halt was found only when a human ran
`nats consumer info` by hand. Neither /ping nor /ready detects a subscription
that is dead underneath a live connection, and greptimedb-freshness-guard stays
green through it because a different consumer keeps writing rows to GreptimeDB.

Adds nats-consumer-guard (CronJob + scripts ConfigMap) covering every JetStream
durable, including the alarm-materializer one, which is created by the Go client
and whose Deployment has no probes at all.

Backlog depth is deliberately NOT the signal. latest-kv runs max_ack_pending=1
as a single-writer serialization mechanism, so under load it holds a large and
often growing backlog while working perfectly — measured at a sustained 6,729
while its ack floor advanced 61,179 -> 100,526. A depth threshold alerts there,
every tick. The guard checks push_bound (is anything subscribed to the deliver
subject) and whether ack_floor.consumer_seq MOVES across two samples, and
requires the fault in both samples so a rolling upgrade does not trip it.

Verified against a real NATS 2.10.26 across all four branches, then in-cluster:
a four-minute NATS outage left all five consumers unbound with every pod still
Running 1/1 and its restart count unchanged, the guard failed the Job naming
each one, and `rollout restart` returned the next tick to green. A single clean
pod restart does NOT reproduce it — every consumer reconnects — which is why the
failure looked intermittent.

Also closes two landmines the guard would have surfaced:
- the questdb durable was created when enabled but never removed when disabled,
  leaving an unbound consumer on a capped stream (it once reached 106k pending
  and never drained). Removed at the source rather than ignored.
- every shipped overlay set latestKv.replicas: 2 against the documented
  single-writer invariant, including the example templates new deployments copy
  from. Pinned to 1 and covered by a test over all tracked values files.

Documents the --reuse-values trap: it reuses the previous release's COMPUTED
values, so new chart defaults are silently shadowed — which nearly shipped
maxAckPending 1024 against the required 1.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 docs/OPERATIONS.md                            |  96 ++++++++
 docs/adr/0003-nats-consumer-progress-guard.md | 184 ++++++++++++++
 .../nats-consumer-guard-scripts.yaml          | 226 ++++++++++++++++++
 .../templates/nats-consumer-guard.yaml        | 138 +++++++++++
 k8s/helm/thingsflow/templates/nats.yaml       |   9 +
 .../thingsflow/values-cluster.example.yaml    |   6 +-
 k8s/helm/thingsflow/values-demo.yaml          |   6 +-
 k8s/helm/thingsflow/values-pilot.example.yaml |   6 +-
 k8s/helm/thingsflow/values.yaml               |  21 ++
 mkdocs.yml                                    |   1 +
 tools/python/test_consumer_guard_progress.py  | 203 ++++++++++++++++
 11 files changed, 893 insertions(+), 3 deletions(-)
 create mode 100644 docs/adr/0003-nats-consumer-progress-guard.md
 create mode 100644 k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml
 create mode 100644 k8s/helm/thingsflow/templates/nats-consumer-guard.yaml
 create mode 100644 tools/python/test_consumer_guard_progress.py

diff --git a/docs/OPERATIONS.md b/docs/OPERATIONS.md
index 3d70519..adeaa5b 100644
--- a/docs/OPERATIONS.md
+++ b/docs/OPERATIONS.md
@@ -596,6 +596,44 @@ Go/no-go criteria:
 - logs contain no bearer tokens, MQTT passwords, OIDC secrets, or raw payload
   dumps.
 
+## Upgrading: never `--reuse-values`
+
+`--reuse-values` reuses the **computed** values of the previous release, not the
+user-supplied ones. Every default the chart has changed since that release is
+silently shadowed by the old computed value, so an upgrade that exists precisely
+to ship a new default ships the old one and reports success.
+
+This is not hypothetical. `natsDataPlane.latestKv.maxAckPending` was changed
+from `1024` to `1` because it became the doc-merge consumer's serialization
+mechanism — with `--reuse-values`, a production dry-run still rendered `1024`,
+which NATS would have rejected at bind time (`configuration requests max ack
+pending to be 1024, but consumer's value is 1`), leaving the consumer unable to
+attach at all.
+
+Upgrade by passing your overlay explicitly instead (start from one of the
+tracked `values-*.example.yaml` templates):
+
+```bash
+helm upgrade thingsflow ./k8s/helm/thingsflow -n thingsflow -f my-overlay.yaml
+```
+
+If the release carries overrides that live only in the cluster, capture them
+first rather than reusing them blindly — this keeps the operator's intent and
+still picks up new chart defaults:
+
+```bash
+helm -n thingsflow get values thingsflow | tail -n +2 > /tmp/live-values.yaml
+helm upgrade thingsflow ./k8s/helm/thingsflow -n thingsflow -f /tmp/live-values.yaml
+```
+
+Always confirm what actually rendered before trusting the upgrade:
+
+```bash
+helm -n thingsflow get values thingsflow --all | grep -A3 latestKv
+kubectl -n thingsflow get deploy thingsflow-nats-latest-kv \
+  -o jsonpath='{.spec.template.spec.containers[0].env[?(@.name=="LATEST_KV_MAX_ACK_PENDING")].value}{"\n"}'
+```
+
 ## Routine Checks
 
 Daily pilot checks:
@@ -649,6 +687,7 @@ All commands in this playbook assume the canonical install (`helm upgrade
 |---|---|---|---|---|
 | `greptimedb-ttl-guard` | GreptimeDB DB/table TTL present and healthy | `0 * * * *` (hourly) | `retention.greptimedb.schedule` | `thingsflow_greptimedb_ttl_ok`, `thingsflow_greptimedb_ttl_drift_repaired` |
 | `greptimedb-freshness-guard` | telemetry still landing in GreptimeDB | `*/5 * * * *` | `monitoring.freshnessGuard.schedule` | `thingsflow_greptimedb_write_stale` |
+| `nats-consumer-guard` | every JetStream durable is bound and draining | `*/5 * * * *` | `monitoring.consumerGuard.schedule` | `thingsflow_nats_consumer_progress_ok` |
 | `device-silence-guard` | individual devices that stopped reporting | `*/5 * * * *` (guard off by default) | `alarms.deviceSilence.schedule` | none — emits `DeviceSilent` alarm intents; a failed Job means the guard itself broke |
 | NATS stream drift-verify | live JetStream stream/consumer config vs chart intent, plus the PVC budget | every `helm install`/`upgrade` (hook, not a CronJob) | n/a — runs with each upgrade | `thingsflow_nats_stream_ok` |
 | `postgres-alarm-retention` | sweep of cleared+acked alarms past the window | `0 3 * * 0` (Sun 03:00) | `retention.alarm.schedule` | `thingsflow_postgres_alarm_deleted_total` |
@@ -780,6 +819,63 @@ it too if entity rows are also stale.) If NATS itself lost its streams, follow
 If GreptimeDB is unreachable, fix that first — the guard cannot distinguish
 further until it can query the store.
 
+### nats-consumer-guard
+
+**The failure it exists for.** A rolling node maintenance restarted NATS and
+left three Bento consumers (`latest-kv`, `alarms`, `entity-greptimedb`) holding
+dead subscriptions. All three reported `Running 1/1` for hours. Kubernetes could
+not see it: every data-plane Deployment sets `livenessProbe: /ping` and
+`readinessProbe: /ready`, and both stay green while the pod spins on
+`nats: connection closed` — `/ready` is satisfied by a `nats.Conn` handle that is
+dead at the *subscription* level, and `/ping` only pings Bento's HTTP server.
+
+This is also the failure `greptimedb-freshness-guard` structurally cannot catch:
+that guard asks whether the *store* is receiving anything, and it stays green
+while `latest-kv` is dead because a different consumer keeps writing rows.
+
+**What it checks, and why not backlog.** Queue depth is the wrong signal.
+`latest-kv` runs `max_ack_pending: 1` as a deliberate single-writer
+serialization mechanism, so under load it holds a large and often *growing*
+backlog while working perfectly — measured at a sustained backlog of 6,729 while
+its ack floor advanced 61,179 → 100,526. A depth threshold alerts there. The
+guard uses two signals instead:
+
+1. `push_bound` — the server-side boolean saying whether anything is subscribed
+   to the durable's deliver subject right now. This is what the nats CLI renders
+   as `Active Interest: No interest`. Necessary but not sufficient: with more
+   than one replica in a deliver group, one dead pod still leaves it `true`.
+2. `ack_floor.consumer_seq` movement across two samples — what separates
+   "serialized but progressing" from "dead". (`consumer_seq`, not `stream_seq`:
+   TF_RAW is discard-old, so retention deleting messages under a stalled
+   consumer drags the stream-side floor forward and reads as false progress.)
+
+An alert requires the fault in **both** samples, so a `helm upgrade` rolling the
+Bento pods does not trip it.
+
+**Reading a failure.**
+
+```bash
+kubectl -n thingsflow logs -l app=nats-consumer-guard --tail=40
+```
+
+| Line | Meaning | Remedy |
+|---|---|---|
+| `no subscriber bound to deliver subject` | the classic silent stall — pod alive, subscription dead | `kubectl -n thingsflow rollout restart deploy/thingsflow-<consumer>` |
+| `ack floor is frozen ... bound but not draining` | subscribed but not acking: a wedged pipeline, a poison message against `maxDeliver`, or one dead pod in a multi-replica deliver group | check that consumer's pod logs before restarting |
+| `consumer could not be read (absent, or NATS refused)` | a durable the chart expects does not exist | re-run `helm upgrade` so the nats-bootstrap hook recreates it |
+| `NATS unreachable` | not evidence of health, deliberately alerts | fix NATS first |
+
+**Scope.** The guard covers every JetStream durable, including the
+`alarm-materializer` one (created by the Go client, and the only consumer with
+no Kubernetes probes at all). It cannot see the WebSocket journal fan-out
+(`flow-core/internal/ws/journal.go`) — that is a *core* NATS subscription with
+its own resubscribe loop and creates no JetStream consumer.
+
+An orphaned durable (one whose component was disabled but whose consumer was
+left behind) will alert forever, correctly: on a capped, discard-old stream an
+unattended consumer is not inert. Remove it at the source rather than adding it
+to `monitoring.consumerGuard.ignore`.
+
 ### device-silence-guard
 
 **Off by default** (`alarms.deviceSilence.enabled: false`): on a fleet still
diff --git a/docs/adr/0003-nats-consumer-progress-guard.md b/docs/adr/0003-nats-consumer-progress-guard.md
new file mode 100644
index 0000000..baedef2
--- /dev/null
+++ b/docs/adr/0003-nats-consumer-progress-guard.md
@@ -0,0 +1,184 @@
+# ADR 0003 — NATS consumer liveness: progress, not backlog
+
+- **Status:** Accepted
+- **Date:** 2026-08-20
+- **Context owners:** data plane (NATS/Bento), platform ops
+- **Relates to:** `k8s/helm/thingsflow/templates/nats-consumer-guard{,-scripts}.yaml`,
+  `greptimedb-freshness-guard.yaml`, `nats-stream-guard-scripts.yaml`, D-07
+
+## Context
+
+On 2026-08-19 a rolling kubelet restart across all four nodes of the production
+cluster restarted NATS. Three Bento consumers — `latest-kv`, `alarms` and
+`entity-greptimedb` — were left holding dead subscriptions. All three kept
+reporting `Running 1/1`. The halt was found only because a human ran
+`nats consumer info` by hand; `entity-greptimedb` had been stopped for roughly
+seven hours by then.
+
+This is a repeat, not a novelty. `benchmarks/FINDING-twin-state-localization.md`
+records the same shape after an earlier NATS OOM-restart: four consumer pods
+spinning on `level=error msg="Failed to read message: nats: connection closed"`
+**with zero reconnection attempts**, a backlog of 100,921 messages, and
+`Active Interest: No interest`.
+
+### Why nothing caught it
+
+**Kubernetes could not.** Every data-plane Deployment already sets
+`livenessProbe: /ping` and `readinessProbe: /ready`
+(`nats-data-plane-bento.yaml`). Both stayed green. `/ready` is satisfied by a
+`nats.Conn` handle that is dead at the *subscription* level; `/ping` only pings
+Bento's own HTTP server. Bento 1.8.1 cannot self-report this failure — that is
+an observation from the field evidence above, not an assumption: if the input
+had surfaced the dead subscription, the reader would have reconnected and
+`/ready` would have flipped the pods to `0/1`. Neither happened.
+
+**The existing guard could not.** `greptimedb-freshness-guard` already contains
+the right instinct — *"what separates the two is whether there is WORK
+WAITING"* — but it probes only the greptimedb durable (the one consumer that did
+**not** stall) and uses backlog as a *secondary* signal qualifying a GreptimeDB
+freshness query. When `latest-kv` dies, GreptimeDB keeps receiving rows from a
+different consumer, so that guard exits 0, green, correctly, and uselessly.
+
+## Decision
+
+Add a dedicated `nats-consumer-guard` CronJob covering **every** JetStream
+durable, following the established convention (D-07: the failed Job **is** the
+alert; no Prometheus, no ServiceMonitor, no push-gateway).
+
+### Backlog depth is the wrong signal — this is the load-bearing decision
+
+The obvious design is "alert when backlog exceeds a threshold, or grows between
+two samples." It is wrong, and it fails in the direction that makes a guard
+worthless.
+
+`latest-kv` runs `max_ack_pending: 1` as a deliberate single-writer
+serialization mechanism, not a throughput knob: the doc-merge must complete a
+whole `cache get → merge → cache set` before the next delivery. Under load it
+therefore holds a large — and frequently *growing* — backlog **while working
+perfectly**. Measured against a real NATS 2.10.26 server: a sustained backlog of
+6,729 messages while the ack floor advanced 61,179 → 100,526 in the same window.
+A depth-based rule reports that healthy consumer as dead on every tick. A
+`>=`-based "not shrinking" rule is worse still: a saturated-but-healthy consumer
+whose arrival rate equals its drain rate holds a *flat* non-zero backlog, which
+is the definition of a working queue at capacity.
+
+Two signals are used instead:
+
+1. **`push_bound`** — all durables are push consumers created with an explicit
+   `--target` and `--deliver-group`, so the server tracks whether anything is
+   subscribed to the deliver subject *right now*. It flips within milliseconds
+   of the subscription dying, needs one sample, and cannot false-positive on a
+   serialized consumer. This is the `Active Interest: No interest` from the
+   incident. It is **necessary but not sufficient**: with more than one replica
+   in a deliver group, one dead pod still leaves it `true` — and
+   `thingsflow-nats-greptimedb` runs two replicas in production.
+
+2. **`ack_floor.consumer_seq` movement** across two samples — what actually
+   separates "serialized but progressing" from "dead". Deliberately
+   `consumer_seq`, **not** `stream_seq`: TF_RAW is `discard old` with a
+   `max_bytes` cap, so retention deleting messages out from under a *stalled*
+   consumer drags its stream-side floor forward and would read as false
+   progress.
+
+Two samples are required rather than preferred: NATS 2.10.26 exposes no
+`ack_floor.last_active` field (verified against the pinned server — `ack_floor`
+carries only `consumer_seq` and `stream_seq`), so there is no stateless
+"seconds since last ack" available.
+
+An alert requires the fault in **both** samples. A `helm upgrade` rolls the
+Bento pods and leaves a brief window with no subscriber; firing on that would
+alert on every deployment, which is how operators learn to ignore a guard.
+
+### Alert only — no auto-remediation
+
+The chart contains no RBAC objects at all. Granting a scheduled Job standing
+patch authority over the data plane to restart Deployments is a far larger
+change than the guard, and it would be reviewed as one by anyone deploying this.
+Beyond that, a guard that fixes the problem and exits 0 destroys the alert
+surface D-07 is built on, and during a real NATS outage a self-remediating tick
+would restart every consumer every five minutes — a stateless CronJob has no
+memory with which to rate-limit itself, so a recoverable outage becomes a
+self-inflicted one.
+
+### Which consumers are checked
+
+Live enumeration (`stream ls` → `consumer ls` → `consumer info`) **union** a
+render-injected expected set. Live enumeration cannot drift from reality but is
+blind by construction to a consumer that *vanished*; the expected set covers
+exactly that. A generic Helm `range` over `.Values.natsDataPlane` would be
+wrong — that map mixes scalars with the consumer sub-maps, and real enablement
+depends on `historyStore`, which the sub-maps do not carry — so the expected set
+mirrors the same gates the Bento Deployments use, plus the `alarm-materializer`
+durable, which is created by the Go client, lives under a different values key,
+and whose Deployment has no probes at all.
+
+## Consequences
+
+- An orphaned durable (component disabled, consumer left behind) now alerts
+  forever, correctly: on a capped, discard-old stream an unattended consumer is
+  not inert. The `thingsflow-questdb-durable` orphan documented in `values.yaml`
+  (grew to 106k pending, never draining) is therefore removed at the source —
+  `templates/nats.yaml` gained an `else` branch that `consumer rm`s it when
+  `questdb.enabled` is false — rather than being added to an ignore list.
+- `monitoring.consumerGuard.ignore` exists as an explicit, reviewable escape
+  hatch, never a silent skip.
+- The guard cannot see the WebSocket journal fan-out
+  (`flow-core/internal/ws/journal.go`): that is a *core* NATS subscription with
+  its own resubscribe loop and creates no JetStream consumer. Stated here so
+  "every consumer is guarded" is not read as "every subscription is guarded."
+- **Every overlay shipped `latestKv.replicas: 2`**, contradicting the
+  single-writer invariant documented in `values.yaml` and
+  `files/bento-nats-latest-kv.yaml`. This was not one stale file: the cluster,
+  pilot and demo overlays all carried it, including the
+  `values-*.example.yaml` templates new deployments copy from, so every fresh
+  cluster inherited it. Production happened to be running 1, which is why it was
+  a landmine for the next clean upgrade rather than a live fault. All are now 1,
+  pinned by a test over every tracked values file.
+
+  Whether 2 would in fact be safe is **not settled** — `max_ack_pending` is a
+  property of the shared consumer, so the server withholds message N+1 until N
+  is acked regardless of which pod acked it, and Bento acks after the cache set.
+  That argument is plausible and untested; scaling out belongs to the
+  sharded-merge work already open as a milestone criterion, with a measurement.
+
+## Verification
+
+Against a real NATS 2.10.26 + nats-box 0.16.0, using the script extracted from
+the **rendered** chart, all four decision branches:
+
+| Scenario | Observed | Verdict |
+|---|---|---|
+| No subscriber, backlog 93,000 | `push_bound false` | ALERT |
+| Subscriber alive, never acking (wedged / dead replica in a group) | `push_bound true`, backlog 4,000, ack floor frozen at 0 | ALERT |
+| Expected durable absent | unreadable in both samples | ALERT |
+| Serialized `max_ack_pending: 1` under sustained load | backlog 10,974, ack floor 117,109 → 149,520 | **OK** |
+| Whole platform healthy under load | all five bound and progressing | OK, exit 0 |
+
+A silent defect was found this way and fixed before shipping: the list-splitting
+helper used `printf '%s'` without a trailing newline, so `while read` dropped the
+**last** element of every comma-separated list — and the last element of the
+rendered expected set is the `alarm-materializer` durable, the one consumer with
+no Kubernetes probes of any kind.
+
+### In-cluster reproduction
+
+A **single clean NATS pod restart did not reproduce the stall**: every consumer
+reconnected and the guard correctly reported health. That is itself worth
+recording — it is why the failure looked intermittent and hard to pin down.
+
+Scaling the NATS StatefulSet to zero for four minutes — long enough to exhaust
+the client's reconnect attempts, which is what a node-level disruption produces
+and a pod restart does not — reproduced it exactly:
+
+- All five consumers came back `push_bound false`.
+- Every Bento pod stayed `Running 1/1` with its restart count **unchanged**, so
+  Kubernetes reported the data plane as healthy throughout.
+- The guard failed its Job and named all five durables.
+
+`kubectl rollout restart` on the five Deployments restored every subscription
+and the next tick returned to `exit 0`, confirming the remedy in the operations
+playbook.
+
+The reproduction recipe matters as much as the guard: it is the only known way
+to exercise this failure on demand, and a pod restart — the obvious thing to
+try — does not do it.
diff --git a/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml b/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml
new file mode 100644
index 0000000..376bfc3
--- /dev/null
+++ b/k8s/helm/thingsflow/templates/nats-consumer-guard-scripts.yaml
@@ -0,0 +1,226 @@
+{{- $cg := (.Values.monitoring | default dict).consumerGuard | default dict }}
+{{- if and (.Values.nats.enabled | default true) ($cg.enabled | default true) }}
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard-scripts
+  namespace: {{ .Release.Namespace | default "thingsflow" }}
+  labels:
+    app: nats-consumer-guard
+data:
+  # check.sh -- liveness guard for every JetStream durable consumer on the platform.
+  #
+  # WHY THIS EXISTS. A rolling node maintenance restarted NATS and left three
+  # Bento consumers (latest-kv, alarms, entity-greptimedb) with dead
+  # subscriptions. All three kept reporting Running 1/1 for hours; the halt was
+  # found only when a human ran `nats consumer info` by hand. Bento cannot
+  # self-report this: every data-plane Deployment already sets livenessProbe
+  # /ping and readinessProbe /ready (nats-data-plane-bento.yaml), and BOTH stayed
+  # green while the pods span on `nats: connection closed` with zero reconnection
+  # attempts (benchmarks/FINDING-twin-state-localization.md).
+  #
+  # WHY NOT BACKLOG DEPTH -- read this before "simplifying" the check. Queue depth
+  # is the WRONG signal and a threshold on it would make this guard useless:
+  # latest-kv runs max_ack_pending=1 as a deliberate single-writer serialization
+  # mechanism (values.yaml), so under load it holds a large and often GROWING
+  # backlog while working perfectly. Measured in a NATS 2.10.26 sandbox: backlog
+  # 6,729 sustained while the ack floor advanced 61,179 -> 100,526. A depth-based
+  # guard alerts there. This one does not.
+  #
+  # THE TWO SIGNALS, in order:
+  #   1. push_bound -- all durables are push consumers with an explicit --target
+  #      and --deliver-group (nats.yaml), so the server tracks whether ANYTHING is
+  #      subscribed to the deliver subject right now. It flips within ms of the
+  #      subscription dying. This is the `Active Interest: No interest` seen in the
+  #      incident. Necessary but NOT sufficient: with >1 replica in one deliver
+  #      group, one dead pod still leaves push_bound true.
+  #   2. ack_floor.consumer_seq movement -- what separates "serialized but
+  #      progressing" from "dead". Requires two samples; NATS 2.10.26 has no
+  #      last_active field on ack_floor (verified against the pinned server), so a
+  #      one-shot version of this check is not available.
+  #
+  #      consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes cap,
+  #      so retention deleting messages out from under a STALLED consumer drags its
+  #      stream-side floor forward and reads as false progress.
+  #
+  # BOTH-SAMPLE RULE. An alert requires the fault in BOTH samples. A helm upgrade
+  # rolls the Bento pods and leaves a brief window with no subscriber; firing on
+  # that would alert on every deployment, which is how operators learn to ignore a
+  # guard (the lesson greptimedb-freshness-guard.yaml already documents).
+  #
+  # HEALTH SIGNAL (failed Job IS the alert -- D-07, no Prometheus):
+  #   thingsflow_nats_consumer_progress_ok{stream=,durable=} 1|0   per consumer
+  #   thingsflow_nats_consumer_progress_ok 1|0                     aggregate
+  # Exit: 0 only when every checked consumer is bound and progressing; else 1.
+  #
+  # NO SCRAPE INFRA (Pitfall 1): there is no Prometheus scrape target for a Job
+  # stdout line. Do NOT add a ServiceMonitor, a push-gateway, or a scrape
+  # annotation -- the metric is consumed from Job logs / exit status.
+  #
+  # SCOPE -- say this out loud so nobody reads "everything is guarded" into it:
+  # the WS journal fan-out (flow-core/internal/ws/journal.go) is a CORE NATS
+  # subscription with its own resubscribe loop and creates no JetStream consumer
+  # at all. No consumer guard can see it.
+  #
+  # ENV CONTRACT -- the CronJob MUST inject:
+  #   NATS_URL             internal NATS URL (nats://[user:pass@]<fullname>-nats:4222)
+  #   GUARD_EXPECTED       comma-separated "stream|durable" pairs the chart intends to exist
+  #   GUARD_IGNORE         comma-separated "stream|durable" pairs to skip (explicit, reviewable)
+  #   GUARD_SAMPLE_SLEEP   seconds between the two samples
+  #   GUARD_NATS_TIMEOUT   per-request timeout for every nats invocation
+  #
+  # POSIX /bin/sh only -- no bashisms (no [[ ]], no arrays, no `local`).
+  check.sh: |
+    # POSIX /bin/sh only -- no bashisms (no [[ ]], no arrays, no `local`).
+    set -u
+
+    SLEEP="${GUARD_SAMPLE_SLEEP:-60}"
+    TMO="${GUARD_NATS_TIMEOUT:-5s}"
+    EXPECTED="${GUARD_EXPECTED:-}"
+    IGNORE="${GUARD_IGNORE:-}"
+    STATE="${GUARD_STATE_DIR:-/tmp/consumer-guard}"
+
+    mkdir -p "$STATE" || exit 2
+    ok=1
+
+    n() { nats --server "$NATS_URL" --timeout "$TMO" "$@" 2>/dev/null; }
+    key() { printf '%s' "$1_$2" | tr -c 'A-Za-z0-9_' '_'; }
+    # printf '%s\n', NOT '%s': without the trailing newline `while read` drops the
+    # LAST element of every list -- silently, and the last element of the rendered
+    # GUARD_EXPECTED is the alarm-materializer durable, the one consumer with no
+    # Kubernetes probes of any kind. Caught in the sandbox, not in production.
+    listify() { printf '%s\n' "$1" | tr ',' '\n' | sed '/^[[:space:]]*$/d'; }
+
+    # Reads one consumer's state as TSV: kind push_bound backlog ack_floor delivered redelivered
+    probe() {
+      n consumer info "$1" "$2" --json | jq -r '
+        [ (if ((.config.deliver_subject // "") | length) > 0 then "push" else "pull" end),
+          (.push_bound // false),
+          ((.num_pending // 0) + (.num_ack_pending // 0)),
+          (.ack_floor.consumer_seq // 0),
+          (.delivered.consumer_seq // 0),
+          (.num_redelivered // 0)
+        ] | @tsv' 2>/dev/null
+    }
+
+    # --- fail-safe: unreachable NATS is ONE clear alert, not six confusing ones ---
+    attempts=0
+    until n server check connection >/dev/null 2>&1; do
+      attempts=$((attempts + 1))
+      [ "$attempts" -ge 3 ] && {
+        echo "thingsflow_nats_consumer_progress_ok 0"
+        echo "[consumer-guard] ALERT: NATS unreachable at ${NATS_URL} after ${attempts} attempts -- consumer liveness cannot be established (not evidence of health)"
+        exit 1
+      }
+      sleep 2
+    done
+
+    # --- build the check set: live enumeration UNION expected ------------------
+    # Live enumeration cannot drift from reality, but is blind by construction to a
+    # consumer that VANISHED. The render-injected expected set covers exactly that.
+    : > "$STATE/set"
+    for s in $(n stream ls --json | jq -r '(. // []) | .[]'); do
+      for c in $(n consumer ls "$s" --json | jq -r '(. // []) | .[]'); do
+        echo "${s}|${c}" >> "$STATE/set"
+      done
+    done
+
+    listify "$EXPECTED" | while read -r e; do
+      grep -qxF "$e" "$STATE/set" 2>/dev/null || echo "$e" >> "$STATE/set"
+    done
+
+    # Ignore list is explicit and reviewable -- never a silent skip.
+    listify "$IGNORE" | while read -r i; do
+      [ -n "$i" ] && echo "[consumer-guard] ignoring ${i} (GUARD_IGNORE)"
+    done
+
+    sort -u "$STATE/set" -o "$STATE/set"
+
+    # --- pass 1 ----------------------------------------------------------------
+    # Printed BEFORE the sleep so a Job killed by activeDeadlineSeconds still
+    # leaves diagnostics rather than an empty log.
+    echo "[consumer-guard] pass 1 (t=0):"
+    while IFS= read -r entry; do
+      st="${entry%%|*}"; du="${entry##*|}"
+      listify "$IGNORE" | grep -qxF "$entry" && continue
+      k=$(key "$st" "$du")
+      probe "$st" "$du" > "$STATE/$k.p1"
+      if [ ! -s "$STATE/$k.p1" ]; then
+        echo "  ${st}/${du}: UNREADABLE"
+      else
+        echo "  ${st}/${du}: $(cat "$STATE/$k.p1" | tr '\t' ' ')"
+      fi
+    done < "$STATE/set"
+
+    sleep "$SLEEP"
+
+    # --- pass 2 + verdict ------------------------------------------------------
+    echo "[consumer-guard] pass 2 (t=${SLEEP}s):"
+    while IFS= read -r entry; do
+      st="${entry%%|*}"; du="${entry##*|}"
+      listify "$IGNORE" | grep -qxF "$entry" && continue
+      k=$(key "$st" "$du")
+      probe "$st" "$du" > "$STATE/$k.p2"
+
+      if [ ! -s "$STATE/$k.p2" ] && [ ! -s "$STATE/$k.p1" ]; then
+        # Deliberately alerts. Not being able to read a consumer is not evidence of
+        # health, and swallowing it reintroduces the silent blind spot this exists for.
+        # Requires BOTH samples to fail: a single blip is a round-trip, not a halt.
+        ok=0
+        echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0"
+        echo "  ALERT ${st}/${du}: consumer could not be read (absent, or NATS refused) -- a halt cannot be ruled out"
+        continue
+      fi
+
+      # If only ONE sample was readable, fall back to it for both -- a single failed
+      # round-trip must not be reported as a stall, nor as health.
+      [ -s "$STATE/$k.p2" ] || cp "$STATE/$k.p1" "$STATE/$k.p2"
+      [ -s "$STATE/$k.p1" ] || cp "$STATE/$k.p2" "$STATE/$k.p1"
+
+      IFS='	' read -r kind bound backlog floor delivered redeliv < "$STATE/$k.p2"
+      IFS='	' read -r _k1 bound1 _bk1 floor1 _d1 _r1 < "$STATE/$k.p1"
+
+      # Layer 1 -- push interest. One sample. Flips within ms of the subscription
+      # dying and cannot false-positive on a deliberately serialized consumer.
+      # Requires the interest to be gone in BOTH samples. A helm upgrade rolls the
+      # Bento pods and leaves a brief window with no subscriber; a guard that fired
+      # on that would alert on every deployment, which is how operators learn to
+      # ignore it. A real stall lasts hours and is unbound in both.
+      if [ "$kind" = "push" ] && [ "$bound" != "true" ] && [ "$bound1" != "true" ]; then
+        ok=0
+        echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0"
+        echo "  ALERT ${st}/${du}: no subscriber bound to deliver subject in either sample over ${SLEEP}s (backlog ${backlog}) -- the pod is Running but its subscription is dead"
+        continue
+      fi
+
+      # Caught up / idle. Nothing waiting means nothing is stuck.
+      if [ "$backlog" = "0" ]; then
+        echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 1"
+        echo "  OK ${st}/${du}: caught up (backlog 0)"
+        continue
+      fi
+
+      # Layer 2 -- ack floor progress. Queue DEPTH is not the signal: latest-kv runs
+      # max_ack_pending=1 as a deliberate serialization mechanism and holds a large,
+      # often growing backlog while working perfectly. What separates "serialized but
+      # progressing" from "dead" is that the ack floor MOVES.
+      #
+      # consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes cap, so
+      # retention deleting messages under a stalled consumer drags the stream-side
+      # floor forward and would read as false progress.
+      if [ "$floor" -gt "$floor1" ] 2>/dev/null; then
+        echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 1"
+        echo "  OK ${st}/${du}: progressing, ack floor ${floor1} -> ${floor} (backlog ${backlog} is irrelevant)"
+        continue
+      fi
+
+      ok=0
+      echo "thingsflow_nats_consumer_progress_ok{stream=\"${st}\",durable=\"${du}\"} 0"
+      echo "  ALERT ${st}/${du}: ${backlog} message(s) waiting and the ack floor is frozen at ${floor} over ${SLEEP}s (delivered ${delivered}, redelivered ${redeliv}) -- bound but not draining"
+    done < "$STATE/set"
+
+    echo "thingsflow_nats_consumer_progress_ok ${ok}"
+    [ "$ok" -eq 1 ] || exit 1
+    echo "[consumer-guard] OK: every consumer is bound and progressing"
+
+{{- end }}
diff --git a/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml b/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml
new file mode 100644
index 0000000..4e6955c
--- /dev/null
+++ b/k8s/helm/thingsflow/templates/nats-consumer-guard.yaml
@@ -0,0 +1,138 @@
+{{- $cg := (.Values.monitoring | default dict).consumerGuard | default dict }}
+{{- if and (.Values.nats.enabled | default true) ($cg.enabled | default true) }}
+{{- $dp := .Values.natsDataPlane | default dict }}
+{{- $historyStore := $dp.historyStore | default (.Values.timeseries.store | default "greptimedb") }}
+{{- $lkv := $dp.latestKv | default dict }}
+{{- $gdb := $dp.greptimedb | default dict }}
+{{- $egdb := $dp.entityGreptimedb | default dict }}
+{{- $qdb := $dp.questdb | default dict }}
+{{- $alm := $dp.alarms | default dict }}
+{{- $am := .Values.alarmMaterializer | default dict }}
+{{- $sleep := $cg.sampleSeconds | default 60 }}
+{{/*
+  Expected set -- mirrors the SAME gates the Bento Deployments use
+  (nats-data-plane-bento.yaml) so "expected but absent" is accurate. Live
+  enumeration cannot drift from reality but is blind by construction to a
+  consumer that VANISHED; this list is what covers that case.
+  A generic `range` over .Values.natsDataPlane would be wrong: that map mixes
+  scalars (enabled, image, natsUrl, rawSubject) with the consumer sub-maps, and
+  real enablement depends on historyStore, which the sub-maps do not carry.
+*/}}
+{{- $expected := list }}
+{{- if $dp.enabled }}
+{{- if ($lkv.enabled | default true) }}
+{{- $expected = append $expected (printf "%s|%s" ($lkv.stream | default "TF_RAW") ($lkv.durable | default "thingsflow-latest-kv-durable")) }}
+{{- end }}
+{{- if and (eq $historyStore "greptimedb") ($gdb.enabled | default true) }}
+{{- $expected = append $expected (printf "%s|%s" ($gdb.stream | default "TF_RAW") ($gdb.durable | default "thingsflow-greptimedb-durable")) }}
+{{- end }}
+{{- if and (eq $historyStore "greptimedb") ($egdb.enabled | default true) }}
+{{- $expected = append $expected (printf "%s|%s" ($egdb.stream | default "TF_ENTITY") ($egdb.durable | default "thingsflow-entity-greptimedb-durable")) }}
+{{- end }}
+{{- if and (eq $historyStore "questdb") ($qdb.enabled | default false) }}
+{{- $expected = append $expected (printf "%s|%s" ($qdb.stream | default "TF_RAW") ($qdb.durable | default "thingsflow-questdb-durable")) }}
+{{- end }}
+{{- if ($alm.enabled | default true) }}
+{{- $expected = append $expected (printf "%s|%s" ($alm.stream | default "TF_RAW") ($alm.durable | default "thingsflow-alarms-durable")) }}
+{{- end }}
+{{- end }}
+{{- if ($am.enabled | default true) }}
+{{/*
+  The alarm-materializer durable is created by the Go client's QueueSubscribe
+  (cmd/alarm-materializer/main.go), NOT by the nats-bootstrap hook, and it lives
+  under .Values.alarmMaterializer rather than natsDataPlane. Its Deployment has
+  no probes at all, so it is the same nats.go silent-death class as the Bento
+  consumers with even less covering it.
+*/}}
+{{- $expected = append $expected (printf "%s|%s" ($am.intentStream | default "TF_ALARMS") ($am.durable | default "thingsflow-alarm-materializer")) }}
+{{- end }}
+#
+# NATS consumer liveness guard.
+#
+# Distinct from greptimedb-freshness-guard, which asks whether the STORE is
+# receiving anything at all. That guard stays green while latest-kv is dead,
+# because a different consumer keeps writing rows to GreptimeDB -- which is
+# exactly how three stalled consumers went unnoticed for hours.
+#
+# Distinct from nats-stream-guard's verify.sh, which owns CONFIG drift
+# (deliver_policy / max_deliver / ack_wait / max_ack_pending / deliver_subject).
+# This guard owns LIVENESS only. One guard per concern.
+#
+# Alert-only, deliberately: remediating would mean granting a scheduled Job
+# standing patch authority over the data plane (the chart has no RBAC objects at
+# all today), and a guard that fixes the problem and exits 0 destroys the very
+# alert surface D-07 is built on. During a real NATS outage a self-remediating
+# tick would also restart every consumer every 5 minutes, turning a recoverable
+# outage into a self-inflicted one -- a stateless CronJob has no memory with
+# which to rate-limit itself.
+apiVersion: batch/v1
+kind: CronJob
+metadata:
+  name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard
+  namespace: {{ .Release.Namespace | default "thingsflow" }}
+  labels:
+    app: nats-consumer-guard
+spec:
+  schedule: {{ $cg.schedule | default "*/5 * * * *" | quote }}
+  # Declared EXPLICITLY, not omitted -- a CronJob without `suspend` in the
+  # manifest can be stopped by hand and no `helm upgrade` will re-enable it
+  # (the freshness guard was silently suspended for 47 hours). Recovering a
+  # guard already suspended by `kubectl patch` needs `helm upgrade
+  # --force-conflicts` ONCE, not `--force`.
+  suspend: false
+  concurrencyPolicy: Forbid
+  startingDeadlineSeconds: 120
+  successfulJobsHistoryLimit: 3
+  # Failed Jobs ARE the alert surface, so keep more of them than successes.
+  failedJobsHistoryLimit: 8
+  jobTemplate:
+    spec:
+      # A failed tick STAYS failed: retrying would paper over the alert.
+      backoffLimit: 0
+      # Must exceed the two-sample window with room for image pull and round
+      # trips. It CANNOT stay at the freshness guard's 90: a DeadlineExceeded
+      # Job is indistinguishable from a real alert but carries no metric line.
+      activeDeadlineSeconds: {{ add $sleep 180 }}
+      template:
+        metadata:
+          labels:
+            app: nats-consumer-guard
+        spec:
+          restartPolicy: Never
+          containers:
+          - name: consumer-guard
+            image: {{ .Values.nats.cliImage | quote }}
+            imagePullPolicy: {{ .Values.images.pullPolicy }}
+            env:
+            # Credentials FIRST: thingsflow.natsURL emits $(NATS_USER)/$(NATS_PASSWORD)
+            # for Kubernetes to expand, and the vars must already be declared.
+            # natsURLLiteral is deliberately NOT used -- it renders NO credentials
+            # when existingSecret is set, which is exactly production.
+{{ include "thingsflow.natsAuthEnv" . | indent 12 }}
+            - name: NATS_URL
+              value: {{ include "thingsflow.natsURL" . | quote }}
+            - name: GUARD_EXPECTED
+              value: {{ join "," $expected | quote }}
+            - name: GUARD_IGNORE
+              value: {{ join "," ($cg.ignore | default (list)) | quote }}
+            - name: GUARD_SAMPLE_SLEEP
+              value: {{ $sleep | quote }}
+            - name: GUARD_NATS_TIMEOUT
+              value: {{ $cg.natsTimeout | default "5s" | quote }}
+            command: ["/bin/sh", "-c", "sh /scripts/check.sh"]
+            volumeMounts:
+            - name: scripts
+              mountPath: /scripts
+            resources:
+              limits:
+                cpu: "200m"
+                memory: "128Mi"
+              requests:
+                cpu: "20m"
+                memory: "48Mi"
+          volumes:
+          - name: scripts
+            configMap:
+              name: {{ include "thingsflow.fullname" . }}-nats-consumer-guard-scripts
+              defaultMode: 0755
+{{- end }}
diff --git a/k8s/helm/thingsflow/templates/nats.yaml b/k8s/helm/thingsflow/templates/nats.yaml
index cf3a053..cc46a01 100644
--- a/k8s/helm/thingsflow/templates/nats.yaml
+++ b/k8s/helm/thingsflow/templates/nats.yaml
@@ -436,6 +436,15 @@ spec:
           # the cluster overlay where questdb.enabled=false). Shares TF_RAW but owns its own ack floor.
           converge_consumer {{ $qdb.stream | default "TF_RAW" | quote }} {{ $qdb.durable | default "thingsflow-questdb-durable" | quote }} {{ $qdb.filterSubject | default "tf.ingest.>" | quote }} \
             {{ $qdb.deliverGroup | default ($qdb.queueGroup | default "thingsflow-nats-questdb") | quote }} {{ $qdb.maxDeliver | default 500 | quote }} {{ $qdb.ackWait | default "30s" | quote }} {{ $qdb.maxAckPending | default 1024 | quote }} {{ $qdb.deliverSubject | default "tf.deliver.questdb" | quote }}
+          {{- else }}
+          {{- $qdbOff := (.Values.natsDataPlane | default dict).questdb | default dict }}
+          # Disabling questdb removed its Deployment but NOT its durable, which then sat on
+          # TF_RAW with no subscriber and its own ack floor, growing monotonically to 106k
+          # pending and never draining (values.yaml documents the live occurrence). On a
+          # capped, discard-old stream an unattended consumer is not inert: it is a
+          # permanently unbound durable that the consumer-liveness guard must legitimately
+          # alert on. Remove it at the source instead of teaching the guard to ignore it.
+          nats --server "$NATS_URL" consumer rm {{ $qdbOff.stream | default "TF_RAW" | quote }} {{ $qdbOff.durable | default "thingsflow-questdb-durable" | quote }} -f 2>/dev/null || true
           {{- end }}
           # TF_LATEST / TF_ALARMS: idempotent create, then converge the PVC cap
           # (max-bytes/max-age) onto the LIVE stream so no file-backed stream is
diff --git a/k8s/helm/thingsflow/values-cluster.example.yaml b/k8s/helm/thingsflow/values-cluster.example.yaml
index 7118cf5..8c08cc2 100644
--- a/k8s/helm/thingsflow/values-cluster.example.yaml
+++ b/k8s/helm/thingsflow/values-cluster.example.yaml
@@ -207,7 +207,11 @@ natsDataPlane:
   enabled: true
   latestKv:
     enabled: true
-    replicas: 2
+    # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge ->
+    # set per message), serialized by the durable's max_ack_pending=1. See
+    # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the
+    # sharded-merge work, not a replica bump.
+    replicas: 1
     resources:
       limits:
         cpu: "1500m"
diff --git a/k8s/helm/thingsflow/values-demo.yaml b/k8s/helm/thingsflow/values-demo.yaml
index c90acdc..44aca22 100644
--- a/k8s/helm/thingsflow/values-demo.yaml
+++ b/k8s/helm/thingsflow/values-demo.yaml
@@ -23,7 +23,11 @@ rmqttEdge:
 
 natsDataPlane:
   latestKv:
-    replicas: 2
+    # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge ->
+    # set per message), serialized by the durable's max_ack_pending=1. See
+    # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the
+    # sharded-merge work, not a replica bump.
+    replicas: 1
   questdb:
     replicas: 2
   alarms:
diff --git a/k8s/helm/thingsflow/values-pilot.example.yaml b/k8s/helm/thingsflow/values-pilot.example.yaml
index d2146da..176eedc 100644
--- a/k8s/helm/thingsflow/values-pilot.example.yaml
+++ b/k8s/helm/thingsflow/values-pilot.example.yaml
@@ -155,7 +155,11 @@ natsDataPlane:
   historyStore: "greptimedb"
   latestKv:
     enabled: true
-    replicas: 2
+    # 1, not 2: this consumer is the single-writer doc-merge (KV get -> merge ->
+    # set per message), serialized by the durable's max_ack_pending=1. See
+    # values.yaml and files/bento-nats-latest-kv.yaml. Scaling it out is the
+    # sharded-merge work, not a replica bump.
+    replicas: 1
     resources:
       limits:
         cpu: "1500m"
diff --git a/k8s/helm/thingsflow/values.yaml b/k8s/helm/thingsflow/values.yaml
index 49ae739..45ef86b 100644
--- a/k8s/helm/thingsflow/values.yaml
+++ b/k8s/helm/thingsflow/values.yaml
@@ -255,6 +255,27 @@ monitoring:
     schedule: "*/5 * * * *"
     staleMinutes: 15
 
+  # Consumer liveness guard: fails a CronJob (the alert surface) when any
+  # JetStream durable is unbound or not draining. This catches the failure the
+  # freshness guard structurally cannot -- a dead Bento subscription while the
+  # pod stays Running 1/1 and OTHER consumers keep GreptimeDB fresh. It happened
+  # to latest-kv, alarms and entity-greptimedb in a single node maintenance.
+  consumerGuard:
+    enabled: true
+    schedule: "*/5 * * * *"
+    # Seconds between the two samples. The check needs two: the ack-floor
+    # progress signal has no stateless equivalent on NATS 2.10.26 (there is no
+    # ack_floor.last_active field on the pinned server). Raising this makes the
+    # guard more patient with slow consumers; activeDeadlineSeconds tracks it
+    # automatically (sampleSeconds + 180).
+    sampleSeconds: 60
+    natsTimeout: "5s"
+    # Explicit, reviewable skip list of "STREAM|durable" pairs -- never a silent
+    # skip. Prefer REMOVING an orphaned consumer at the source (templates/nats.yaml)
+    # over ignoring it here: an unattended durable accumulating backlog on a
+    # capped stream is a real problem, not noise.
+    ignore: []
+
 retention:
   greptimedb:
     enabled: true
diff --git a/mkdocs.yml b/mkdocs.yml
index 7b166f4..96e4221 100644
--- a/mkdocs.yml
+++ b/mkdocs.yml
@@ -103,3 +103,4 @@ nav:
   - Decisions:
       - "ADR-0001: Edge store-and-forward backfill": adr/0001-edge-store-and-forward-backfill.md
       - "ADR-0002: TB UI contract data-fidelity audit": adr/0002-ui-contract-data-fidelity-audit.md
+      - "ADR-0003: NATS consumer liveness guard": adr/0003-nats-consumer-progress-guard.md
diff --git a/tools/python/test_consumer_guard_progress.py b/tools/python/test_consumer_guard_progress.py
new file mode 100644
index 0000000..ad044ba
--- /dev/null
+++ b/tools/python/test_consumer_guard_progress.py
@@ -0,0 +1,203 @@
+"""The consumer guard must judge progress, not queue depth.
+
+Why this exists: a rolling node maintenance restarted NATS and left three Bento
+consumers (latest-kv, alarms, entity-greptimedb) holding dead subscriptions.
+All three reported Running 1/1 for hours -- neither /ping nor /ready detects a
+subscription that is dead underneath a live connection -- and the halt was found
+only when a human ran `nats consumer info` by hand.
+
+The obvious detector is a backlog threshold, and it is wrong in the direction
+that makes a guard worthless. latest-kv runs max_ack_pending=1 as a deliberate
+single-writer serialization mechanism, so under load it holds a large and often
+GROWING backlog while working perfectly: measured against a real NATS 2.10.26
+server at a sustained backlog of 6,729 while its ack floor advanced 61,179 ->
+100,526. A depth rule alerts there, every tick, forever.
+
+So the guard checks push_bound (is anything subscribed to the deliver subject)
+and whether ack_floor.consumer_seq MOVES between two samples. These tests pin
+the properties that are easy to "simplify" away later and that no rendering
+error would surface on its own.
+
+Verified in a NATS 2.10.26 sandbox across all four branches before this test was
+written: unbound with 93,000 waiting exits 1; bound but never acking (4,000
+waiting, ack floor frozen) exits 1; an expected-but-absent durable exits 1; and a
+serialized consumer with 10,974 waiting and an advancing ack floor exits 0.
+"""
+import pathlib
+import subprocess
+import unittest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+CHART = ROOT / "k8s/helm/thingsflow"
+SCRIPTS = "templates/nats-consumer-guard-scripts.yaml"
+CRONJOB = "templates/nats-consumer-guard.yaml"
+
+
+def render(template, *set_args, values=None):
+    cmd = ["helm", "template", "tf", str(CHART), "--show-only", template]
+    if values:
+        cmd += ["-f", str(CHART / values)]
+    for arg in set_args:
+        cmd += ["--set", arg]
+    out = subprocess.run(cmd, capture_output=True, text=True, timeout=180)
+    if out.returncode != 0:
+        raise AssertionError(f"helm template failed: {out.stderr}")
+    return out.stdout
+
+
+class ConsumerGuardScriptTest(unittest.TestCase):
+    def setUp(self):
+        self.script = render(SCRIPTS)
+
+    def test_progress_is_measured_on_the_consumer_ack_floor(self):
+        # consumer_seq, NOT stream_seq: TF_RAW is discard-old with a max_bytes
+        # cap, so retention deleting messages out from under a STALLED consumer
+        # drags its stream-side floor forward and reads as false progress.
+        self.assertIn("ack_floor.consumer_seq", self.script)
+        self.assertNotIn("ack_floor.stream_seq", self.script)
+
+    def test_backlog_alone_never_decides(self):
+        # Backlog may only be used to short-circuit to HEALTHY (nothing waiting
+        # means nothing is stuck). It must never be compared against a threshold
+        # or against its own earlier value -- that is the rule that would report
+        # a serialized latest-kv under load as dead.
+        self.assertIn('[ "$backlog" = "0" ]', self.script)
+        self.assertNotIn("$backlog -gt", self.script)
+        self.assertNotIn("$backlog1", self.script)
+        self.assertNotIn("FLOOR", self.script)
+
+    def test_absent_push_bound_is_read_as_false(self):
+        # push_bound is omitempty: the server OMITS it when no subscriber is
+        # bound, so it arrives as null. Without the `// false` default a naive
+        # test would treat the stalled case as healthy -- the exact blind spot.
+        self.assertIn(".push_bound // false", self.script)
+
+    def test_an_alert_requires_the_fault_in_both_samples(self):
+        # A helm upgrade rolls the Bento pods and leaves a brief window with no
+        # subscriber. Firing on that would alert on every deployment.
+        self.assertIn('[ "$bound" != "true" ] && [ "$bound1" != "true" ]', self.script)
+
+    def test_lists_keep_their_last_element(self):
+        # printf '%s' without a trailing newline makes `while read` silently drop
+        # the LAST element of every comma-separated list -- and the last element
+        # of the rendered expected set is the alarm-materializer durable, the one
+        # consumer with no Kubernetes probes at all.
+        self.assertIn("printf '%s\\n' \"$1\" | tr ',' '\\n'", self.script)
+
+    def test_unreadable_consumers_alert_rather_than_pass(self):
+        # Not being able to read a consumer is not evidence of health.
+        self.assertIn("a halt cannot be ruled out", self.script)
+
+    def test_unreachable_nats_is_one_alert_not_a_storm(self):
+        self.assertIn("NATS unreachable", self.script)
+        self.assertIn("thingsflow_nats_consumer_progress_ok 0", self.script)
+
+
+class ConsumerGuardCronJobTest(unittest.TestCase):
+    def test_expected_set_tracks_the_history_store(self):
+        # The expected set must mirror the same gates the Bento Deployments use.
+        # A generic range over natsDataPlane cannot: that map mixes scalars with
+        # the consumer sub-maps, and enablement depends on historyStore, which
+        # the sub-maps do not carry.
+        gt = render(CRONJOB, "natsDataPlane.historyStore=greptimedb")
+        self.assertIn("thingsflow-greptimedb-durable", gt)
+        self.assertIn("thingsflow-entity-greptimedb-durable", gt)
+
+        qdb = render(CRONJOB, "natsDataPlane.historyStore=questdb")
+        self.assertNotIn("thingsflow-greptimedb-durable", qdb)
+        self.assertNotIn("thingsflow-entity-greptimedb-durable", qdb)
+
+    def test_the_probeless_alarm_materializer_durable_is_covered(self):
+        # Created by the Go client, not the bootstrap hook, and living under a
+        # different values key -- so a values-driven guard would miss it. Its
+        # Deployment has no probes of any kind.
+        rendered = render(CRONJOB)
+        self.assertIn("TF_ALARMS|thingsflow-alarm-materializer", rendered)
+
+    def test_credentials_are_declared_before_the_url_that_expands_them(self):
+        # thingsflow.natsURL emits $(NATS_USER)/$(NATS_PASSWORD) for Kubernetes
+        # to expand; the vars must already be declared at that point.
+        rendered = render(CRONJOB, "nats.auth.enabled=true", "nats.auth.password=x")
+        self.assertLess(
+            rendered.index("name: NATS_USER"), rendered.index("name: NATS_URL")
+        )
+        # natsURLLiteral renders NO credentials under existingSecret, which is
+        # exactly production, so the env-expanding form must be the one used.
+        self.assertIn("$(NATS_USER)", rendered)
+
+    def test_deadline_outlives_the_two_sample_window(self):
+        # The freshness guard's 90s would be exceeded by the sleep alone, and a
+        # DeadlineExceeded Job is indistinguishable from a real alert but carries
+        # no metric line.
+        rendered = render(CRONJOB, "monitoring.consumerGuard.sampleSeconds=120")
+        self.assertIn("activeDeadlineSeconds: 300", rendered)
+
+    def test_a_failed_tick_stays_failed_and_is_kept(self):
+        rendered = render(CRONJOB)
+        self.assertIn("backoffLimit: 0", rendered)
+        self.assertIn("suspend: false", rendered)
+        self.assertIn("concurrencyPolicy: Forbid", rendered)
+
+    def test_no_scrape_infrastructure_is_introduced(self):
+        # D-07: the failed Job IS the alert. There is no scrape target for a Job
+        # stdout line.
+        rendered = render(CRONJOB) + render(SCRIPTS)
+        # Narrow assertions: the templates' own comments say "do NOT add a
+        # ServiceMonitor", so a bare substring check matches the warning itself.
+        self.assertNotIn("kind: ServiceMonitor", rendered)
+        self.assertNotIn("prometheus.io/scrape:", rendered)
+
+
+class OrphanDurableTest(unittest.TestCase):
+    def test_disabling_questdb_removes_its_durable(self):
+        # An orphaned durable is not inert on a capped, discard-old stream: the
+        # questdb one grew to 106k pending and never drained. The guard would
+        # alert on it forever and correctly, so it is removed at the source
+        # rather than added to an ignore list.
+        rendered = render("templates/nats.yaml", "natsDataPlane.questdb.enabled=false")
+        self.assertIn('consumer rm "TF_RAW" "thingsflow-questdb-durable"', rendered)
+
+
+class LatestKvSingleWriterTest(unittest.TestCase):
+    """latest-kv must never ship with more than one replica.
+
+    The overlays carried `replicas: 2` against the invariant documented in
+    values.yaml and files/bento-nats-latest-kv.yaml -- in the example files new
+    deployments copy from, so every fresh cluster inherited it. Production was
+    running 1, which is why it was a landmine rather than a live fault.
+    """
+
+    def test_every_shipped_values_file_keeps_the_single_writer(self):
+        import yaml
+
+        # values-cluster.yaml is gitignored (the tracked template is
+        # values-cluster.example.yaml), so it is deliberately not listed here --
+        # a test may only depend on files the repo actually ships.
+        for name in sorted(CHART.glob("values*.yaml")):
+            with self.subTest(values=name.name):
+                doc = yaml.safe_load(name.read_text()) or {}
+                latest = (doc.get("natsDataPlane") or {}).get("latestKv") or {}
+                if "replicas" not in latest:
+                    continue
+                self.assertEqual(
+                    latest["replicas"],
+                    1,
+                    f"{name.name} scales the single-writer doc-merge consumer",
+                )
+
+    def test_the_rendered_deployment_is_a_single_writer(self):
+        import yaml
+
+        rendered = render("templates/nats-data-plane-bento.yaml")
+        deploys = {
+            d["metadata"]["name"]: d
+            for d in yaml.safe_load_all(rendered)
+            if d and d.get("kind") == "Deployment"
+        }
+        latest = next(n for n in deploys if n.endswith("nats-latest-kv"))
+        self.assertEqual(deploys[latest]["spec"]["replicas"], 1)
+
+
+if __name__ == "__main__":
+    unittest.main()

From 152b9a358c25539b6cb2e45034c176f56df8f3d3 Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Thu, 20 Aug 2026 15:49:50 +0000
Subject: [PATCH 2/6] fix: unbreak the two CI gates that were red on main
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Three defects, all from the ADR-0002 / twin-state Phase 3 work, found by
reading the failing runs rather than assuming they were pre-existing noise.

Test flow-core — POST /api/entityViews returned 500
entity_id is a uuid column and ExtractEntityID yields "" for an absent field,
so a body without entityId handed the driver an empty string and failed with
`invalid input syntax for type uuid` — a malformed request answered as a server
fault. Now goes through dbutil.NullUUID, the same way customerId beside it
always has. Covered both ways: the realistic body round-trips its entityId, and
the omitted-entityId body stores NULL instead of 500-ing.

Test flow-core — nil-pointer panic in the async audit writer
The writer is a goroutine draining a queue, so an event can reach doWrite after
the pool it needs is gone: during shutdown, or in tests where the harness resets
the pool at cleanup while a write is still in flight. Killing the process to
avoid losing one audit row is the wrong trade — the stores are the source of
truth and this journal is secondary. Drops and counts the event instead.

Fresh-install smoke — latest-kv never started
When latest-kv became the single-writer doc-merge it gained
`max_ack_pending: ${LATEST_KV_MAX_ACK_PENDING}`. The Helm template was updated;
docker-compose was not. Bento treats a ${VAR} with no default as REQUIRED and
refuses to start on it, so the container never came up, the twin_state KV was
never populated, and the smoke failed three steps downstream on "NATS KV
twin_state did not hold both keys within 60s" with nothing pointing at the
cause. The compose bootstrap also created every durable with a shared
--max-pending 1024, including latest-kv's, which must be 1 — Bento asserts its
own value on bind, so that alone would have failed the attach.

Adds test_bento_env_contract.py: the chart and compose mount the SAME config
files, so every required var has to be satisfied twice. That contract is now
explicit instead of depending on whoever remembers both call sites.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 docker/docker-compose-nats.yml              | 18 +++-
 flow-core/assets_entityviews_post_test.go   | 48 ++++++++++-
 flow-core/internal/audit/audit.go           | 11 +++
 flow-core/internal/entityview/entityview.go | 10 ++-
 tools/python/test_bento_env_contract.py     | 96 +++++++++++++++++++++
 5 files changed, 176 insertions(+), 7 deletions(-)
 create mode 100644 tools/python/test_bento_env_contract.py

diff --git a/docker/docker-compose-nats.yml b/docker/docker-compose-nats.yml
index 4141230..c5eff97 100644
--- a/docker/docker-compose-nats.yml
+++ b/docker/docker-compose-nats.yml
@@ -114,9 +114,14 @@ services:
         # The Bento materializers attach with `bind: true`, so their durable consumers must
         # exist before they start — mirrors the chart's nats-bootstrap hook. Without these
         # the local stack fails with "nats: consumer not found".
-        for c in "TF_RAW thingsflow-greptimedb-durable tf.ingest.> tf.deliver.greptimedb thingsflow-nats-greptimedb" \
-                 "TF_RAW thingsflow-latest-kv-durable tf.ingest.> tf.deliver.latest-kv thingsflow-nats-latest-kv" \
-                 "TF_RAW thingsflow-alarms-durable tf.ingest.> tf.deliver.alarms thingsflow-nats-alarms"; do
+        # 6th field is max-pending. It is per-consumer, not shared: the latest-kv
+        # durable must be 1 -- that is the doc-merge's serialization mechanism, and
+        # Bento asserts its own value on bind, so a mismatch here fails the bind
+        # with "configuration requests max ack pending to be N, but consumer's
+        # value is M" and the materializer never attaches.
+        for c in "TF_RAW thingsflow-greptimedb-durable tf.ingest.> tf.deliver.greptimedb thingsflow-nats-greptimedb 1024" \
+                 "TF_RAW thingsflow-latest-kv-durable tf.ingest.> tf.deliver.latest-kv thingsflow-nats-latest-kv 1" \
+                 "TF_RAW thingsflow-alarms-durable tf.ingest.> tf.deliver.alarms thingsflow-nats-alarms 1024"; do
           set -- $$c
           nats --server nats://nats:4222 consumer add "$1" "$2" \
             --filter "$3" \
@@ -127,7 +132,7 @@ services:
             --replay instant \
             --max-deliver 500 \
             --wait 30s \
-            --max-pending 1024 \
+            --max-pending "$6" \
             --defaults || true
         done
     depends_on:
@@ -313,6 +318,11 @@ services:
       NATS_DURABLE: thingsflow-latest-kv-durable
       NATS_FILTER_SUBJECT: "tf.ingest.>"
       NATS_KV_BUCKET: twin_state
+      # Must equal the durable's max-pending above (1). Bento's config declares
+      # this as a required env var with no default, so omitting it does not fall
+      # back -- it fails the lint and the container never starts, which is how
+      # the fresh-install smoke lost its latest-values pipeline entirely.
+      LATEST_KV_MAX_ACK_PENDING: "1"
       DEFAULT_TENANT_ID: aaaaaaaa-1dd2-11b2-8080-808080808080
       METRICS_PORT: "4197"
     ports:
diff --git a/flow-core/assets_entityviews_post_test.go b/flow-core/assets_entityviews_post_test.go
index 1aeb4be..4dc5124 100644
--- a/flow-core/assets_entityviews_post_test.go
+++ b/flow-core/assets_entityviews_post_test.go
@@ -100,7 +100,15 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) {
 	})
 
 	t.Run("POST /api/entityViews creates a real row", func(t *testing.T) {
-		body, _ := json.Marshal(map[string]interface{}{"name": "view-1", "type": "default"})
+		body, _ := json.Marshal(map[string]interface{}{
+			"name": "view-1", "type": "default",
+			// The real UI always names a source entity; a view without one is
+			// meaningless. Exercise that path rather than the degenerate body.
+			"entityId": map[string]interface{}{
+				"entityType": "DEVICE",
+				"id":         "88888888-8888-8888-8888-888888888888",
+			},
+		})
 		req := httptest.NewRequest(http.MethodPost, "/api/entityViews", bytes.NewReader(body))
 		req.Header.Set("X-Authorization", "Bearer "+tok)
 		req.Header.Set("Content-Type", "application/json")
@@ -117,5 +125,43 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) {
 		if count != 1 {
 			t.Fatalf("entity_view rows named view-1: got %d, want 1 (a prior version silently re-listed instead of creating)", count)
 		}
+		var storedEntity sql.NullString
+		if err := db.QueryRow(
+			"SELECT entity_id FROM entity_view WHERE tenant_id = $1 AND name = 'view-1'", tenantID,
+		).Scan(&storedEntity); err != nil {
+			t.Fatalf("read back entity_id: %v", err)
+		}
+		if !storedEntity.Valid || storedEntity.String != "88888888-8888-8888-8888-888888888888" {
+			t.Errorf("entity_id did not round-trip: %+v", storedEntity)
+		}
+	})
+
+	// entity_id is a uuid column and ExtractEntityID yields "" for an absent
+	// field, so a body without entityId used to hand the driver an empty string
+	// and fail with `invalid input syntax for type uuid` -- reported as a 500,
+	// i.e. a malformed request answered as a server fault. It must store NULL.
+	t.Run("POST /api/entityViews without entityId does not 500", func(t *testing.T) {
+		body, _ := json.Marshal(map[string]interface{}{"name": "view-no-entity", "type": "default"})
+		req := httptest.NewRequest(http.MethodPost, "/api/entityViews", bytes.NewReader(body))
+		req.Header.Set("X-Authorization", "Bearer "+tok)
+		req.Header.Set("Content-Type", "application/json")
+		rec := httptest.NewRecorder()
+		mux.ServeHTTP(rec, req)
+
+		if rec.Code == http.StatusInternalServerError {
+			t.Fatalf("got 500 for a body with no entityId: %s", rec.Body.String())
+		}
+		if rec.Code != http.StatusOK {
+			t.Fatalf("status: got %d, body=%s", rec.Code, rec.Body.String())
+		}
+		var stored sql.NullString
+		if err := db.QueryRow(
+			"SELECT entity_id FROM entity_view WHERE tenant_id = $1 AND name = 'view-no-entity'", tenantID,
+		).Scan(&stored); err != nil {
+			t.Fatalf("verify: %v", err)
+		}
+		if stored.Valid {
+			t.Errorf("entity_id should be NULL when omitted, got %q", stored.String)
+		}
 	})
 }
diff --git a/flow-core/internal/audit/audit.go b/flow-core/internal/audit/audit.go
index d07cbc7..bb67b83 100644
--- a/flow-core/internal/audit/audit.go
+++ b/flow-core/internal/audit/audit.go
@@ -152,6 +152,17 @@ func Write(e Event) {
 }
 
 func doWrite(e Event) {
+	// The writer is a goroutine draining a queue, so an event can reach here
+	// after the pool it needs is gone -- during shutdown, or in tests where the
+	// harness resets the pool at cleanup while a write is still in flight.
+	// Dereferencing a nil pool panics the whole process, and killing a server to
+	// avoid losing one audit row is the wrong trade: the stores are the source of
+	// truth and this journal is secondary. Drop the event and count it.
+	if dbpkg.Pool == nil {
+		drops.Add(1)
+		metricDropped.Inc()
+		return
+	}
 	now := time.Now().UnixMilli()
 	_, err := dbpkg.Pool.Exec(`
 		INSERT INTO audit_log (
diff --git a/flow-core/internal/entityview/entityview.go b/flow-core/internal/entityview/entityview.go
index 50fcb71..f8a2d2e 100644
--- a/flow-core/internal/entityview/entityview.go
+++ b/flow-core/internal/entityview/entityview.go
@@ -36,6 +36,12 @@ func Save(w http.ResponseWriter, r *http.Request) {
 	if viewType == "" {
 		viewType = "default"
 	}
+	// entity_id is a uuid column, and ExtractEntityID returns "" when the field
+	// is absent or malformed. Passing that empty string straight to the driver
+	// made a request without an entityId fail with `invalid input syntax for
+	// type uuid` and surface as a 500 -- a malformed body answered as a server
+	// fault. NullUUID stores NULL instead, matching how customerId below has
+	// always been handled.
 	entityId := httputil.ExtractEntityID(body, "entityId")
 	entityType := ""
 	if v, ok := body["entityId"].(map[string]interface{}); ok {
@@ -69,7 +75,7 @@ func Save(w http.ResponseWriter, r *http.Request) {
 			UPDATE entity_view SET name = $1, type = $2, entity_id = $3, entity_type = $4,
 			    keys = $5, customer_id = $6, start_ts = $7, end_ts = $8,
 			    version = COALESCE(version, 1) + 1 WHERE id = $9`,
-			name, viewType, entityId, entityType, keys, dbutil.NullUUID(customerId), startTs, endTs, id)
+			name, viewType, dbutil.NullUUID(entityId), entityType, keys, dbutil.NullUUID(customerId), startTs, endTs, id)
 		if err != nil {
 			httputil.WriteError(w, http.StatusInternalServerError, "Failed to update entity view")
 			return
@@ -83,7 +89,7 @@ func Save(w http.ResponseWriter, r *http.Request) {
 		INSERT INTO entity_view (id, created_time, name, type, entity_id, entity_type, tenant_id,
 		                       customer_id, keys, start_ts, end_ts, version)
 		VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, 1)`,
-		id, now, name, viewType, entityId, entityType, tenantId,
+		id, now, name, viewType, dbutil.NullUUID(entityId), entityType, tenantId,
 		dbutil.NullUUID(customerId), keys, startTs, endTs)
 	if err != nil {
 		httputil.WriteError(w, http.StatusInternalServerError, "Failed to create entity view")
diff --git a/tools/python/test_bento_env_contract.py b/tools/python/test_bento_env_contract.py
new file mode 100644
index 0000000..f903769
--- /dev/null
+++ b/tools/python/test_bento_env_contract.py
@@ -0,0 +1,96 @@
+"""Every env var a Bento config requires must be supplied by BOTH deploy paths.
+
+Why this exists: the latest-kv config gained `max_ack_pending:
+${LATEST_KV_MAX_ACK_PENDING}` when that consumer became the single-writer
+doc-merge. The Helm template was updated; docker-compose was not. Bento treats a
+`${VAR}` with no default as REQUIRED and refuses to start on it -- not with a
+warning, and not by falling back:
+
+    Config lint error  lint="(1,1) required environment variables were not set:
+                             [LATEST_KV_MAX_ACK_PENDING]"
+    Shutting down due to linter errors
+
+So the container never came up at all, the twin-state KV was never populated,
+and the fresh-install smoke failed on a symptom three steps downstream ("NATS KV
+twin_state did not hold both keys within 60s"). Nothing pointed at the missing
+variable.
+
+The chart and docker-compose mount the SAME config files, so any required
+variable has to be satisfied twice. This test makes that contract explicit
+rather than leaving it to whoever remembers both call sites.
+"""
+import pathlib
+import re
+import unittest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+FILES = ROOT / "k8s/helm/thingsflow/files"
+COMPOSE = ROOT / "docker/docker-compose-nats.yml"
+BENTO_TEMPLATE = ROOT / "k8s/helm/thingsflow/templates/nats-data-plane-bento.yaml"
+INGEST_TEMPLATE = ROOT / "k8s/helm/thingsflow/templates/http-ingest.yaml"
+
+# `${VAR}` is required; `${VAR:-default}` and `${VAR:default}` are not.
+REQUIRED = re.compile(r"\$\{([A-Z_][A-Z0-9_]*)\}")
+
+# Supplied by the runtime rather than by either deploy path.
+RUNTIME_PROVIDED = {"BENTO_HTTP_BIND_ADDRESS", "BENTO_HTTP_PORT"}
+
+
+def required_vars(path):
+    return {v for v in REQUIRED.findall(path.read_text())} - RUNTIME_PROVIDED
+
+
+class BentoEnvContractTest(unittest.TestCase):
+    def test_every_required_var_is_set_in_both_deploy_paths(self):
+        compose = COMPOSE.read_text()
+        charts = BENTO_TEMPLATE.read_text()
+        if INGEST_TEMPLATE.exists():
+            charts += INGEST_TEMPLATE.read_text()
+
+        configs = sorted(FILES.glob("bento-*.yaml"))
+        self.assertTrue(configs, "no bento configs found — did the path move?")
+
+        for config in configs:
+            # Only assert against compose for configs compose actually MOUNTS.
+            # The questdb path is the legacy backend and has no compose service;
+            # demanding its vars there would be a standing false alarm. If anyone
+            # adds the service, the mount appears and this starts covering it.
+            mounted_by_compose = config.name in compose
+
+            for var in sorted(required_vars(config)):
+                with self.subTest(config=config.name, var=var):
+                    self.assertIn(
+                        var,
+                        charts,
+                        f"{config.name} requires {var}, absent from the chart "
+                        f"templates — the pod will fail Bento's lint and never start",
+                    )
+                    if mounted_by_compose:
+                        self.assertIn(
+                            var,
+                            compose,
+                            f"{config.name} requires {var}, absent from "
+                            f"docker-compose-nats.yml — the local stack and the "
+                            f"fresh-install smoke will fail Bento's lint and never start",
+                        )
+
+    def test_latest_kv_serialization_matches_across_both_paths(self):
+        """The doc-merge value must agree in three places or the bind fails.
+
+        Bento asserts its own max_ack_pending against the durable's, so a
+        mismatch is not a performance difference — NATS refuses the bind with
+        "configuration requests max ack pending to be N, but consumer's value is
+        M" and the consumer never attaches.
+        """
+        compose = COMPOSE.read_text()
+        self.assertIn('LATEST_KV_MAX_ACK_PENDING: "1"', compose)
+        # The compose bootstrap must create that durable with a matching 1, so
+        # max-pending cannot be a single shared value across the consumer loop.
+        self.assertIn("thingsflow-latest-kv-durable", compose)
+        self.assertNotIn("--max-pending 1024", compose)
+        self.assertIn('--max-pending "$6"', compose)
+
+
+if __name__ == "__main__":
+    unittest.main()

From f8ff53c0b89558ce5678e16e286067b3f9d43928 Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Thu, 20 Aug 2026 15:54:50 +0000
Subject: [PATCH 3/6] fix(smoke): assert the doc-per-device twin state, not the
 old per-key fan-out
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The latest-kv consumer became a doc-merge writing the whole device state under
DEVICE.<tenant>.<device>; the DEVICE.<tenant>.<device>.telemetry.<key> fan-out
it replaced no longer exists. The smoke still read the old per-key paths, so it
failed with "twin_state did not hold both keys" — which reads as a dead
latest-values pipeline rather than a stale assertion, and sent the last
investigation looking at Bento instead of at this line.

Now reads the single device document and requires BOTH keys to carry the
published value inside it, which is what actually proves the merge. The failure
message prints the document so the next mismatch is diagnosable from the log.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 tools/smoke/fresh-install-smoke.sh | 22 +++++++++++++++++-----
 1 file changed, 17 insertions(+), 5 deletions(-)

diff --git a/tools/smoke/fresh-install-smoke.sh b/tools/smoke/fresh-install-smoke.sh
index f376e22..ac79cef 100755
--- a/tools/smoke/fresh-install-smoke.sh
+++ b/tools/smoke/fresh-install-smoke.sh
@@ -186,19 +186,31 @@ done
 log "history (GreptimeDB) read OK: $(printf '%s' "$HIST_BODY" | jq -c .)"
 
 log "polling NATS KV twin_state bucket directly for both keys"
-kv_get() {
+# ONE document per device, not one entry per key. The latest-kv consumer became
+# a doc-merge (KV get -> merge -> set) writing the whole device state under
+# DEVICE.<tenant>.<device>; the old DEVICE.<tenant>.<device>.telemetry.<key>
+# fan-out no longer exists. Asserting the old shape here failed with "did not
+# hold both keys", which reads as a dead pipeline rather than a stale assertion.
+kv_doc() {
   docker compose -f "$COMPOSE_FILE" run --rm nats-box -c \
-    "nats --server nats://nats:4222 kv get twin_state 'DEVICE.$TENANT_UUID.$DEVICE_ID.telemetry.$1' --raw" 2>/dev/null
+    "nats --server nats://nats:4222 kv get twin_state 'DEVICE.$TENANT_UUID.$DEVICE_ID' --raw" 2>/dev/null
 }
 kv_found=0
+KV_BODY=""
 for _ in $(seq 1 $((ROW_TIMEOUT_SECS / 2))); do
-  if kv_get smoke_http | grep -q "$EPOCH" && kv_get smoke_mqtt | grep -q "$EPOCH"; then
+  KV_BODY="$(kv_doc)"
+  # Both keys must be present in the SAME document with the published value —
+  # that is what proves the merge, not just that something was written.
+  if printf '%s' "$KV_BODY" | jq -e \
+    --arg v "$EPOCH" \
+    '(.telemetry.smoke_http.value | tostring | contains($v)) and
+     (.telemetry.smoke_mqtt.value | tostring | contains($v))' >/dev/null 2>&1; then
     kv_found=1; break
   fi
   sleep 2
 done
-[ "$kv_found" -eq 1 ] || die "NATS KV twin_state did not hold both keys within ${ROW_TIMEOUT_SECS}s (latest-values pipeline dead?)"
-log "latest (NATS KV) bucket holds both keys"
+[ "$kv_found" -eq 1 ] || die "NATS KV twin_state doc DEVICE.$TENANT_UUID.$DEVICE_ID did not hold both keys within ${ROW_TIMEOUT_SECS}s (latest-values pipeline dead?); last doc: ${KV_BODY:-<empty>}"
+log "latest (NATS KV) device doc holds both keys"
 
 # --- h. verdict (cleanup runs via the EXIT trap) -----------------------------
 log "PASS — fresh install serves login, device identity, HTTP edge, MQTT edge, GreptimeDB history, and the NATS KV latest pipeline"

From 530c15c138d2ad859fc3195638821fd8236dc2e9 Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Fri, 21 Aug 2026 13:15:51 +0000
Subject: [PATCH 4/6] test: confine database harnesses to a throwaway schema
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

29 harnesses opened FLOW_TEST_PG_DSN and ran `DROP TABLE IF EXISTS asset
CASCADE` (and a dozen siblings) for a clean slate. Against the disposable
Postgres CI starts as a service container that is fine. Against any database a
human might point that variable at — a local dev stack, a shared scratch
database, a pilot — it is a data-loss bug waiting for one careless export. They
also collided with each other; CI hid that with `go test -p 1`, so the breakage
only reached whoever ran `go test ./...` locally, where packages run in parallel
by default.

internal/testdb.Scoped creates a private schema and returns a DSN carrying
`search_path=<schema>`, so the existing destructive DDL resolves inside it and
cannot reach public. The harness bodies did not have to change — only the DSN
they open.

search_path is a CONNECTION PARAMETER, not a `SET`. sql.DB is a pool: `SET
search_path` binds only the connection that served it, so under concurrency a
later query lands on one still pointed at public — the confinement failing open,
silently, exactly when load makes it matter. Verified against postgres:16: 64
concurrent queries, 0 unconfined.

The migration then exposed two real defects that had been invisible:

- internal/twin's tests only passed because internal/customer, internal/dashboard
  and friends had created `customer`, `dashboard`, `entity_view`,
  `device_profile` and `asset_profile` in the shared public schema earlier in the
  run. The expand CTE resolves a relation's endpoints against all of them. Once
  each package got its own schema that borrowed state vanished and the traversal
  500'd. The isolation did not break those tests; it revealed what they leaned on.

- The 0014/0015 migration tests queried information_schema and pg_constraint
  without pinning current_schema(). Those catalogues span the whole database, so
  once packages hold concurrent copies of `policy` the assertion counts them all
  and reports "found 2" for a correct migration. Measured: 2 of 5 parallel runs
  failed before, 0 of 6 after.

tools/python/test_db_test_isolation.py enforces all three rules so none of them
can quietly come back.

Also: POST/PUT/DELETE /api/queues fell through to the SELECT and answered 200
with the queue LIST, so the UI reported success for a queue that was never
created. The audit left this open as a product decision; the decision is not to
implement the write. Queues and their consumers are Helm/k8s-owned here — the
durables come from the nats-bootstrap hook and nothing in the data plane reads
the `queue` table — so a row written through the API would configure nothing,
which is a more expensive lie than a refusal. Answered honestly with 501 naming
where queue configuration actually lives.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 docs/DIGITAL_TWIN.md                          |  22 +++
 flow-core/assets_entityviews_post_test.go     |   3 +-
 flow-core/attributes_roundtrip_test.go        |   3 +-
 .../alarmmaterializer/postgres_insert_test.go |   4 +-
 flow-core/internal/asset/asset_test.go        |   3 +-
 flow-core/internal/audit/audit_test.go        |   3 +-
 flow-core/internal/customer/customer_test.go  |   3 +-
 .../internal/dashboard/dashboard_test.go      |   3 +-
 flow-core/internal/device/device_test.go      |   3 +-
 .../migrations/migrations_0014_test.go        |   9 +-
 .../migrations/migrations_0015_test.go        |  13 +-
 .../provisioning/provisioning_db_test.go      |   3 +-
 .../internal/relations/relations_test.go      |   3 +-
 .../internal/resource/image_import_test.go    |   3 +-
 .../system/admin_settings_authz_test.go       |   3 +-
 .../internal/system/asset_bulk_import_test.go |   3 +-
 .../internal/system/oauth2_infos_test.go      |   5 +-
 .../internal/system/queues_write_test.go      |  89 +++++++++++
 flow-core/internal/system/stubs_handlers.go   |  17 ++
 .../system/tenant_dashboard_home_test.go      |   3 +-
 flow-core/internal/system/usage_test.go       |   3 +-
 .../internal/system/user_settings_test.go     |   3 +-
 .../telemetry/questdb_integration_test.go     |   4 +-
 .../telemetry/reader_entity_tenant_test.go    |   3 +-
 .../internal/telemetry/reader_tenant_test.go  |   3 +-
 .../tenant/tenant_handler_authz_test.go       |   3 +-
 flow-core/internal/testdb/testdb.go           | 105 ++++++++++++
 flow-core/internal/testdb/testdb_test.go      | 150 ++++++++++++++++++
 .../topology/topology_benchmark_test.go       |   6 +-
 flow-core/internal/topology/topology_test.go  |   3 +-
 .../internal/transport/transport_test.go      |   3 +-
 flow-core/internal/twin/twin_test.go          |  23 ++-
 .../twinmodel/normalization_sql_test.go       |   4 +-
 flow-core/internal/user/auth_handler_test.go  |   3 +-
 .../internal/user/crud_user_authz_test.go     |   3 +-
 flow-core/internal/ws/ws_tenant_test.go       |   3 +-
 tools/python/test_db_test_isolation.py        | 130 +++++++++++++++
 37 files changed, 615 insertions(+), 35 deletions(-)
 create mode 100644 flow-core/internal/system/queues_write_test.go
 create mode 100644 flow-core/internal/testdb/testdb.go
 create mode 100644 flow-core/internal/testdb/testdb_test.go
 create mode 100644 tools/python/test_db_test_isolation.py

diff --git a/docs/DIGITAL_TWIN.md b/docs/DIGITAL_TWIN.md
index aac7ee7..e62d575 100644
--- a/docs/DIGITAL_TWIN.md
+++ b/docs/DIGITAL_TWIN.md
@@ -762,6 +762,28 @@ FLOW_TEST_PG_DSN='postgres://postgres:postgres@localhost:5432/flowtest_twin?sslm
   go test ./internal/twin
 ```
 
+The throwaway database is belt-and-braces, not the mechanism. Every database
+harness routes its DSN through `internal/testdb.Scoped`, which creates a private
+schema per test and returns a DSN carrying `search_path=<schema>`, so the
+`DROP TABLE ... CASCADE` statements the harnesses run resolve inside that schema
+and cannot reach `public`. Two properties are load-bearing and easy to undo by
+accident:
+
+- **`search_path` is a connection parameter, never a `SET` statement.**
+  `sql.DB` is a pool; `SET search_path` binds only the connection that served
+  it, so under concurrency a later query lands on a connection still pointed at
+  `public` — the confinement failing open, silently, exactly when load makes it
+  matter.
+- **Catalogue queries must pin `current_schema()`.**
+  `information_schema` and `pg_constraint` span the whole database, so an
+  assertion like `WHERE table_name='policy'` counts every concurrent test
+  schema. That one made 2 of 5 parallel runs fail while passing every time under
+  CI's `-p 1`.
+
+`tools/python/test_db_test_isolation.py` enforces all three rules. Because the
+harnesses are confined, `go test ./...` now runs packages in parallel — the
+default — instead of needing `-p 1`.
+
 Full release gate:
 
 ```bash
diff --git a/flow-core/assets_entityviews_post_test.go b/flow-core/assets_entityviews_post_test.go
index 4dc5124..44dc439 100644
--- a/flow-core/assets_entityviews_post_test.go
+++ b/flow-core/assets_entityviews_post_test.go
@@ -13,6 +13,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002,
@@ -26,7 +27,7 @@ func TestAssetsAndEntityViewsPost_DispatchToRealCreate(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/attributes_roundtrip_test.go b/flow-core/attributes_roundtrip_test.go
index 4889093..3472c7e 100644
--- a/flow-core/attributes_roundtrip_test.go
+++ b/flow-core/attributes_roundtrip_test.go
@@ -23,6 +23,7 @@ import (
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
 	"flow-core/internal/tenant"
+	"flow-core/internal/testdb"
 	"flow-core/internal/transport"
 )
 
@@ -43,7 +44,7 @@ func newRootAttrDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/alarmmaterializer/postgres_insert_test.go b/flow-core/internal/alarmmaterializer/postgres_insert_test.go
index 30341a8..fc41b35 100644
--- a/flow-core/internal/alarmmaterializer/postgres_insert_test.go
+++ b/flow-core/internal/alarmmaterializer/postgres_insert_test.go
@@ -1,6 +1,8 @@
 package alarmmaterializer
 
 import (
+	"flow-core/internal/testdb"
+
 	"context"
 	"database/sql"
 	"errors"
@@ -26,7 +28,7 @@ func insertTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/asset/asset_test.go b/flow-core/internal/asset/asset_test.go
index 02bf0cc..94dc104 100644
--- a/flow-core/internal/asset/asset_test.go
+++ b/flow-core/internal/asset/asset_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 func newTestDB(t *testing.T) *sql.DB {
@@ -24,7 +25,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	// per-test Pool swap below (see internal/device/device_test.go for the
 	// full rationale — first Write fixes the mode for the whole binary).
 	t.Setenv("AUDIT_LOG_QUEUE_SIZE", "0")
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/audit/audit_test.go b/flow-core/internal/audit/audit_test.go
index ed3c448..e2f1ee7 100644
--- a/flow-core/internal/audit/audit_test.go
+++ b/flow-core/internal/audit/audit_test.go
@@ -8,6 +8,7 @@ import (
 	"time"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 	_ "github.com/lib/pq"
 )
 
@@ -20,7 +21,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("sql.Open: %v", err)
 	}
diff --git a/flow-core/internal/customer/customer_test.go b/flow-core/internal/customer/customer_test.go
index 3405fd9..d8d896c 100644
--- a/flow-core/internal/customer/customer_test.go
+++ b/flow-core/internal/customer/customer_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 const tenantA = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
@@ -23,7 +24,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/dashboard/dashboard_test.go b/flow-core/internal/dashboard/dashboard_test.go
index 60afe60..6499f73 100644
--- a/flow-core/internal/dashboard/dashboard_test.go
+++ b/flow-core/internal/dashboard/dashboard_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 const tenantA = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
@@ -23,7 +24,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/device/device_test.go b/flow-core/internal/device/device_test.go
index 2d1677b..ac84881 100644
--- a/flow-core/internal/device/device_test.go
+++ b/flow-core/internal/device/device_test.go
@@ -13,6 +13,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 func newTestDB(t *testing.T) *sql.DB {
@@ -29,7 +30,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	// EVERY test of the package: audit's writer starts under a sync.Once, so
 	// the first Write decides the mode for the whole test binary.
 	t.Setenv("AUDIT_LOG_QUEUE_SIZE", "0")
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/migrations/migrations_0014_test.go b/flow-core/internal/migrations/migrations_0014_test.go
index 9c1a7ef..cd1c565 100644
--- a/flow-core/internal/migrations/migrations_0014_test.go
+++ b/flow-core/internal/migrations/migrations_0014_test.go
@@ -257,9 +257,14 @@ func assert0014Rollback(t *testing.T, db *sql.DB) {
 func assertConstraintContains0014(t *testing.T, db *sql.DB, table, name, fragment string) {
 	t.Helper()
 	var definition string
+	// Scoped to current_schema(): relname+conname are unique per schema, not per
+	// database, so a concurrent test schema holding the same table makes this
+	// return multiple rows and QueryRow silently takes whichever comes first.
 	if err := db.QueryRow(`SELECT pg_get_constraintdef(c.oid)
-		FROM pg_constraint c JOIN pg_class r ON r.oid=c.conrelid
-		WHERE r.relname=$1 AND c.conname=$2`, table, name).Scan(&definition); err != nil {
+		FROM pg_constraint c
+		JOIN pg_class r ON r.oid=c.conrelid
+		JOIN pg_namespace n ON n.oid=r.relnamespace
+		WHERE n.nspname=current_schema() AND r.relname=$1 AND c.conname=$2`, table, name).Scan(&definition); err != nil {
 		t.Fatalf("read constraint %s: %v", name, err)
 	}
 	normalized := strings.ToLower(strings.ReplaceAll(definition, "\"", ""))
diff --git a/flow-core/internal/migrations/migrations_0015_test.go b/flow-core/internal/migrations/migrations_0015_test.go
index 9a0530d..2235915 100644
--- a/flow-core/internal/migrations/migrations_0015_test.go
+++ b/flow-core/internal/migrations/migrations_0015_test.go
@@ -83,15 +83,24 @@ func TestPolicyCatalog0015(t *testing.T) {
 	// The kind column and the version CHECK backstop exist on the real table
 	// (mirroring twin_model). An out-of-band non-canonical version is rejected.
 	var kindColumn int
+	// table_schema=current_schema(): the catalogue spans the WHOLE database, so
+	// without it this counts every concurrent test schema that also has a
+	// `policy` table and reports "found 2" for a perfectly correct migration.
+	// It only shows up when packages run in parallel, which is why it survived.
 	if err := db.QueryRow(`SELECT count(*) FROM information_schema.columns
-		WHERE table_name='policy' AND column_name='kind'`).Scan(&kindColumn); err != nil {
+		WHERE table_schema=current_schema()
+		  AND table_name='policy' AND column_name='kind'`).Scan(&kindColumn); err != nil {
 		t.Fatalf("check kind column: %v", err)
 	}
 	if kindColumn != 1 {
 		t.Fatalf("expected policy.kind column, found %d", kindColumn)
 	}
 	var checkCount int
-	if err := db.QueryRow(`SELECT count(*) FROM pg_constraint WHERE conname='policy_version_chk'`).Scan(&checkCount); err != nil {
+	// Same blindness: constraint names are unique per schema, not per database.
+	if err := db.QueryRow(`SELECT count(*) FROM pg_constraint c
+		JOIN pg_class r ON r.oid=c.conrelid
+		JOIN pg_namespace n ON n.oid=r.relnamespace
+		WHERE n.nspname=current_schema() AND c.conname='policy_version_chk'`).Scan(&checkCount); err != nil {
 		t.Fatalf("check version constraint: %v", err)
 	}
 	if checkCount != 1 {
diff --git a/flow-core/internal/provisioning/provisioning_db_test.go b/flow-core/internal/provisioning/provisioning_db_test.go
index 488ed54..2609799 100644
--- a/flow-core/internal/provisioning/provisioning_db_test.go
+++ b/flow-core/internal/provisioning/provisioning_db_test.go
@@ -6,6 +6,7 @@ import (
 	"testing"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 const provTestTenant = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
@@ -17,7 +18,7 @@ func newProvisioningTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/relations/relations_test.go b/flow-core/internal/relations/relations_test.go
index 36762fa..406a6c0 100644
--- a/flow-core/internal/relations/relations_test.go
+++ b/flow-core/internal/relations/relations_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 	"flow-core/internal/topology"
 )
 
@@ -29,7 +30,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/resource/image_import_test.go b/flow-core/internal/resource/image_import_test.go
index 2450fee..ac7ddfb 100644
--- a/flow-core/internal/resource/image_import_test.go
+++ b/flow-core/internal/resource/image_import_test.go
@@ -13,6 +13,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002,
@@ -25,7 +26,7 @@ func TestImageImport_ReturnsFullFieldSet(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/admin_settings_authz_test.go b/flow-core/internal/system/admin_settings_authz_test.go
index 7ca37fa..957cb1b 100644
--- a/flow-core/internal/system/admin_settings_authz_test.go
+++ b/flow-core/internal/system/admin_settings_authz_test.go
@@ -13,6 +13,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Coverage for the admin-settings leak fix: the jwt/mail/security keys are
@@ -26,7 +27,7 @@ func newAdminSettingsDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/asset_bulk_import_test.go b/flow-core/internal/system/asset_bulk_import_test.go
index 02d4054..397650e 100644
--- a/flow-core/internal/system/asset_bulk_import_test.go
+++ b/flow-core/internal/system/asset_bulk_import_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 const bulkImportTenant = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
@@ -22,7 +23,7 @@ func newAssetBulkImportDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/oauth2_infos_test.go b/flow-core/internal/system/oauth2_infos_test.go
index 8237a00..d784da9 100644
--- a/flow-core/internal/system/oauth2_infos_test.go
+++ b/flow-core/internal/system/oauth2_infos_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // fakeSystemJWT mints a TENANT_ADMIN token for tenantID, shared across this
@@ -42,7 +43,7 @@ func TestOAuth2ClientInfos_ReadsRealTable(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
@@ -89,7 +90,7 @@ func TestOAuth2ConfigTemplate_ReadsRealTable(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/queues_write_test.go b/flow-core/internal/system/queues_write_test.go
new file mode 100644
index 0000000..6fce29a
--- /dev/null
+++ b/flow-core/internal/system/queues_write_test.go
@@ -0,0 +1,89 @@
+package system
+
+import (
+	"bytes"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+
+	authpkg "flow-core/internal/auth"
+)
+
+// Regression coverage for the ui-contract data-fidelity audit
+// (docs/adr/0002, docs/UI_CONTRACT_DATA_FIDELITY.md): /api/queues was
+// registered method-agnostic, so a POST fell through to the SELECT and
+// answered 200 with the queue LIST. The UI's "save queue" therefore reported
+// success for a queue that was never created — the same shape as the
+// widgetType DELETE that returned the row it had not deleted.
+//
+// The audit left this one open as a product decision rather than a missed
+// wire-up, and the decision is not to implement the write: on this platform
+// queues and their consumers are Helm/k8s-owned (the durables are created by
+// the nats-bootstrap hook from chart values) and nothing in the data plane
+// reads the `queue` table. A row written through the API would configure
+// nothing, which is a more expensive lie than a refusal — so the endpoint
+// answers honestly, like the other capabilities the platform deliberately
+// does not have.
+//
+// Both cases reject before touching the database, so no Postgres is needed.
+
+func queuesJWT(t *testing.T) string {
+	t.Helper()
+	t.Setenv("JWT_TOKEN_SIGNING_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=")
+	authpkg.InitConfig()
+	tok, err := authpkg.GenerateAccess(authpkg.Subject{
+		UserID:    "00000000-0000-0000-0000-000000000001",
+		Email:     "x@x.org",
+		Authority: "TENANT_ADMIN",
+		TenantID:  "11111111-1111-1111-1111-111111111111",
+	}, "test")
+	if err != nil {
+		t.Fatalf("jwt: %v", err)
+	}
+	return tok
+}
+
+func TestQueuesWriteIsAnsweredHonestly(t *testing.T) {
+	tok := queuesJWT(t)
+
+	for _, method := range []string{http.MethodPost, http.MethodPut, http.MethodDelete} {
+		t.Run(method, func(t *testing.T) {
+			body := bytes.NewReader([]byte(`{"name":"Main","topic":"tb_rule_engine.main"}`))
+			req := httptest.NewRequest(method, "/api/queues", body)
+			req.Header.Set("X-Authorization", "Bearer "+tok)
+			req.Header.Set("Content-Type", "application/json")
+			rec := httptest.NewRecorder()
+
+			HandleQueues(rec, req)
+
+			if rec.Code == http.StatusOK {
+				t.Fatalf("%s /api/queues returned 200 — a prior version answered "+
+					"with the queue list, reporting success for a write that never "+
+					"happened; body=%s", method, rec.Body.String())
+			}
+			if rec.Code != http.StatusNotImplemented {
+				t.Errorf("got %d, want %d", rec.Code, http.StatusNotImplemented)
+			}
+			// The message has to say WHERE queues actually come from, or the
+			// next operator reads 501 as "unfinished" and files a bug.
+			if !bytes.Contains(rec.Body.Bytes(), []byte("Helm")) {
+				t.Errorf("the refusal does not point at the real source of queue "+
+					"configuration: %s", rec.Body.String())
+			}
+		})
+	}
+}
+
+func TestQueuesReadStillRequiresAuth(t *testing.T) {
+	// The honest-refusal branch must not run before authentication — an
+	// unauthenticated caller should still get 401, not a 501 that leaks that
+	// the endpoint exists in this shape.
+	req := httptest.NewRequest(http.MethodPost, "/api/queues", nil)
+	rec := httptest.NewRecorder()
+
+	HandleQueues(rec, req)
+
+	if rec.Code != http.StatusUnauthorized {
+		t.Errorf("got %d, want 401", rec.Code)
+	}
+}
diff --git a/flow-core/internal/system/stubs_handlers.go b/flow-core/internal/system/stubs_handlers.go
index 324b8e8..afab778 100644
--- a/flow-core/internal/system/stubs_handlers.go
+++ b/flow-core/internal/system/stubs_handlers.go
@@ -412,6 +412,23 @@ func HandleQueues(w http.ResponseWriter, r *http.Request) {
 		httputil.WriteError(w, http.StatusUnauthorized, "Unauthorized")
 		return
 	}
+	// Non-GET fell through to the SELECT below and answered 200 with the queue
+	// LIST, so the UI's "save queue" reported success for a queue that was
+	// never created -- the same shape as the widgetType DELETE that returned
+	// the row it had not deleted.
+	//
+	// The fix is not to implement the write. On this platform queues and their
+	// consumers are Helm/k8s-owned: the durables are created by the
+	// nats-bootstrap hook from chart values, and nothing in the data plane
+	// reads the `queue` table. A row written here would configure nothing,
+	// which is a more expensive lie than a refusal. So this is answered
+	// honestly, the same way the platform answers the other capabilities it
+	// deliberately does not have.
+	if r.Method != http.MethodGet {
+		httputil.WriteError(w, http.StatusNotImplemented,
+			"Queue configuration is managed by the platform deployment (Helm values), not through the API")
+		return
+	}
 	tenantId, _ := claims["tenantId"].(string)
 	rows, err := dbpkg.Pool.Query(
 		`SELECT id, created_time, name, topic, poll_interval, partitions, consumer_per_partition,
diff --git a/flow-core/internal/system/tenant_dashboard_home_test.go b/flow-core/internal/system/tenant_dashboard_home_test.go
index 0a8ae82..16d3479 100644
--- a/flow-core/internal/system/tenant_dashboard_home_test.go
+++ b/flow-core/internal/system/tenant_dashboard_home_test.go
@@ -12,6 +12,7 @@ import (
 	_ "github.com/lib/pq"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Regression coverage for the ui-contract data-fidelity audit (docs/adr/0002,
@@ -26,7 +27,7 @@ func TestTenantDashboardHomeInfo_RoundTrips(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/usage_test.go b/flow-core/internal/system/usage_test.go
index 97a28dc..e99016f 100644
--- a/flow-core/internal/system/usage_test.go
+++ b/flow-core/internal/system/usage_test.go
@@ -12,6 +12,7 @@ import (
 
 	dbpkg "flow-core/internal/db"
 	"flow-core/internal/quotas"
+	"flow-core/internal/testdb"
 )
 
 // Regression coverage for the ui-contract data-fidelity audit
@@ -34,7 +35,7 @@ func TestHandleUsage_ReadsRealQuotaLimits(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/system/user_settings_test.go b/flow-core/internal/system/user_settings_test.go
index c2e6b94..4531704 100644
--- a/flow-core/internal/system/user_settings_test.go
+++ b/flow-core/internal/system/user_settings_test.go
@@ -9,6 +9,7 @@ import (
 	_ "github.com/lib/pq"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Coverage for the sidebar-menu fix: the TB UI reads
@@ -23,7 +24,7 @@ func newUserSettingsDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/telemetry/questdb_integration_test.go b/flow-core/internal/telemetry/questdb_integration_test.go
index 8d2d605..2517899 100644
--- a/flow-core/internal/telemetry/questdb_integration_test.go
+++ b/flow-core/internal/telemetry/questdb_integration_test.go
@@ -1,6 +1,8 @@
 package telemetry
 
 import (
+	"flow-core/internal/testdb"
+
 	"database/sql"
 	"fmt"
 	"os"
@@ -32,7 +34,7 @@ func newQuestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_QUESTDB_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/telemetry/reader_entity_tenant_test.go b/flow-core/internal/telemetry/reader_entity_tenant_test.go
index d16f59c..99f01be 100644
--- a/flow-core/internal/telemetry/reader_entity_tenant_test.go
+++ b/flow-core/internal/telemetry/reader_entity_tenant_test.go
@@ -14,6 +14,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // These tests close the non-DEVICE cross-tenant read gap (adversarial review
@@ -38,7 +39,7 @@ func newEntityTenantTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/telemetry/reader_tenant_test.go b/flow-core/internal/telemetry/reader_tenant_test.go
index 59f88ff..e120944 100644
--- a/flow-core/internal/telemetry/reader_tenant_test.go
+++ b/flow-core/internal/telemetry/reader_tenant_test.go
@@ -13,6 +13,7 @@ import (
 	_ "github.com/lib/pq"
 
 	authpkg "flow-core/internal/auth"
+	"flow-core/internal/testdb"
 )
 
 // These tests verify the tenant predicate added to the device telemetry read
@@ -32,7 +33,7 @@ func newTelemetryTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/tenant/tenant_handler_authz_test.go b/flow-core/internal/tenant/tenant_handler_authz_test.go
index 016d507..0758ed5 100644
--- a/flow-core/internal/tenant/tenant_handler_authz_test.go
+++ b/flow-core/internal/tenant/tenant_handler_authz_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Cross-tenant IDOR coverage for the GET-by-id handlers: a TENANT_ADMIN of
@@ -33,7 +34,7 @@ func newAuthzDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/testdb/testdb.go b/flow-core/internal/testdb/testdb.go
new file mode 100644
index 0000000..2a1df51
--- /dev/null
+++ b/flow-core/internal/testdb/testdb.go
@@ -0,0 +1,105 @@
+// Package testdb hands tests a Postgres connection confined to a private,
+// throwaway schema.
+//
+// WHY. The database harnesses in this repo open FLOW_TEST_PG_DSN and then run
+// `DROP TABLE IF EXISTS asset CASCADE` (and a dozen siblings) to get a clean
+// slate. Against the disposable Postgres that CI starts as a service container
+// that is fine. Against any database a human might plausibly point the variable
+// at — a local dev stack, a shared scratch database, a pilot — it is a data-loss
+// bug waiting for one careless export. The harnesses also collide with each
+// other: two packages dropping and creating `asset` at the same moment leave
+// each other with half a schema. CI hides that today by running `go test -p 1`,
+// so the failure only appears for whoever runs `go test ./...` locally, where
+// packages run in parallel by default.
+//
+// HOW. `Scoped` creates a schema named for the test and returns a DSN carrying
+// `search_path=<schema>`. Every unqualified DDL statement the harness already
+// runs then resolves inside that schema: `DROP TABLE asset` drops the copy the
+// test made, never `public.asset`. Nothing in the harness bodies has to change.
+//
+// search_path is passed as a CONNECTION PARAMETER, not as a `SET` statement.
+// sql.DB is a pool: `SET search_path` applies to whichever pooled connection
+// happened to serve it, so under any concurrency the next query can land on a
+// connection still pointed at `public` — which is precisely the confinement
+// failing open, silently, exactly when it matters. As a DSN parameter the
+// server applies it at connection setup, to every connection the pool opens.
+// Verified against postgres:16 with lib/pq: 40 concurrent queries, 0 wrong.
+package testdb
+
+import (
+	"database/sql"
+	"fmt"
+	"regexp"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+// seq disambiguates two schemas created inside the same nanosecond, which does
+// happen for subtests that set up back to back.
+var seq atomic.Uint64
+
+var unsafeChars = regexp.MustCompile(`[^a-z0-9_]+`)
+
+// Scoped creates a throwaway schema and returns a DSN bound to it. The schema
+// is dropped when the test ends.
+//
+// Call it in place of the raw DSN:
+//
+//	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
+//
+// Accepts testing.TB so benchmarks get the same confinement as tests --
+// a benchmark that drops real tables is no less destructive.
+func Scoped(t testing.TB, dsn string) string {
+	t.Helper()
+
+	admin, err := sql.Open("postgres", dsn)
+	if err != nil {
+		t.Fatalf("testdb: open admin connection: %v", err)
+	}
+	defer admin.Close()
+	if err := admin.Ping(); err != nil {
+		t.Fatalf("testdb: ping: %v", err)
+	}
+
+	schema := schemaName(t.Name())
+	if _, err := admin.Exec(`CREATE SCHEMA ` + schema); err != nil {
+		t.Fatalf("testdb: create schema %s: %v", schema, err)
+	}
+
+	t.Cleanup(func() {
+		// A fresh connection: the test's own pool is closed by its cleanup,
+		// and cleanup order between the two is not guaranteed.
+		cleaner, err := sql.Open("postgres", dsn)
+		if err != nil {
+			t.Logf("testdb: could not reopen to drop schema %s: %v", schema, err)
+			return
+		}
+		defer cleaner.Close()
+		if _, err := cleaner.Exec(`DROP SCHEMA IF EXISTS ` + schema + ` CASCADE`); err != nil {
+			t.Logf("testdb: drop schema %s: %v", schema, err)
+		}
+	})
+
+	sep := "?"
+	if strings.Contains(dsn, "?") {
+		sep = "&"
+	}
+	return dsn + sep + "search_path=" + schema
+}
+
+// schemaName derives a readable, unique, injection-safe identifier. The test
+// name is carried through so a leaked schema names its own culprit.
+func schemaName(testName string) string {
+	base := unsafeChars.ReplaceAllString(strings.ToLower(testName), "_")
+	base = strings.Trim(base, "_")
+	if len(base) > 24 {
+		base = base[:24]
+	}
+	if base == "" {
+		base = "test"
+	}
+	// Postgres truncates identifiers at 63 bytes; this stays well inside.
+	return fmt.Sprintf("t_%s_%d_%d", base, time.Now().UnixNano(), seq.Add(1))
+}
diff --git a/flow-core/internal/testdb/testdb_test.go b/flow-core/internal/testdb/testdb_test.go
new file mode 100644
index 0000000..6f6eb4c
--- /dev/null
+++ b/flow-core/internal/testdb/testdb_test.go
@@ -0,0 +1,150 @@
+package testdb
+
+import (
+	"database/sql"
+	"os"
+	"strings"
+	"sync"
+	"testing"
+
+	_ "github.com/lib/pq"
+)
+
+// These tests pin the two properties the whole isolation scheme rests on. Both
+// were verified against postgres:16 before the 29 harnesses were migrated, and
+// both fail silently if someone "simplifies" Scoped into a `SET search_path`.
+
+func dsn(t *testing.T) string {
+	t.Helper()
+	d := os.Getenv("FLOW_TEST_PG_DSN")
+	if d == "" {
+		t.Skip("FLOW_TEST_PG_DSN not set")
+	}
+	return d
+}
+
+// An unqualified DROP TABLE — which is what every migrated harness runs — must
+// not reach a same-named table outside the throwaway schema. This is the whole
+// point: the harnesses keep their destructive DDL, confinement makes it safe.
+func TestScopedDropCannotReachRealTables(t *testing.T) {
+	base := dsn(t)
+
+	// Unique per run. A fixed name in `public` couples one run to the next --
+	// the very defect this package exists to remove, and it duly showed up as a
+	// stale row the first time the suite ran in parallel.
+	canary := "testdb_canary_" + strings.Map(func(r rune) rune {
+		if r >= 'a' && r <= 'z' || r >= '0' && r <= '9' {
+			return r
+		}
+		return '_'
+	}, strings.ToLower(t.Name())) + "_" + schemaName("c")
+
+	admin, err := sql.Open("postgres", base)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	defer admin.Close()
+	if _, err := admin.Exec(`CREATE TABLE public.` + canary + ` (x int)`); err != nil {
+		t.Fatalf("seed canary: %v", err)
+	}
+	t.Cleanup(func() { _, _ = admin.Exec(`DROP TABLE IF EXISTS public.` + canary) })
+	if _, err := admin.Exec(`INSERT INTO public.` + canary + ` VALUES (1)`); err != nil {
+		t.Fatalf("seed row: %v", err)
+	}
+
+	scoped, err := sql.Open("postgres", Scoped(t, base))
+	if err != nil {
+		t.Fatalf("open scoped: %v", err)
+	}
+	defer scoped.Close()
+
+	// Exactly what a harness does: drop, then recreate its own copy.
+	if _, err := scoped.Exec(`DROP TABLE IF EXISTS ` + canary + ` CASCADE`); err != nil {
+		t.Fatalf("scoped drop: %v", err)
+	}
+	if _, err := scoped.Exec(`CREATE TABLE ` + canary + ` (y text)`); err != nil {
+		t.Fatalf("scoped create: %v", err)
+	}
+
+	var n int
+	if err := admin.QueryRow(`SELECT count(*) FROM public.` + canary).Scan(&n); err != nil {
+		t.Fatalf("the scoped harness destroyed a table outside its schema: %v", err)
+	}
+	if n != 1 {
+		t.Errorf("public.%s lost rows: got %d, want 1", canary, n)
+	}
+}
+
+// search_path must be a CONNECTION parameter, not a SET. sql.DB is a pool: a
+// `SET search_path` binds only the connection that served it, so under
+// concurrency later queries land on connections still pointed at public — the
+// confinement failing open exactly when load makes it matter.
+func TestScopedAppliesToEveryPooledConnection(t *testing.T) {
+	base := dsn(t)
+
+	scopedDSN := Scoped(t, base)
+	if !strings.Contains(scopedDSN, "search_path=") {
+		t.Fatalf("Scoped did not bind search_path in the DSN: %q", scopedDSN)
+	}
+
+	db, err := sql.Open("postgres", scopedDSN)
+	if err != nil {
+		t.Fatalf("open: %v", err)
+	}
+	defer db.Close()
+	db.SetMaxOpenConns(8)
+
+	var wg sync.WaitGroup
+	wrong := make(chan string, 64)
+	for i := 0; i < 64; i++ {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			var sp string
+			if err := db.QueryRow("SHOW search_path").Scan(&sp); err != nil {
+				wrong <- "query failed: " + err.Error()
+				return
+			}
+			if !strings.Contains(sp, "t_") {
+				wrong <- sp
+			}
+		}()
+	}
+	wg.Wait()
+	close(wrong)
+
+	if n := len(wrong); n != 0 {
+		t.Errorf("%d/64 pooled connections were not confined (first: %q)", n, <-wrong)
+	}
+}
+
+// Two Scoped calls must never collide, including within the same nanosecond —
+// which is what subtests setting up back to back actually do.
+func TestScopedSchemasAreUnique(t *testing.T) {
+	seen := map[string]bool{}
+	for i := 0; i < 200; i++ {
+		name := schemaName("TestSomething/case")
+		if seen[name] {
+			t.Fatalf("duplicate schema name generated: %s", name)
+		}
+		seen[name] = true
+	}
+}
+
+func TestSchemaNameIsSafeAndReadable(t *testing.T) {
+	// Test names carry slashes, spaces and case; all of that has to become a
+	// legal identifier, and the origin must stay legible so a leaked schema
+	// names its own culprit.
+	got := schemaName("TestFoo/sub case #2")
+	for _, bad := range []string{"/", " ", "#", "'", `"`, ";"} {
+		if strings.Contains(got, bad) {
+			t.Errorf("schema name %q contains unsafe %q", got, bad)
+		}
+	}
+	if !strings.HasPrefix(got, "t_testfoo_sub_case") {
+		t.Errorf("schema name lost its origin: %q", got)
+	}
+	if len(got) > 63 {
+		t.Errorf("schema name exceeds the Postgres identifier limit: %d", len(got))
+	}
+}
diff --git a/flow-core/internal/topology/topology_benchmark_test.go b/flow-core/internal/topology/topology_benchmark_test.go
index 29aa753..8927d80 100644
--- a/flow-core/internal/topology/topology_benchmark_test.go
+++ b/flow-core/internal/topology/topology_benchmark_test.go
@@ -1,6 +1,8 @@
 package topology
 
 import (
+	"flow-core/internal/testdb"
+
 	"database/sql"
 	"fmt"
 	"os"
@@ -20,7 +22,7 @@ func BenchmarkTopologyNeighborsDepth5(b *testing.B) {
 	if dsn == "" {
 		b.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(b, dsn))
 	if err != nil {
 		b.Fatalf("open: %v", err)
 	}
@@ -182,7 +184,7 @@ func benchmarkExpandCTE(b *testing.B, depth int) {
 	if dsn == "" {
 		b.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(b, dsn))
 	if err != nil {
 		b.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/topology/topology_test.go b/flow-core/internal/topology/topology_test.go
index 18c2f61..a63acc5 100644
--- a/flow-core/internal/topology/topology_test.go
+++ b/flow-core/internal/topology/topology_test.go
@@ -15,6 +15,7 @@ import (
 	"unicode/utf8"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 const (
@@ -34,7 +35,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/transport/transport_test.go b/flow-core/internal/transport/transport_test.go
index a7aba8b..623462b 100644
--- a/flow-core/internal/transport/transport_test.go
+++ b/flow-core/internal/transport/transport_test.go
@@ -7,6 +7,7 @@ import (
 	"time"
 
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 func newTestDB(t *testing.T) *sql.DB {
@@ -15,7 +16,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/twin/twin_test.go b/flow-core/internal/twin/twin_test.go
index 40b39ff..90b3794 100644
--- a/flow-core/internal/twin/twin_test.go
+++ b/flow-core/internal/twin/twin_test.go
@@ -12,6 +12,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 	"flow-core/internal/twinstore"
 
 	_ "github.com/lib/pq"
@@ -31,7 +32,7 @@ func newTwinTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
@@ -52,6 +53,11 @@ func setupTwinTables(t *testing.T, db *sql.DB) {
 		`DROP TABLE IF EXISTS twin_registry CASCADE`,
 		`DROP TABLE IF EXISTS twin_model CASCADE`,
 		`DROP TABLE IF EXISTS topology_edge CASCADE`,
+		`DROP TABLE IF EXISTS customer CASCADE`,
+		`DROP TABLE IF EXISTS entity_view CASCADE`,
+		`DROP TABLE IF EXISTS dashboard CASCADE`,
+		`DROP TABLE IF EXISTS device_profile CASCADE`,
+		`DROP TABLE IF EXISTS asset_profile CASCADE`,
 		`DROP TABLE IF EXISTS asset CASCADE`,
 		`DROP TABLE IF EXISTS device CASCADE`,
 		`DROP TABLE IF EXISTS ts_kv_latest CASCADE`,
@@ -65,6 +71,21 @@ func setupTwinTables(t *testing.T, db *sql.DB) {
 			id uuid PRIMARY KEY, created_time bigint, tenant_id uuid,
 			customer_id uuid, name text, type text, label text,
 			additional_info text, version bigint default 1)`,
+		// The expand CTE resolves a relation's endpoints against every entity
+		// table it can point at, so this package needs them present even though its
+		// own fixtures only use assets and devices. They were never declared here:
+		// the tests passed because other packages (internal/customer,
+		// internal/dashboard, ...) had created them in the shared `public` schema
+		// earlier in the run. Once each package got a private schema that borrowed
+		// state vanished and the traversal 500'd -- the isolation did not break
+		// these tests, it revealed what they had been leaning on.
+		//
+		// Only id and tenant_id are read by the join, so the shapes stay minimal.
+		`CREATE TABLE customer (id uuid PRIMARY KEY, tenant_id uuid)`,
+		`CREATE TABLE entity_view (id uuid PRIMARY KEY, tenant_id uuid)`,
+		`CREATE TABLE dashboard (id uuid PRIMARY KEY, tenant_id uuid)`,
+		`CREATE TABLE device_profile (id uuid PRIMARY KEY, tenant_id uuid)`,
+		`CREATE TABLE asset_profile (id uuid PRIMARY KEY, tenant_id uuid)`,
 		`CREATE TABLE topology_edge (
 			tenant_id uuid not null, from_id uuid not null, from_type text not null,
 			to_id uuid not null, to_type text not null, relation_type text not null,
diff --git a/flow-core/internal/twinmodel/normalization_sql_test.go b/flow-core/internal/twinmodel/normalization_sql_test.go
index 37a29b1..c7b9e1a 100644
--- a/flow-core/internal/twinmodel/normalization_sql_test.go
+++ b/flow-core/internal/twinmodel/normalization_sql_test.go
@@ -1,6 +1,8 @@
 package twinmodel
 
 import (
+	"flow-core/internal/testdb"
+
 	"database/sql"
 	"os"
 	"strings"
@@ -14,7 +16,7 @@ func TestSQLNormalizationMatchesGo(t *testing.T) {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open postgres: %v", err)
 	}
diff --git a/flow-core/internal/user/auth_handler_test.go b/flow-core/internal/user/auth_handler_test.go
index afb9b79..c3c07f1 100644
--- a/flow-core/internal/user/auth_handler_test.go
+++ b/flow-core/internal/user/auth_handler_test.go
@@ -15,6 +15,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // newTestDB pulls the postgres pointed at by FLOW_TEST_PG_DSN. Tests
@@ -26,7 +27,7 @@ func newTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/user/crud_user_authz_test.go b/flow-core/internal/user/crud_user_authz_test.go
index 2ac7a1c..dad3b55 100644
--- a/flow-core/internal/user/crud_user_authz_test.go
+++ b/flow-core/internal/user/crud_user_authz_test.go
@@ -14,6 +14,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Privilege-escalation coverage for POST/PUT /api/user
@@ -35,7 +36,7 @@ func newUserAuthzDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	db, err := sql.Open("postgres", dsn)
+	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/flow-core/internal/ws/ws_tenant_test.go b/flow-core/internal/ws/ws_tenant_test.go
index 5142c7c..e4a3b66 100644
--- a/flow-core/internal/ws/ws_tenant_test.go
+++ b/flow-core/internal/ws/ws_tenant_test.go
@@ -15,6 +15,7 @@ import (
 
 	authpkg "flow-core/internal/auth"
 	dbpkg "flow-core/internal/db"
+	"flow-core/internal/testdb"
 )
 
 // Phase 5b — the WebSocket plane was never covered by the Phase 2 HTTP tenant
@@ -48,7 +49,7 @@ func newWSTenantTestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_PG_DSN not set")
 	}
-	pool, err := sql.Open("postgres", dsn)
+	pool, err := sql.Open("postgres", testdb.Scoped(t, dsn))
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/tools/python/test_db_test_isolation.py b/tools/python/test_db_test_isolation.py
new file mode 100644
index 0000000..04d418f
--- /dev/null
+++ b/tools/python/test_db_test_isolation.py
@@ -0,0 +1,130 @@
+"""Database test harnesses must be confined to a throwaway schema.
+
+Why this exists: 29 harnesses opened FLOW_TEST_PG_DSN and ran
+`DROP TABLE IF EXISTS asset CASCADE` (and a dozen siblings) to get a clean
+slate. Against the disposable Postgres CI starts as a service container that is
+fine. Against any database a human might point that variable at -- a local dev
+stack, a shared scratch database, a pilot -- it is a data-loss bug waiting for
+one careless export. They also collided with each other; CI hid that by running
+`go test -p 1`, so the breakage only reached whoever ran `go test ./...`
+locally, where packages run in parallel by default.
+
+`internal/testdb.Scoped` fixes it by handing back a DSN bound to a private
+schema, so the existing destructive DDL resolves inside that schema and can
+never reach `public`. The rule this test enforces is the one that is easy to
+forget: a NEW harness that opens the raw DSN gets none of that protection, and
+nothing about it looks wrong in review.
+
+Migrating those harnesses also surfaced real hidden coupling: internal/twin's
+tests only passed because internal/customer, internal/dashboard and friends had
+created `customer`, `dashboard`, `entity_view`, `device_profile` and
+`asset_profile` in the shared `public` schema earlier in the run. The isolation
+did not break those tests; it revealed what they had been leaning on.
+"""
+import pathlib
+import re
+import unittest
+
+
+ROOT = pathlib.Path(__file__).resolve().parents[2]
+FLOW_CORE = ROOT / "flow-core"
+HELPER = "internal/testdb"
+
+# `sql.Open("postgres", <something>)` where the something is a bare dsn-ish
+# variable rather than a testdb.Scoped(...) call.
+RAW_OPEN = re.compile(r'sql\.Open\(\s*"postgres"\s*,\s*(?!testdb\.Scoped)([A-Za-z_][\w.]*)\s*\)')
+
+
+def test_files():
+    for p in sorted(FLOW_CORE.rglob("*_test.go")):
+        yield p
+
+
+class DBTestIsolationTest(unittest.TestCase):
+    def test_no_harness_opens_an_unconfined_connection(self):
+        offenders = []
+        for path in test_files():
+            text = path.read_text()
+            if "FLOW_TEST_PG_DSN" not in text:
+                continue
+            # The helper's own tests must open an UNCONFINED connection: that is
+            # how they place a canary table in `public` and prove the confined
+            # side cannot reach it. Exempting them is not a loophole — without
+            # the raw connection there is nothing to prove.
+            if path.parent.name == "testdb":
+                continue
+            # A harness that builds its own schema by hand predates the helper
+            # and is already confined; it just does it the long way.
+            if "CREATE SCHEMA" in text:
+                continue
+            for m in RAW_OPEN.finditer(text):
+                offenders.append(f"{path.relative_to(ROOT)} -> sql.Open(\"postgres\", {m.group(1)})")
+
+        self.assertEqual(
+            [],
+            offenders,
+            "these harnesses open FLOW_TEST_PG_DSN without confinement, so their "
+            "DROP TABLE statements reach whatever database the variable points at:\n  "
+            + "\n  ".join(offenders)
+            + f"\n\nWrap the DSN: sql.Open(\"postgres\", testdb.Scoped(t, dsn)) "
+            f"(package flow-core/{HELPER}).",
+        )
+
+    def test_the_helper_binds_search_path_as_a_connection_parameter(self):
+        """Not as a `SET`.
+
+        sql.DB is a pool. `SET search_path` applies only to the connection that
+        served it, so under any concurrency a later query lands on a connection
+        still pointed at `public` — the confinement failing open, silently, and
+        exactly when load makes it matter. As a DSN parameter the server applies
+        it at connection setup, to every connection the pool opens.
+        """
+        helper = (FLOW_CORE / HELPER / "testdb.go").read_text()
+        self.assertIn('"search_path="', helper)
+        # Strip comments before asserting absence: the helper's own doc comment
+        # explains why `SET search_path` is wrong, so a bare substring check
+        # matches the warning rather than the code.
+        code = "\n".join(
+            l for l in helper.splitlines() if not l.lstrip().startswith("//")
+        )
+        self.assertNotIn("SET search_path", code)
+
+    def test_catalogue_queries_are_scoped_to_the_current_schema(self):
+        """A catalogue query spans the whole database, not the test's schema.
+
+        `SELECT count(*) FROM information_schema.columns WHERE table_name='policy'`
+        counts that table in EVERY schema, so once packages run concurrently and
+        each holds its own copy the assertion reports "found 2" for a perfectly
+        correct migration. `pg_constraint` has the same shape: constraint names
+        are unique per schema, not per database, and QueryRow silently takes
+        whichever row comes first.
+
+        Measured before the fix: 2 of 5 parallel runs failed. It never showed up
+        under CI's `-p 1`, which is exactly why it survived.
+        """
+        offenders = []
+        for path in test_files():
+            text = path.read_text()
+            for catalogue in ("information_schema.", "pg_constraint", "pg_indexes"):
+                if catalogue not in text:
+                    continue
+                # Every query touching a catalogue must constrain the namespace.
+                if "current_schema()" not in text:
+                    offenders.append(f"{path.relative_to(ROOT)} ({catalogue})")
+        self.assertEqual(
+            [],
+            offenders,
+            "these tests query a database-wide catalogue without pinning it to "
+            "current_schema(), so they see other packages' concurrent schemas:\n  "
+            + "\n  ".join(sorted(set(offenders))),
+        )
+
+    def test_the_helper_drops_what_it_creates(self):
+        helper = (FLOW_CORE / HELPER / "testdb.go").read_text()
+        self.assertIn("CREATE SCHEMA", helper)
+        self.assertIn("DROP SCHEMA IF EXISTS", helper)
+        self.assertIn("t.Cleanup", helper)
+
+
+if __name__ == "__main__":
+    unittest.main()

From 06a6ada8cf46a8e724754fae5557b8bdd183b631 Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Fri, 21 Aug 2026 13:22:00 +0000
Subject: [PATCH 5/6] fix(migrations): make 0008 actually roll back, and
 round-trip cleanly
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The up-down-up gate has been red on develop since 2026-08-10 and on main since
2026-08-13. It only ran again now because this branch touches a path under
flow-core/internal/migrations/, and the failure is real:

  pq: cannot drop column security_status of table device because other objects
      depend on it, view device_info_active_attribute_view depends on ...

device_info_active_attribute_view, device_info_active_ts_view and
device_info_view are all `SELECT d.*` over `device`, so they depend on every
column it has. Postgres refuses the DROP COLUMN, and 0008 could never roll back.
`DROP COLUMN ... CASCADE` would have "worked" by silently deleting all three
views and leaving nothing that recreates them — a rollback that quietly breaks
every device list query is worse than one that refuses.

The down migration now drops the views, drops the column, and rebuilds them from
the same definitions the baseline installs. Rebuilt after the column is gone,
`SELECT d.*` yields exactly the pre-migration shape.

That alone still left the cycle asymmetric, which the gate does not check and
which was only visible on inspection: a view's column list is fixed at creation
time, so after up -> down -> up the views came back WITHOUT security_status —
one column short, silently. Nothing in flow-core reads it through a view today
(every query goes straight to `device`), so this was a latent trap rather than a
live bug, but "the schema after a round trip differs from the schema before" is
exactly what this gate exists to prevent. The up migration now rebuilds the
views too, so the pair is symmetric.

Verified locally against postgres:16 by reproducing the gate step for step:
baseline -> force 1 -> up -> goto 1 -> up, then asserting the column, all three
views, their exposed columns, the CHECK constraint and the index are all back.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../0008_device_security_status.down.sql      | 52 +++++++++++++++++++
 .../0008_device_security_status.up.sql        | 38 ++++++++++++++
 2 files changed, 90 insertions(+)

diff --git a/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql b/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql
index 6d8819a..65bdb27 100644
--- a/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql
+++ b/flow-core/internal/migrations/migrations/0008_device_security_status.down.sql
@@ -1,3 +1,55 @@
+-- Rolling this back means removing a column three views depend on.
+--
+-- device_info_active_attribute_view, device_info_active_ts_view and
+-- device_info_view are all `SELECT d.*` over `device` (installed by the baseline
+-- schema, k8s/helm/thingsflow/files/sql/02_schema-views.sql), so they pick up
+-- every column the table has -- including this one. Postgres therefore refuses
+-- the DROP COLUMN outright and this migration could never roll back:
+--
+--   pq: cannot drop column security_status of table device because other
+--       objects depend on it, view device_info_active_attribute_view depends on
+--       column security_status of table device
+--
+-- `DROP COLUMN ... CASCADE` would "succeed" by silently deleting all three
+-- views and leaving the schema three objects short, with nothing in the up
+-- direction that recreates them -- a rollback that quietly breaks every device
+-- list query is worse than one that refuses.
+--
+-- So: drop the views, drop the column, rebuild the views from the same
+-- definitions the baseline installs. Rebuilt after the column is gone, `SELECT
+-- d.*` yields exactly the pre-migration shape, which is the point.
+--
+-- Keep these definitions in step with 02_schema-views.sql. They are duplicated
+-- deliberately: a down migration has to restore the state it disturbed, and it
+-- cannot reach into the baseline to do it.
+
+DROP VIEW IF EXISTS device_info_view CASCADE;
+DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE;
+DROP VIEW IF EXISTS device_info_active_ts_view CASCADE;
+
 DROP INDEX IF EXISTS idx_device_tenant_security_status;
 ALTER TABLE device DROP CONSTRAINT IF EXISTS device_security_status_chk;
 ALTER TABLE device DROP COLUMN IF EXISTS security_status;
+
+CREATE OR REPLACE VIEW device_info_active_attribute_view AS
+SELECT d.*
+       , c.title as customer_title
+       , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public
+       , d.type as device_profile_name
+       , COALESCE(da.bool_v, FALSE) as active
+FROM device d
+         LEFT JOIN customer c ON c.id = d.customer_id
+         LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary  where key = 'active');
+
+CREATE OR REPLACE VIEW device_info_active_ts_view AS
+SELECT d.*
+       , c.title as customer_title
+       , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public
+       , d.type as device_profile_name
+       , COALESCE(dt.bool_v, FALSE) as active
+FROM device d
+         LEFT JOIN customer c ON c.id = d.customer_id
+         LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active');
+
+-- Depends on the attribute view above, so it is rebuilt last.
+CREATE OR REPLACE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view;
diff --git a/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql b/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql
index 883bdf6..8134efc 100644
--- a/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql
+++ b/flow-core/internal/migrations/migrations/0008_device_security_status.up.sql
@@ -20,3 +20,41 @@ END $$;
 
 CREATE INDEX IF NOT EXISTS idx_device_tenant_security_status
     ON device (tenant_id, security_status);
+
+-- Rebuild the three device_info views so they pick up the new column.
+--
+-- They are `SELECT d.*` over `device`, so their column list is fixed at
+-- creation time: a view built before this migration keeps the OLD shape and
+-- never shows security_status. Without this, up -> down -> up leaves a schema
+-- that differs from the one it started with -- the views come back one column
+-- short, silently, and only a view-based query would ever notice.
+--
+-- CREATE OR REPLACE cannot be used: it refuses to change a view's column list.
+-- Definitions mirror k8s/helm/thingsflow/files/sql/02_schema-views.sql; keep
+-- the two in step.
+DROP VIEW IF EXISTS device_info_view CASCADE;
+DROP VIEW IF EXISTS device_info_active_attribute_view CASCADE;
+DROP VIEW IF EXISTS device_info_active_ts_view CASCADE;
+
+CREATE VIEW device_info_active_attribute_view AS
+SELECT d.*
+       , c.title as customer_title
+       , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public
+       , d.type as device_profile_name
+       , COALESCE(da.bool_v, FALSE) as active
+FROM device d
+         LEFT JOIN customer c ON c.id = d.customer_id
+         LEFT JOIN attribute_kv da ON da.entity_id = d.id AND da.attribute_type = 2 AND da.attribute_key = (select key_id from key_dictionary  where key = 'active');
+
+CREATE VIEW device_info_active_ts_view AS
+SELECT d.*
+       , c.title as customer_title
+       , COALESCE((c.additional_info::json->>'isPublic')::bool, FALSE) as customer_is_public
+       , d.type as device_profile_name
+       , COALESCE(dt.bool_v, FALSE) as active
+FROM device d
+         LEFT JOIN customer c ON c.id = d.customer_id
+         LEFT JOIN ts_kv_latest dt ON dt.entity_id = d.id and dt.key = (select key_id from key_dictionary where key = 'active');
+
+-- Depends on the attribute view above, so it is rebuilt last.
+CREATE VIEW device_info_view AS SELECT * FROM device_info_active_attribute_view;

From 3241d913d80298ad077fbc1e6b81bf57a9f4db3e Mon Sep 17 00:00:00 2001
From: "Nilson.Henao" <nilson.henao@uqtr.ca>
Date: Fri, 21 Aug 2026 13:35:33 +0000
Subject: [PATCH 6/6] =?UTF-8?q?fix(test):=20do=20not=20confine=20the=20Que?=
 =?UTF-8?q?stDB=20harness=20=E2=80=94=20it=20has=20no=20schemas?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The mechanical migration wrapped every `sql.Open("postgres", dsn)` in
testdb.Scoped, including the one in internal/telemetry whose DSN points at
QuestDB. QuestDB speaks the Postgres wire protocol but has no schemas, so
CREATE SCHEMA and search_path both fail there and two working tests started
failing.

It stayed invisible locally because FLOW_TEST_QUESTDB_DSN is normally unset and
the tests skip; only CI sets it. Verified the fix the way I should have verified
the migration: against a real questdb:7.4.2, with both DSNs set, serially and in
parallel.

Confinement is also unnecessary there — the target is a dedicated QuestDB
instance, not a shared Postgres someone might also be using.

The isolation guard now rejects wrapping a non-Postgres DSN, and reads CODE
rather than raw source. A substring assertion over raw source matches the very
comment explaining why something must not be done — which it did here, flagging
the new "NOT testdb.Scoped" comment as an offender.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
---
 .../telemetry/questdb_integration_test.go     |  9 ++--
 tools/python/test_db_test_isolation.py        | 50 +++++++++++++++----
 2 files changed, 47 insertions(+), 12 deletions(-)

diff --git a/flow-core/internal/telemetry/questdb_integration_test.go b/flow-core/internal/telemetry/questdb_integration_test.go
index 2517899..ff20c2f 100644
--- a/flow-core/internal/telemetry/questdb_integration_test.go
+++ b/flow-core/internal/telemetry/questdb_integration_test.go
@@ -1,8 +1,6 @@
 package telemetry
 
 import (
-	"flow-core/internal/testdb"
-
 	"database/sql"
 	"fmt"
 	"os"
@@ -34,7 +32,12 @@ func newQuestDB(t *testing.T) *sql.DB {
 	if dsn == "" {
 		t.Skip("FLOW_TEST_QUESTDB_DSN not set")
 	}
-	db, err := sql.Open("postgres", testdb.Scoped(t, dsn))
+	// NOT testdb.Scoped: this DSN points at QuestDB, which speaks the
+	// Postgres wire protocol but has no schemas -- CREATE SCHEMA and
+	// search_path both fail there. Confinement is neither available nor
+	// needed: the target is a dedicated QuestDB instance, not a shared
+	// Postgres a human might also be using.
+	db, err := sql.Open("postgres", dsn)
 	if err != nil {
 		t.Fatalf("open: %v", err)
 	}
diff --git a/tools/python/test_db_test_isolation.py b/tools/python/test_db_test_isolation.py
index 04d418f..38afaa0 100644
--- a/tools/python/test_db_test_isolation.py
+++ b/tools/python/test_db_test_isolation.py
@@ -35,6 +35,17 @@
 RAW_OPEN = re.compile(r'sql\.Open\(\s*"postgres"\s*,\s*(?!testdb\.Scoped)([A-Za-z_][\w.]*)\s*\)')
 
 
+def code_of(path):
+    """Source with `//` comment lines removed.
+
+    A substring assertion over raw source matches the very comment that explains
+    why something must not be done, so every check here reads code only.
+    """
+    return "\n".join(
+        l for l in path.read_text().splitlines() if not l.lstrip().startswith("//")
+    )
+
+
 def test_files():
     for p in sorted(FLOW_CORE.rglob("*_test.go")):
         yield p
@@ -70,6 +81,33 @@ def test_no_harness_opens_an_unconfined_connection(self):
             f"(package flow-core/{HELPER}).",
         )
 
+    def test_non_postgres_backends_are_not_wrapped(self):
+        """QuestDB speaks the Postgres wire protocol but has no schemas.
+
+        `CREATE SCHEMA` and `search_path` both fail against it, so wrapping its
+        DSN in Scoped turns a working test into a failing one. This is not
+        hypothetical: the mechanical migration did exactly that, and it stayed
+        invisible locally because FLOW_TEST_QUESTDB_DSN is normally unset and the
+        test skips — it only surfaced in CI, which does set it.
+
+        Confinement is also unnecessary there: the target is a dedicated QuestDB
+        instance, not a shared Postgres someone might also be using.
+        """
+        offenders = []
+        for path in test_files():
+            text = code_of(path)
+            if "testdb.Scoped(" not in text:
+                continue
+            for foreign in ("FLOW_TEST_QUESTDB_DSN", "GREPTIME"):
+                if foreign in text:
+                    offenders.append(f"{path.relative_to(ROOT)} ({foreign})")
+        self.assertEqual(
+            [],
+            offenders,
+            "these tests wrap a non-Postgres DSN in testdb.Scoped, which needs "
+            "CREATE SCHEMA and search_path:\n  " + "\n  ".join(sorted(set(offenders))),
+        )
+
     def test_the_helper_binds_search_path_as_a_connection_parameter(self):
         """Not as a `SET`.
 
@@ -79,15 +117,9 @@ def test_the_helper_binds_search_path_as_a_connection_parameter(self):
         exactly when load makes it matter. As a DSN parameter the server applies
         it at connection setup, to every connection the pool opens.
         """
-        helper = (FLOW_CORE / HELPER / "testdb.go").read_text()
-        self.assertIn('"search_path="', helper)
-        # Strip comments before asserting absence: the helper's own doc comment
-        # explains why `SET search_path` is wrong, so a bare substring check
-        # matches the warning rather than the code.
-        code = "\n".join(
-            l for l in helper.splitlines() if not l.lstrip().startswith("//")
-        )
-        self.assertNotIn("SET search_path", code)
+        path = FLOW_CORE / HELPER / "testdb.go"
+        self.assertIn('"search_path="', path.read_text())
+        self.assertNotIn("SET search_path", code_of(path))
 
     def test_catalogue_queries_are_scoped_to_the_current_schema(self):
         """A catalogue query spans the whole database, not the test's schema.