fix(review): quick FT path enforces 3-state exclusivity + twin propagation; changelog + twin caveat

set_false_trigger now clears reviewed_real when flagging false_trigger=True
(mirrors update_event_review's exclusivity), and the quick
PATCH /db/events/{id}/false_trigger endpoint now calls
propagate_review_to_twins after the flag write, matching the sidecar PATCH
path's try/except-with-log.warning pattern. Previously the quick path could
leave both flags set and never touched twins.

Also corrects the v0.25.0 CHANGELOG bullet (exclusivity was not actually
enforced on the quick path until this commit) and adds a caveat comment on
find_twins about rare clamped/saturated-PVS false-positive twin matches.

CHANGELOG.md

@@ -8,6 +8,66 @@ All notable changes to seismo-relay are documented here.
 
 ---
 
+## v0.25.0 — 2026-08-25
+
+**reviewed_real 3-state review flag + twin review-propagation.** The
+`events` table and the `/db/events` feed now carry `reviewed_real`, a
+3-state review flag mutually exclusive with `false_trigger`, mirrored from
+the sidecar review block — plus histogram/waveform twin review-propagation
+(flagging one flags both, matched by serial + identical PVS + timestamp
+window).
+
+### Added
+
+- **`events` column** `reviewed_real` — `INTEGER NOT NULL DEFAULT 0`, added
+  via the existing incremental `_migrate` ADD COLUMN pass (auto-migrates on
+  `SeismoDb()` construction, no manual migration). `query_events` /
+  `get_event` (and thus `/db/events`) return it automatically (`SELECT *`).
+- **Mutual exclusivity with `false_trigger`** — setting `reviewed_real=1`
+  clears `false_trigger`, and vice versa, enforced on both review paths:
+  the sidecar review PATCH (`update_event_review`) and the quick
+  `PATCH /db/events/{id}/false_trigger` endpoint (`set_false_trigger`).
+- **`find_twins`** — matches an event's histogram/waveform twins by serial +
+  identical peak-vector-sum + a timestamp window.
+- **`propagate_review_to_twins`** — copies an event's `false_trigger`/
+  `reviewed_real` state onto its twins, wired into both the
+  `PATCH /db/events/{id}/sidecar` review path and the quick
+  `PATCH /db/events/{id}/false_trigger` path, so flagging one flags both
+  regardless of which endpoint made the change.
+
+---
+
+## v0.24.0 — 2026-08-22
+
+**Waveform-shape metrics on events.** The `events` table and the `/db/events`
+feed now carry per-event crest factor and points-near-peak, computed from the
+decoded waveform samples at ingest — groundwork for Terra-View's
+false-trigger detection (Phase B).
+
+### Added
+
+- **`events` columns** `shape_crest_factor`, `shape_near_peak_count`,
+  `shape_sample_count`, `shape_axis`. Added via the existing incremental
+  `_migrate` ADD COLUMN pass — **auto-migrates on `SeismoDb()` construction,
+  no manual migration**. `query_events` / `get_event` (and thus `/db/events`)
+  return them automatically (`SELECT *`).
+- **Populated at ingest** — crest factor + near-peak-count are computed from
+  the decoded samples in every save path (`shape_from_h5`/
+  `shape_from_samples`), and `insert_events` persists them on INSERT and
+  UPSERT.
+- **Backfill** `scripts/backfill_event_shape.py` — fills the columns for
+  existing events from their on-disk `.h5` waveform samples (idempotent,
+  UPDATE-only).
+
+### Upgrade Notes
+
+Run the backfill once after deploying, against the events DB + waveform store:
+`python3 scripts/backfill_event_shape.py --db-path <seismo_relay.db> --store-root <waveforms/>`
+Events with no decodable samples (or no waveform file) stay NULL and render
+"—" downstream.
+
+---
+
 ## v0.23.0 — 2026-08-04
 
 **Per-channel ZC frequency in the events store.** The `events` table and the

docs/superpowers/plans/2026-08-21-waveform-ft-detection-phase-a.md

@@ -0,0 +1,628 @@
+# Waveform-shape FT detection — Phase A (seismo-relay) Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Compute per-event waveform-shape metrics (crest factor + points-near-peak) in SFM and store them as `events` columns, populated at ingest and by a backfill script, so Terra-View can read them from `/db/events`.
+
+**Architecture:** A pure DSP module (`sfm/shape_metrics.py`) turns decoded `.h5` samples into shape metrics. New nullable `events.shape_*` columns are added via the existing `_migrate` ADD COLUMN loop. The three `WaveformStore.save*` paths compute shape from the just-written `.h5` and hand it to `insert_events`; a backfill script does the same over existing events. Mirrors exactly how per-channel ZC frequency was added.
+
+**Tech Stack:** Python 3.10, numpy, h5py, sqlite3 (raw), pytest. seismo-relay venv: `/home/serversdown/seismo-relay/.venv/bin/python3`.
+
+## Global Constraints
+
+- Metrics are read from the `.h5` `samples/{Tran,Vert,Long}` float32 arrays (physical in/s). The measured channel is the max-|peak| geophone channel.
+- All new columns are nullable; histogram records and events without usable samples store NULL (Terra-View falls back to cheap signals). Legacy rows stay valid.
+- Crest factor = `max(|x|) / rms(x)`; near-peak count = number of samples with `|x| ≥ 0.5·peak`. Threshold `0.5` is a module constant so calibration can tune it.
+- No manual migration: columns are added in `SeismoDb._migrate`, run at `SeismoDb()` construction.
+- `/db/events` needs no change — it returns all columns via `SELECT *` (verify with a test).
+- Run tests with `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest`.
+
+---
+
+### Task 1: Shape DSP module
+
+**Files:**
+- Create: `sfm/shape_metrics.py`
+- Test: `tests/test_shape_metrics.py`
+
+**Interfaces:**
+- Produces:
+  - `channel_shape(x) -> dict | None` — `{"crest_factor": float, "near_peak_count": int, "sample_count": int}` or None for unusable input (size < 2, flat, all-zero).
+  - `shape_from_samples(chans: dict[str, ArrayLike]) -> dict | None` — picks the max-peak geophone channel; returns `{"crest_factor","near_peak_count","sample_count","axis"}` or None.
+  - `shape_from_h5(path) -> dict | None` — reads `samples/{Tran,Vert,Long}` and delegates to `shape_from_samples`; None on any read error.
+  - Constant `NEAR_PEAK_FRACTION = 0.5`.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_shape_metrics.py
+import numpy as np
+from sfm.shape_metrics import channel_shape, shape_from_samples
+
+def test_needle_spike_high_crest_few_near_peak():
+    x = np.zeros(1024); x[500] = 1.0          # one isolated spike
+    s = channel_shape(x)
+    assert s["sample_count"] == 1024
+    assert s["crest_factor"] > 15             # peak towers over rms
+    assert s["near_peak_count"] <= 3          # almost nothing near the peak
+
+def test_ringing_low_crest_many_near_peak():
+    t = np.arange(1024)
+    x = np.sin(2*np.pi*t/32) * np.exp(-t/4000)  # decaying oscillation
+    s = channel_shape(x)
+    assert s["crest_factor"] < 6
+    assert s["near_peak_count"] > 30          # many samples near the peak
+
+def test_channel_shape_none_for_unusable():
+    assert channel_shape(np.zeros(1024)) is None   # flat / all-zero
+    assert channel_shape(np.array([1.0])) is None  # too short
+
+def test_shape_from_samples_picks_max_peak_axis():
+    chans = {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": np.zeros(1024)}
+    chans["Long"][10] = 0.5
+    chans["Vert"] = np.sin(np.arange(1024)/5) * 0.01
+    s = shape_from_samples(chans)
+    assert s["axis"] == "Long"                # Long has the biggest peak
+    assert s["near_peak_count"] <= 3
+
+def test_shape_from_samples_none_when_no_geo():
+    assert shape_from_samples({"MicL": np.ones(1024)}) is None
+```
+
+- [ ] **Step 2: Run test to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics.py -q`
+Expected: FAIL — `ModuleNotFoundError: sfm.shape_metrics`.
+
+- [ ] **Step 3: Write minimal implementation**
+
+```python
+# sfm/shape_metrics.py
+"""Waveform-shape metrics for false-trigger detection.
+
+A false trigger is an isolated impulse (quiet → spike → quiet); a real event
+rings for many cycles. Two numbers separate them: crest factor (how far the
+peak stands above the typical sample) and how many samples sit near the peak.
+"""
+from __future__ import annotations
+import numpy as np
+
+_GEO_CHANNELS = ("Tran", "Vert", "Long")
+NEAR_PEAK_FRACTION = 0.5   # a sample "near the peak" is >= this * peak amplitude
+
+
+def channel_shape(x) -> dict | None:
+    x = np.asarray(x, dtype=float)
+    if x.size < 2:
+        return None
+    peak = float(np.max(np.abs(x)))
+    if peak <= 0:
+        return None
+    rms = float(np.sqrt(np.mean(x ** 2)))
+    if rms <= 0:
+        return None
+    near = int(np.sum(np.abs(x) >= NEAR_PEAK_FRACTION * peak))
+    return {"crest_factor": peak / rms, "near_peak_count": near,
+            "sample_count": int(x.size)}
+
+
+def shape_from_samples(chans: dict) -> dict | None:
+    best_axis, best_peak, best_x = None, -1.0, None
+    for ax in _GEO_CHANNELS:
+        x = chans.get(ax)
+        if x is None:
+            continue
+        x = np.asarray(x, dtype=float)
+        if x.size < 2:
+            continue
+        p = float(np.max(np.abs(x)))
+        if p > best_peak:
+            best_axis, best_peak, best_x = ax, p, x
+    if best_axis is None:
+        return None
+    s = channel_shape(best_x)
+    if s is None:
+        return None
+    s["axis"] = best_axis
+    return s
+
+
+def shape_from_h5(path) -> dict | None:
+    import h5py
+    try:
+        with h5py.File(path, "r") as f:
+            chans = {ax: f[f"samples/{ax}"][:] for ax in _GEO_CHANNELS
+                     if f"samples/{ax}" in f}
+    except Exception:
+        return None
+    return shape_from_samples(chans)
+```
+
+- [ ] **Step 4: Run test to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics.py -q`
+Expected: PASS (5 tests).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/shape_metrics.py tests/test_shape_metrics.py
+git commit -m "feat(shape): crest-factor + points-near-peak waveform metrics"
+```
+
+---
+
+### Task 2: shape_from_h5 round-trips a real .h5
+
+**Files:**
+- Test: `tests/test_shape_metrics_h5.py`
+
+**Interfaces:**
+- Consumes: `sfm.shape_metrics.shape_from_h5`; `h5py`.
+
+- [ ] **Step 1: Write the failing test** (writes a tiny .h5 the same shape SFM writes, then reads it back)
+
+```python
+# tests/test_shape_metrics_h5.py
+import numpy as np, h5py
+from sfm.shape_metrics import shape_from_h5
+
+def _write_h5(path, chans):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k, v in chans.items():
+            g.create_dataset(k, data=np.asarray(v, dtype="float32"))
+
+def test_shape_from_h5_reads_dominant_axis(tmp_path):
+    p = tmp_path / "ev.h5"
+    long = np.zeros(1024, dtype="float32"); long[100] = 0.48
+    _write_h5(p, {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": long,
+                  "MicL": np.ones(1024)})
+    s = shape_from_h5(str(p))
+    assert s["axis"] == "Long" and s["near_peak_count"] <= 3
+
+def test_shape_from_h5_none_on_missing_or_degenerate(tmp_path):
+    assert shape_from_h5(str(tmp_path / "nope.h5")) is None
+    p = tmp_path / "degen.h5"
+    _write_h5(p, {"Tran": np.zeros(1), "Vert": np.zeros(1), "Long": np.zeros(1)})
+    assert shape_from_h5(str(p)) is None
+```
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics_h5.py -q`
+Expected: FAIL (assertion or, if Task 1 incomplete, import error).
+
+- [ ] **Step 3: Implementation** — none needed; `shape_from_h5` already exists from Task 1. If a test fails, fix `shape_from_h5` (not the test).
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_metrics_h5.py -q`
+Expected: PASS (2 tests).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add tests/test_shape_metrics_h5.py
+git commit -m "test(shape): shape_from_h5 round-trips a real .h5"
+```
+
+---
+
+### Task 3: Add shape columns to the events schema + migration
+
+**Files:**
+- Modify: `sfm/database.py` — `_SCHEMA` CREATE TABLE `events` (after `mic_zc_above_range`); `_migrate` ADD COLUMN loop (the tuple around line 205-218).
+- Test: `tests/test_shape_columns.py`
+
+**Interfaces:**
+- Produces: `events` columns `shape_crest_factor REAL`, `shape_near_peak_count INTEGER`, `shape_sample_count INTEGER`, `shape_axis TEXT`.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_shape_columns.py
+import sqlite3
+from sfm.database import SeismoDb
+
+_SHAPE_COLS = {"shape_crest_factor", "shape_near_peak_count",
+               "shape_sample_count", "shape_axis"}
+
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+
+def test_fresh_db_has_shape_columns(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    assert _SHAPE_COLS <= _cols(db)
+
+def test_existing_db_migrates_shape_columns(tmp_path):
+    p = tmp_path / "s.db"
+    db = SeismoDb(p)
+    with sqlite3.connect(p) as c:      # simulate an older DB missing the columns
+        for col in _SHAPE_COLS:
+            c.execute(f"ALTER TABLE events DROP COLUMN {col}")
+    assert not (_SHAPE_COLS <= _cols(SeismoDb(p)))  # sanity: dropped
+    SeismoDb(p)                        # re-open triggers _migrate
+    assert _SHAPE_COLS <= _cols(SeismoDb(p))
+```
+
+> Note: sqlite `DROP COLUMN` needs sqlite ≥ 3.35 (bundled py3.10 has it). If the runner's sqlite lacks it, replace the "simulate older DB" block with building a table without the columns; keep the assertion that re-open adds them.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_columns.py -q`
+Expected: FAIL — columns absent.
+
+- [ ] **Step 3: Implementation**
+
+In `_SCHEMA`, after the `mic_zc_above_range      INTEGER,` line in the `events` CREATE TABLE, add:
+
+```
+    shape_crest_factor      REAL,                   -- peak / rms of the triggering channel
+    shape_near_peak_count   INTEGER,                -- samples >= 0.5 * peak (FT: few; real: many)
+    shape_sample_count      INTEGER,                -- total samples (to normalize near_peak_count)
+    shape_axis              TEXT,                   -- geophone channel measured ("Tran"/"Vert"/"Long")
+```
+
+In `_migrate`, extend the ADD COLUMN tuple (the `for col, ddl in (...)` list) with:
+
+```python
+            ("shape_crest_factor",    "REAL"),
+            ("shape_near_peak_count", "INTEGER"),
+            ("shape_sample_count",    "INTEGER"),
+            ("shape_axis",            "TEXT"),
+```
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_shape_columns.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/database.py tests/test_shape_columns.py
+git commit -m "feat(db): shape_* columns on events (+ auto-migrate)"
+```
+
+---
+
+### Task 4: insert_events persists shape from the waveform record
+
+**Files:**
+- Modify: `sfm/database.py` — `insert_events` INSERT (column list + placeholders + values) and the UPSERT `UPDATE` block.
+- Test: `tests/test_insert_events_shape.py`
+
+**Interfaces:**
+- Consumes: a `waveform_records` rec dict that may carry `shape_crest_factor`, `shape_near_peak_count`, `shape_sample_count`, `shape_axis`.
+- Produces: those four values stored on the row; refreshed on UPSERT.
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_insert_events_shape.py
+from sfm.database import SeismoDb
+from tests.helpers_events import make_event  # existing helper used by other insert tests
+
+def test_insert_stores_shape_from_record(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = make_event(serial="BE1", key="0111abcd")
+    rec = {ev._waveform_key.hex(): {
+        "filename": "F.CE0W", "filesize": 10,
+        "shape_crest_factor": 34.0, "shape_near_peak_count": 3,
+        "shape_sample_count": 1024, "shape_axis": "Long"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec)
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 34.0
+    assert row["shape_near_peak_count"] == 3
+    assert row["shape_axis"] == "Long"
+```
+
+> If `tests/helpers_events.make_event` doesn't exist, build the `Event` inline the way `tests/test_zc_freq_columns.py` does (copy its event-construction helper). Keep the assertion identical.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_insert_events_shape.py -q`
+Expected: FAIL — `KeyError`/`sqlite3.OperationalError` (columns not in INSERT) or values are NULL.
+
+- [ ] **Step 3: Implementation**
+
+In `insert_events` INSERT: add `shape_crest_factor, shape_near_peak_count, shape_sample_count, shape_axis` to the column list, add four `?` placeholders, and add these to the VALUES tuple (after the `mic_zc_above_range` value):
+
+```python
+                            rec.get("shape_crest_factor"),
+                            rec.get("shape_near_peak_count"),
+                            rec.get("shape_sample_count"),
+                            rec.get("shape_axis"),
+```
+
+In the UPSERT `UPDATE ... SET`: add
+
+```sql
+                               shape_crest_factor    = COALESCE(?, shape_crest_factor),
+                               shape_near_peak_count = COALESCE(?, shape_near_peak_count),
+                               shape_sample_count    = COALESCE(?, shape_sample_count),
+                               shape_axis            = COALESCE(?, shape_axis),
+```
+
+and the matching params (before `serial, ts`):
+
+```python
+                            rec.get("shape_crest_factor") if rec else None,
+                            rec.get("shape_near_peak_count") if rec else None,
+                            rec.get("shape_sample_count") if rec else None,
+                            rec.get("shape_axis") if rec else None,
+```
+
+(`COALESCE` on UPSERT so a re-import that lacks samples doesn't wipe a previously-computed shape.)
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_insert_events_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/database.py tests/test_insert_events_shape.py
+git commit -m "feat(db): insert_events persists shape_* from waveform record"
+```
+
+---
+
+### Task 5: Populate shape at ingest (the three save paths)
+
+**Files:**
+- Modify: `sfm/waveform_store.py` — in `save`, `save_imported_bw`, `save_imported_idf`, after the `.h5` is written, add its shape to the returned `rec` dict.
+- Test: `tests/test_save_shape.py`
+
+**Interfaces:**
+- Consumes: `sfm.shape_metrics.shape_from_h5`.
+- Produces: `save*` return dicts carry `shape_crest_factor / shape_near_peak_count / shape_sample_count / shape_axis` (present only when the `.h5` had usable samples).
+
+- [ ] **Step 1: Write the failing test** (drives the BW-import path, which the existing suite already exercises)
+
+```python
+# tests/test_save_shape.py
+from sfm.waveform_store import WaveformStore
+from tests.helpers_bw import sample_bw_bytes, sample_serial  # reuse existing import-test fixtures
+
+def test_save_imported_bw_attaches_shape(tmp_path):
+    store = WaveformStore(tmp_path / "waveforms")
+    ev, rec = store.save_imported_bw(sample_bw_bytes(), serial=sample_serial())
+    # A real BW waveform → shape present with a geo axis.
+    assert rec.get("shape_axis") in ("Tran", "Vert", "Long")
+    assert rec["shape_crest_factor"] > 0
+    assert rec["shape_sample_count"] > 200
+```
+
+> Reuse whatever fixture `tests/test_save_imported_bw*.py` already uses for BW bytes; match its import. If the existing BW fixture produces a degenerate/short waveform, use the fixture from the test that asserts a full h5.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_save_shape.py -q`
+Expected: FAIL — `rec` has no `shape_*` keys.
+
+- [ ] **Step 3: Implementation**
+
+Add a helper near the top of `WaveformStore` methods (module-level import `from sfm.shape_metrics import shape_from_h5`). In each of `save`, `save_imported_bw`, `save_imported_idf`, immediately before building/returning the `rec` dict — and only when the `.h5` was written (i.e. `hdf5_filename`/`hdf5_path` is set) — compute and merge:
+
+```python
+        shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        # ... in the returned rec dict literal, add:
+        #   **(shape and {
+        #       "shape_crest_factor":    shape["crest_factor"],
+        #       "shape_near_peak_count": shape["near_peak_count"],
+        #       "shape_sample_count":    shape["sample_count"],
+        #       "shape_axis":            shape["axis"],
+        #   } or {}),
+```
+
+Concretely, after each method computes `hdf5_filename`, add before its `return {...}`:
+
+```python
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
+```
+
+and spread `**_shape_rec` into the returned dict. (`save_imported_bw`/`save_imported_idf` use their own hdf5 path variable names — use whichever local holds the written `.h5` path in each method.)
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_save_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add sfm/waveform_store.py tests/test_save_shape.py
+git commit -m "feat(ingest): compute shape from the written .h5 in all save paths"
+```
+
+---
+
+### Task 6: Backfill script for existing events
+
+**Files:**
+- Create: `scripts/backfill_event_shape.py` (mirror `scripts/backfill_event_zc_freq.py`, but read the `.h5` for samples instead of the sidecar).
+- Test: `tests/test_backfill_event_shape.py`
+
+**Interfaces:**
+- Produces: `backfill_shape(db, store, *, dry_run=False) -> dict` with counts `{"updated","skipped_no_h5","skipped_no_samples"}`; `main(argv)` CLI mirroring the zc-freq script's args (`--db-path`, `--store-root`, `--dry-run`).
+
+- [ ] **Step 1: Write the failing test**
+
+```python
+# tests/test_backfill_event_shape.py
+import numpy as np, h5py
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from scripts.backfill_event_shape import backfill_shape
+from tests.helpers_events import make_event   # or inline as in test_zc_freq_columns
+
+def _h5(path, long):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k in ("Tran", "Vert"): g.create_dataset(k, data=np.zeros(1024, "float32"))
+        g.create_dataset("Long", data=np.asarray(long, "float32"))
+
+def test_backfill_updates_shape_and_is_idempotent(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    store = WaveformStore(tmp_path / "waveforms")
+    ev = make_event(serial="BE1", key="0111abcd")
+    db.insert_events([ev], serial="BE1",
+                     waveform_records={ev._waveform_key.hex():
+                                       {"filename": "F.CE0W", "filesize": 10}})
+    # place the .h5 where store.paths_for expects it
+    long = np.zeros(1024); long[100] = 0.48
+    _h5(store.hdf5_path_for("BE1", "F.CE0W"), long)
+
+    c1 = backfill_shape(db, store)
+    assert c1["updated"] == 1
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_axis"] == "Long" and row["shape_near_peak_count"] <= 3
+    c2 = backfill_shape(db, store)                  # idempotent: re-run overwrites same values
+    assert db.query_events(serial="BE1")[0]["shape_crest_factor"] == row["shape_crest_factor"]
+```
+
+> Match `make_event` / column names to whatever `tests/test_zc_freq_columns.py` uses. `store.hdf5_path_for(serial, filename)` is the existing helper that returns the `.h5` path.
+
+- [ ] **Step 2: Run to verify it fails**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_backfill_event_shape.py -q`
+Expected: FAIL — `ModuleNotFoundError: scripts.backfill_event_shape`.
+
+- [ ] **Step 3: Implementation** (mirror the zc-freq script structure)
+
+```python
+#!/usr/bin/env python3
+"""Backfill events.shape_* from each event's .h5 waveform samples. Idempotent."""
+from __future__ import annotations
+import argparse, logging, sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from sfm.shape_metrics import shape_from_h5
+
+log = logging.getLogger("backfill_event_shape")
+
+def backfill_shape(db: SeismoDb, store: WaveformStore, *, dry_run: bool = False) -> dict:
+    counts = {"updated": 0, "skipped_no_h5": 0, "skipped_no_samples": 0}
+    for row in db.query_events(limit=1_000_000):
+        serial, filename = row.get("serial"), row.get("blastware_filename")
+        if not serial or not filename:
+            counts["skipped_no_h5"] += 1; continue
+        h5_path = store.hdf5_path_for(serial, filename)
+        if not h5_path.exists():
+            counts["skipped_no_h5"] += 1; continue
+        shape = shape_from_h5(h5_path)
+        if shape is None:
+            counts["skipped_no_samples"] += 1; continue
+        if not dry_run:
+            with db._connect() as conn:
+                conn.execute(
+                    "UPDATE events SET shape_crest_factor=?, shape_near_peak_count=?, "
+                    "shape_sample_count=?, shape_axis=? WHERE id=?",
+                    (shape["crest_factor"], shape["near_peak_count"],
+                     shape["sample_count"], shape["axis"], row["id"]))
+        counts["updated"] += 1
+    log.info("backfill_shape: %s", counts)
+    return counts
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db-path", required=True)
+    ap.add_argument("--store-root", required=True)
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    logging.basicConfig(level=logging.INFO)
+    counts = backfill_shape(SeismoDb(a.db_path), WaveformStore(a.store_root), dry_run=a.dry_run)
+    print(counts)
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())
+```
+
+> `store.hdf5_path_for` and `db._connect` are existing internals used the same way by other scripts. If `hdf5_path_for` isn't public, use `store.paths_for(...)` sibling `.h5` path exactly as `save()` derives `hdf5_path`.
+
+- [ ] **Step 4: Run to verify it passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_backfill_event_shape.py -q`
+Expected: PASS.
+
+- [ ] **Step 5: Run the full suite + commit**
+
+```bash
+/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest -q
+git add scripts/backfill_event_shape.py tests/test_backfill_event_shape.py
+git commit -m "feat(scripts): backfill events.shape_* from .h5 samples"
+```
+
+---
+
+### Task 7: Confirm /db/events carries shape + version bump
+
+**Files:**
+- Test: `tests/test_db_events_exposes_shape.py`
+- Modify: `pyproject.toml` version; `sfm/server.py` version string; `CHANGELOG.md`.
+
+**Interfaces:**
+- Consumes: the running `/db/events` route (already returns `SELECT *`).
+
+- [ ] **Step 1: Write the failing test** (guards that the feed dict includes the new keys)
+
+```python
+# tests/test_db_events_exposes_shape.py
+from sfm.database import SeismoDb
+def test_query_events_row_includes_shape_keys(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    # query_events returns dict(row); a fresh insert has the keys (values may be None)
+    from tests.helpers_events import make_event
+    db.insert_events([make_event(serial="BE1", key="0111abcd")], serial="BE1")
+    row = db.query_events(serial="BE1")[0]
+    for k in ("shape_crest_factor","shape_near_peak_count","shape_sample_count","shape_axis"):
+        assert k in row
+```
+
+- [ ] **Step 2: Run to verify it fails / passes**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest tests/test_db_events_exposes_shape.py -q`
+Expected: PASS immediately if Task 3 landed (columns present in `SELECT *`). If it fails, the columns weren't added — fix Task 3. (This task is the guard, not new behavior.)
+
+- [ ] **Step 3: Version bump**
+
+Bump `pyproject.toml` `version` 0.23.0 → 0.24.0; set `sfm/server.py` `version="0.24.0"`; add a `## v0.24.0` CHANGELOG entry ("waveform-shape metrics on events: crest factor + points-near-peak, ingest + backfill").
+
+- [ ] **Step 4: Full suite**
+
+Run: `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest -q`
+Expected: PASS (no regressions).
+
+- [ ] **Step 5: Commit**
+
+```bash
+git add tests/test_db_events_exposes_shape.py pyproject.toml sfm/server.py CHANGELOG.md
+git commit -m "chore(release): v0.24.0 — waveform-shape metrics on events"
+```
+
+---
+
+## Self-Review
+
+**Spec coverage:** SFM shape columns (Tasks 3–4) ✓; DSP crest + near-peak (Task 1) ✓; ingest population (Task 5) ✓; backfill (Task 6) ✓; `/db/events` exposure (Task 7) ✓; NULL for histogram/no-sample events (Tasks 1/5/6 return None → NULL) ✓; 94%-coverage / series-4 fallback handled by NULL-then-Terra-View-fallback (Phase B) ✓. Terra-View scoring, 3-state review, twin flagging, export Notes column → **Phase B plan** (separate, depends on this feed). Calibration → Phase C.
+
+**Placeholder scan:** No TBD/TODO; every code step has real code. The two "reuse existing fixture" notes point at concrete existing tests (`test_zc_freq_columns.py`, `test_save_imported_bw*.py`) rather than leaving blanks.
+
+**Type consistency:** `shape_from_h5`/`shape_from_samples`/`channel_shape` return the same dict keys (`crest_factor`, `near_peak_count`, `sample_count`, `axis`) throughout; the DB columns (`shape_crest_factor`, `shape_near_peak_count`, `shape_sample_count`, `shape_axis`) and rec keys match across Tasks 4–6.
+
+## Deferred to Phase B (terra-view, separate plan)
+
+Scoring service combining shape + cheap signals; suspicion column + reason chips; `reviewed_real` mirror + 3-state review; twin-aware flag propagation; Notes-column export + Maximums "(excludes N flagged)". Written once this feed is live so column names/values are real.

docs/superpowers/plans/2026-08-25-b2a-reviewed-real-twin-propagation.md

@@ -0,0 +1,264 @@
+# Phase B2-A (seismo-relay) — reviewed_real + 3-state mirror + twin propagation
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Give SFM a persisted, queryable `reviewed_real` flag alongside `false_trigger` (mirrored from the sidecar review block, mutually exclusive), and propagate a review to an event's histogram/waveform twin so flagging one flags both.
+
+**Architecture:** Mirror the existing `false_trigger` mechanism. New `events.reviewed_real` column (auto-migrate). `update_event_review` mirrors BOTH flags from the sidecar review block and enforces mutual exclusivity on the columns. A `find_twins` matcher (same serial + identical peak_vector_sum + timestamp within a window) drives `propagate_review_to_twins`, which the sidecar-PATCH endpoint calls after mirroring the primary. Terra-View reads `reviewed_real` from `/db/events` (SELECT *).
+
+**Tech Stack:** Python 3.10, raw sqlite3, FastAPI, pytest. Runner: `/home/serversdown/seismo-relay/.venv/bin/python3`.
+
+## Global Constraints
+
+- 3 review states are **mutually exclusive**: an event is `false_trigger=1` XOR `reviewed_real=1` XOR neither. Setting one true forces the other's column to 0.
+- The sidecar JSON stays the source of truth for full review state; the `false_trigger`/`reviewed_real` columns are derived indexes (like today). B2-A propagates twins at the **column** level (what the feed/peak/export read); twin sidecars are not rewritten (known limitation — noted).
+- Twin match = **same serial AND identical `peak_vector_sum` (exact equality) AND `timestamp` within ± window (default 300 s), excluding the event itself.** Identical PVS is the safety anchor.
+- Known pre-existing test failures (~16, missing gitignored fixtures under `tests/fixtures/`) are unrelated — confirm zero NEW failures, don't try to fix them.
+- Run tests with `/home/serversdown/seismo-relay/.venv/bin/python3 -m pytest`.
+
+---
+
+### Task 1: `reviewed_real` column on events
+
+**Files:** Modify `sfm/database.py` (`_SCHEMA` CREATE TABLE `events` + the Migration-1 rebuild `CREATE TABLE` + the `_migrate` ADD COLUMN loop). Test: `tests/test_reviewed_real_column.py`.
+
+**Interfaces:** Produces `events.reviewed_real INTEGER NOT NULL DEFAULT 0`.
+
+- [ ] **Step 1: Failing test**
+```python
+# tests/test_reviewed_real_column.py
+import sqlite3
+from sfm.database import SeismoDb
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+def test_fresh_db_has_reviewed_real(tmp_path):
+    assert "reviewed_real" in _cols(SeismoDb(tmp_path/"s.db"))
+def test_existing_db_migrates_reviewed_real(tmp_path):
+    p = tmp_path/"s.db"; db = SeismoDb(p)
+    with sqlite3.connect(p) as c:
+        c.execute("ALTER TABLE events DROP COLUMN reviewed_real")
+    assert "reviewed_real" not in _cols(db)      # dropped (read via existing handle/connection)
+    SeismoDb(p)                                   # re-open migrates
+    assert "reviewed_real" in _cols(SeismoDb(p))
+```
+> If sqlite < 3.35 lacks DROP COLUMN, fall back to building a table without the column and asserting re-open adds it (same as the shape-columns test did).
+
+- [ ] **Step 2: Run → FAIL** (`pytest tests/test_reviewed_real_column.py -q`).
+- [ ] **Step 3: Implement**
+  - In `_SCHEMA` `events` CREATE TABLE, after the `false_trigger ... DEFAULT 0,` line add: `    reviewed_real           INTEGER NOT NULL DEFAULT 0,  -- 0=no, 1=operator-confirmed real (mutually exclusive with false_trigger)`
+  - In the `_migrate` ADD COLUMN loop tuple add: `("reviewed_real", "INTEGER NOT NULL DEFAULT 0"),`
+  - **Do NOT** add it to the Migration-1 rebuild `CREATE TABLE events (...)` block — that block uses a positional `INSERT ... SELECT * FROM events_old` and, by convention, contains only the columns that existed when Migration 1 was written; every later column is added by the ADD COLUMN loop only. Adding it there crashes `_migrate` on genuinely legacy (pre-Migration-1) DBs.
+- [ ] **Step 4: Run → PASS.**
+- [ ] **Step 5: Commit** `feat(db): reviewed_real column on events (+ auto-migrate)`
+
+---
+
+### Task 2: `update_event_review` mirrors both flags + mutual exclusivity
+
+**Files:** Modify `sfm/database.py` `update_event_review`. Test: `tests/test_update_event_review_reviewed_real.py`.
+
+**Interfaces:** Consumes a `review` dict that may carry `false_trigger` and/or `reviewed_real` (bools). Produces mutually-exclusive column state.
+
+- [ ] **Step 1: Failing test**
+```python
+# tests/test_update_event_review_reviewed_real.py
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+def _ins(db, eid_key="01110000", serial="BE1"):
+    ev = Event(index=0); ev._waveform_key = bytes.fromhex(eid_key)
+    db.insert_events([ev], serial=serial)
+    return db.query_events(serial=serial)[0]["id"]
+def test_confirm_real_sets_and_clears_ft(tmp_path):
+    db = SeismoDb(tmp_path/"s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"false_trigger": True})
+    assert db.get_event(eid)["false_trigger"] == 1
+    db.update_event_review(eid, {"reviewed_real": True})   # confirming real clears FT
+    row = db.get_event(eid)
+    assert row["reviewed_real"] == 1 and row["false_trigger"] == 0
+def test_flag_ft_clears_reviewed_real(tmp_path):
+    db = SeismoDb(tmp_path/"s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"reviewed_real": True})
+    db.update_event_review(eid, {"false_trigger": True})
+    row = db.get_event(eid)
+    assert row["false_trigger"] == 1 and row["reviewed_real"] == 0
+```
+> Build the Event inline like `tests/test_zc_freq_columns.py` if the import differs.
+
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement** — replace the body of `update_event_review` so it handles both keys:
+```python
+    if not isinstance(review, dict):
+        return False
+    has_ft = "false_trigger" in review
+    has_real = "reviewed_real" in review
+    if not has_ft and not has_real:
+        with self._connect() as conn:
+            row = conn.execute("SELECT 1 FROM events WHERE id=?", (event_id,)).fetchone()
+        return row is not None
+    sets = {}
+    if has_ft:
+        sets["false_trigger"] = 1 if review.get("false_trigger") else 0
+    if has_real:
+        sets["reviewed_real"] = 1 if review.get("reviewed_real") else 0
+    # mutual exclusivity: a true in one forces the other column to 0
+    if sets.get("false_trigger") == 1:
+        sets["reviewed_real"] = 0
+    if sets.get("reviewed_real") == 1:
+        sets["false_trigger"] = 0
+    assign = ", ".join(f"{k}=?" for k in sets)
+    params = list(sets.values()) + [event_id]
+    with self._connect() as conn:
+        cur = conn.execute(f"UPDATE events SET {assign} WHERE id=?", params)
+    return cur.rowcount > 0
+```
+- [ ] **Step 4: Run → PASS** (+ run `tests/test_*false_trigger*`/existing review tests to confirm no regression).
+- [ ] **Step 5: Commit** `feat(db): update_event_review mirrors reviewed_real + enforces 3-state exclusivity`
+
+---
+
+### Task 3: `find_twins` matcher
+
+**Files:** Modify `sfm/database.py` (add `find_twins`). Test: `tests/test_find_twins.py`.
+
+**Interfaces:** Produces `find_twins(event_id, *, window_seconds=300) -> list[dict]` — same serial, identical peak_vector_sum, timestamp within ±window, excluding self.
+
+- [ ] **Step 1: Failing test**
+```python
+# tests/test_find_twins.py
+import datetime
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp
+def _ins(db, key, serial, pvs, ts):
+    ev = Event(index=0); ev._waveform_key = bytes.fromhex(key)
+    ev.timestamp = ts
+    # peak_vector_sum comes from peak_values; simplest: insert then UPDATE pvs directly
+    db.insert_events([ev], serial=serial)
+    row = [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]
+    import sqlite3
+    with sqlite3.connect(db.db_path) as c:
+        c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"]))
+    return row["id"]
+def test_find_twins_matches_same_serial_pvs_near_time(tmp_path):
+    db = SeismoDb(tmp_path/"s.db")
+    base = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=5)
+    twin = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=45)
+    far  = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=21, minute=0,  second=0)
+    a = _ins(db, "01110001", "BE1", 0.4763, base)
+    b = _ins(db, "01110002", "BE1", 0.4763, twin)   # twin: same serial+pvs, 40s apart
+    c = _ins(db, "01110003", "BE1", 0.4763, far)    # same pvs but >window away
+    d = _ins(db, "01110004", "BE1", 0.9999, twin)   # near time but different pvs
+    ids = {r["id"] for r in db.find_twins(a, window_seconds=300)}
+    assert ids == {b}
+```
+> Adjust the Event/Timestamp construction to match how `tests/test_waveform_store.py::_make_synthetic_event` builds them if fields differ.
+
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement**
+```python
+    def find_twins(self, event_id: str, *, window_seconds: int = 300) -> list[dict]:
+        row = self.get_event(event_id)
+        if not row:
+            return []
+        serial = row.get("serial"); pvs = row.get("peak_vector_sum"); ts = row.get("timestamp")
+        if serial is None or pvs is None or not ts:
+            return []
+        try:
+            t = datetime.datetime.fromisoformat(ts.replace(" ", "T"))
+        except ValueError:
+            return []
+        lo = (t - datetime.timedelta(seconds=window_seconds)).isoformat()
+        hi = (t + datetime.timedelta(seconds=window_seconds)).isoformat()
+        with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=? "
+                "AND timestamp BETWEEN ? AND ?",
+                (serial, event_id, pvs, lo, hi),
+            ).fetchall()
+        return [dict(r) for r in rows]
+```
+- [ ] **Step 4: Run → PASS.**
+- [ ] **Step 5: Commit** `feat(db): find_twins (serial + identical PVS + time window)`
+
+---
+
+### Task 4: propagate a review to twins + wire into the sidecar-PATCH endpoint
+
+**Files:** Modify `sfm/database.py` (add `propagate_review_to_twins`); `sfm/server.py` (`db_event_sidecar_patch`). Test: `tests/test_twin_propagation.py`.
+
+**Interfaces:** `propagate_review_to_twins(event_id, *, window_seconds=300) -> list[str]` copies the event's `false_trigger`/`reviewed_real` columns onto each twin; returns twin ids.
+
+- [ ] **Step 1: Failing test** (DB-level)
+```python
+# tests/test_twin_propagation.py — reuse the _ins helper pattern from test_find_twins
+def test_propagate_copies_flags_to_twins(tmp_path):
+    db = SeismoDb(tmp_path/"s.db")
+    # (build primary + twin via the _ins helper as in test_find_twins)
+    # flag the primary FT, then propagate:
+    db.update_event_review(primary_id, {"false_trigger": True})
+    moved = db.propagate_review_to_twins(primary_id)
+    assert twin_id in moved
+    assert db.get_event(twin_id)["false_trigger"] == 1
+```
+- [ ] **Step 2: Run → FAIL.**
+- [ ] **Step 3: Implement**
+  - `sfm/database.py`:
+```python
+    def propagate_review_to_twins(self, event_id: str, *, window_seconds: int = 300) -> list[str]:
+        row = self.get_event(event_id)
+        if not row:
+            return []
+        ft = 1 if row.get("false_trigger") else 0
+        real = 1 if row.get("reviewed_real") else 0
+        twins = self.find_twins(event_id, window_seconds=window_seconds)
+        moved = []
+        with self._connect() as conn:
+            for tw in twins:
+                conn.execute("UPDATE events SET false_trigger=?, reviewed_real=? WHERE id=?",
+                             (ft, real, tw["id"]))
+                moved.append(tw["id"])
+        return moved
+```
+  - `sfm/server.py` `db_event_sidecar_patch`: after the existing `_get_db().update_event_review(event_id, new_sidecar.get("review", {}))`, add:
+```python
+        # Propagate the review to the event's histogram/waveform twin(s) so
+        # flagging one flags both (column-level; twins share serial+PVS+near time).
+        try:
+            _get_db().propagate_review_to_twins(event_id)
+        except Exception as exc:
+            log.warning("twin review-propagation failed for %s: %s", event_id, exc)
+```
+  (Guard the `if body.review is not None:` block so propagation only runs when review changed.)
+- [ ] **Step 4: Run → PASS.**
+- [ ] **Step 5: Commit** `feat(review): propagate false_trigger/reviewed_real to twins on sidecar PATCH`
+
+---
+
+### Task 5: expose in feed guard + version bump
+
+**Files:** Test `tests/test_reviewed_real_in_feed.py`; `pyproject.toml`, `sfm/server.py` version, `CHANGELOG.md`.
+
+- [ ] **Step 1: Guard test** — a `query_events` row dict includes `reviewed_real` (SELECT * returns it).
+```python
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+def test_query_events_includes_reviewed_real(tmp_path):
+    db = SeismoDb(tmp_path/"s.db")
+    ev = Event(index=0); ev._waveform_key = bytes.fromhex("01110000")
+    db.insert_events([ev], serial="BE1")
+    assert "reviewed_real" in db.query_events(serial="BE1")[0]
+```
+- [ ] **Step 2: Run → PASS** (columns already present from Task 1).
+- [ ] **Step 3: Bump** `pyproject.toml` 0.24.0 → 0.25.0; `sfm/server.py` version="0.25.0"; add `## v0.25.0` CHANGELOG entry ("reviewed_real 3-state review flag + histogram/waveform twin review-propagation").
+- [ ] **Step 4: Full suite** — confirm zero NEW failures beyond the ~16 pre-existing.
+- [ ] **Step 5: Commit** `chore(release): v0.25.0 — reviewed_real + twin review-propagation`
+
+---
+
+## Self-Review
+
+**Spec coverage:** reviewed_real column (T1) ✓; mirror + mutual exclusivity (T2) ✓; twin match serial+identical-PVS+window (T3) ✓; twin propagation wired into the review path (T4) ✓; feed exposure + version (T5) ✓. Twin **sidecar** rewrite intentionally deferred (column-level only — documented in Global Constraints); this is the SFM half — the 3-state UI + confirm-real PATCH is the separate **B2-B (terra-view)** plan.
+
+**Placeholder scan:** none — every step has real code (the two "adjust Event construction to match test_waveform_store" notes point at a concrete existing helper).
+
+**Type consistency:** `false_trigger`/`reviewed_real` INTEGER columns used identically across T1/T2/T4; `find_twins`→`propagate_review_to_twins` both key on the same match; server calls the DB methods by the exact names T2–T4 define.

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "seismo-relay"
-version = "0.23.0"
+version = "0.25.0"
 description = "Python client and REST server for MiniMate Plus seismographs"
 requires-python = ">=3.10"
 dependencies = [

scripts/backfill_event_shape.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+"""Backfill events.shape_* from each event's .h5 waveform samples. Idempotent."""
+from __future__ import annotations
+import argparse, logging, sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from sfm.shape_metrics import shape_from_h5
+
+log = logging.getLogger("backfill_event_shape")
+
+def backfill_shape(db: SeismoDb, store: WaveformStore, *, dry_run: bool = False) -> dict:
+    counts = {"updated": 0, "skipped_no_h5": 0, "skipped_no_samples": 0}
+    for row in db.query_events(limit=1_000_000):
+        serial, filename = row.get("serial"), row.get("blastware_filename")
+        if not serial or not filename:
+            counts["skipped_no_h5"] += 1; continue
+        h5_path = store.hdf5_path_for(serial, filename)
+        if not h5_path.exists():
+            counts["skipped_no_h5"] += 1; continue
+        shape = shape_from_h5(h5_path)
+        if shape is None:
+            counts["skipped_no_samples"] += 1; continue
+        if not dry_run:
+            with db._connect() as conn:
+                conn.execute(
+                    "UPDATE events SET shape_crest_factor=?, shape_near_peak_count=?, "
+                    "shape_sample_count=?, shape_axis=? WHERE id=?",
+                    (shape["crest_factor"], shape["near_peak_count"],
+                     shape["sample_count"], shape["axis"], row["id"]))
+        counts["updated"] += 1
+    log.info("backfill_shape: %s", counts)
+    return counts
+
+def main(argv=None) -> int:
+    ap = argparse.ArgumentParser()
+    ap.add_argument("--db-path", required=True)
+    ap.add_argument("--store-root", required=True)
+    ap.add_argument("--dry-run", action="store_true")
+    a = ap.parse_args(argv)
+    logging.basicConfig(level=logging.INFO)
+    counts = backfill_shape(SeismoDb(a.db_path), WaveformStore(a.store_root), dry_run=a.dry_run)
+    print(counts)
+    return 0
+
+if __name__ == "__main__":
+    raise SystemExit(main())

sfm/database.py

@@ -81,6 +81,7 @@ CREATE TABLE IF NOT EXISTS events (
     sample_rate             INTEGER,
     record_type             TEXT,                   -- "single_shot" | "continuous"
     false_trigger           INTEGER NOT NULL DEFAULT 0,  -- 0=no, 1=yes (manual flag)
+    reviewed_real           INTEGER NOT NULL DEFAULT 0,  -- 0=no, 1=operator-confirmed real (mutually exclusive with false_trigger)
     blastware_filename      TEXT,                   -- event file within waveform store; extension is per-event (AB0T encodes timestamp)
     blastware_filesize      INTEGER,                -- bytes; NULL if no event file saved
     a5_pickle_filename      TEXT,                   -- "<filename>.a5.pkl" sidecar
@@ -94,6 +95,10 @@ CREATE TABLE IF NOT EXISTS events (
     vert_zc_above_range     INTEGER,
     long_zc_above_range     INTEGER,
     mic_zc_above_range      INTEGER,
+    shape_crest_factor      REAL,                   -- peak / rms of the triggering channel
+    shape_near_peak_count   INTEGER,                -- samples >= 0.5 * peak (FT: few; real: many)
+    shape_sample_count      INTEGER,                -- total samples (to normalize near_peak_count)
+    shape_axis              TEXT,                   -- geophone channel measured ("Tran"/"Vert"/"Long")
     created_at              TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
     UNIQUE(serial, timestamp)
 );
@@ -216,6 +221,11 @@ class SeismoDb:
             ("vert_zc_above_range", "INTEGER"),
             ("long_zc_above_range", "INTEGER"),
             ("mic_zc_above_range",  "INTEGER"),
+            ("shape_crest_factor",    "REAL"),
+            ("shape_near_peak_count", "INTEGER"),
+            ("shape_sample_count",    "INTEGER"),
+            ("shape_axis",            "TEXT"),
+            ("reviewed_real",         "INTEGER NOT NULL DEFAULT 0"),
         ):
             if col not in existing_cols:
                 log.info("_migrate: events ADD COLUMN %s %s", col, ddl)
@@ -418,9 +428,11 @@ class SeismoDb:
                              device_family,
                              tran_zc_freq, vert_zc_freq, long_zc_freq, mic_zc_freq,
                              tran_zc_above_range, vert_zc_above_range,
-                             long_zc_above_range, mic_zc_above_range)
+                             long_zc_above_range, mic_zc_above_range,
+                             shape_crest_factor, shape_near_peak_count,
+                             shape_sample_count, shape_axis)
                         VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?,
-                                ?, ?, ?, ?, ?, ?, ?, ?)
+                                ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                         """,
                         (
                             self._new_id(), serial, key, session_id, ts,
@@ -448,6 +460,10 @@ class SeismoDb:
                             (1 if (pv and pv.vert_zc_above_range) else 0),
                             (1 if (pv and pv.long_zc_above_range) else 0),
                             (1 if (pv and pv.mic_zc_above_range)  else 0),
+                            rec.get("shape_crest_factor"),
+                            rec.get("shape_near_peak_count"),
+                            rec.get("shape_sample_count"),
+                            rec.get("shape_axis"),
                         ),
                     )
                     inserted += 1
@@ -497,7 +513,11 @@ class SeismoDb:
                                tran_zc_above_range = ?,
                                vert_zc_above_range = ?,
                                long_zc_above_range = ?,
-                               mic_zc_above_range  = ?
+                               mic_zc_above_range  = ?,
+                               shape_crest_factor    = COALESCE(?, shape_crest_factor),
+                               shape_near_peak_count = COALESCE(?, shape_near_peak_count),
+                               shape_sample_count    = COALESCE(?, shape_sample_count),
+                               shape_axis            = COALESCE(?, shape_axis)
                          WHERE serial = ? AND timestamp = ?
                         """,
                         (
@@ -525,6 +545,10 @@ class SeismoDb:
                             (1 if (pv and pv.vert_zc_above_range) else 0),
                             (1 if (pv and pv.long_zc_above_range) else 0),
                             (1 if (pv and pv.mic_zc_above_range)  else 0),
+                            rec.get("shape_crest_factor")    if rec else None,
+                            rec.get("shape_near_peak_count") if rec else None,
+                            rec.get("shape_sample_count")    if rec else None,
+                            rec.get("shape_axis")            if rec else None,
                             serial,
                             ts,
                         ),
@@ -579,12 +603,83 @@ class SeismoDb:
             ).fetchall()
         return [dict(r) for r in rows]
 
-    def set_false_trigger(self, event_id: str, value: bool) -> bool:
+    def find_twins(self, event_id: str, *, window_seconds: int = 300) -> list[dict]:
-        """Set or clear the false_trigger flag on an event. Returns True if found."""
+        """
+        Find this event's histogram/waveform twin(s): rows sharing the same
+        serial and an identical peak_vector_sum, whose timestamp falls
+        within ``window_seconds`` of this event's timestamp.  Excludes the
+        event itself.  Returns [] if the event or any required field
+        (serial / peak_vector_sum / timestamp) is missing.
+
+        Caveat: identical-PVS matching is a proxy for "same physical event
+        recorded twice," not a guarantee. In the rare case where the device
+        clamps/saturates PVS (clamped to sqrt(3) * geo_range), two distinct
+        saturated events on the same serial within the window can share the
+        same clamped PVS value and be matched as twins even though they are
+        different events. This is harmless in practice — false_trigger/
+        reviewed_real are derived/index columns re-derivable from the
+        sidecar source of truth — but worth knowing if twin counts look
+        surprising on a saturated/clamped run.
+        """
+        row = self.get_event(event_id)
+        if not row:
+            return []
+        serial = row.get("serial"); pvs = row.get("peak_vector_sum"); ts = row.get("timestamp")
+        if serial is None or pvs is None or not ts:
+            return []
+        try:
+            t = datetime.datetime.fromisoformat(ts.replace(" ", "T"))
+        except ValueError:
+            return []
+        lo = (t - datetime.timedelta(seconds=window_seconds)).isoformat()
+        hi = (t + datetime.timedelta(seconds=window_seconds)).isoformat()
         with self._connect() as conn:
+            rows = conn.execute(
+                "SELECT * FROM events WHERE serial=? AND id!=? AND peak_vector_sum=? "
+                "AND timestamp BETWEEN ? AND ?",
+                (serial, event_id, pvs, lo, hi),
+            ).fetchall()
+        return [dict(r) for r in rows]
+
+    def propagate_review_to_twins(self, event_id: str, *, window_seconds: int = 300) -> list[str]:
+        """
+        Copy this event's `false_trigger`/`reviewed_real` columns onto each
+        of its histogram/waveform twins (see `find_twins`), so flagging one
+        twin flags both.  Returns the list of twin ids updated.
+        """
+        row = self.get_event(event_id)
+        if not row:
+            return []
+        ft = 1 if row.get("false_trigger") else 0
+        real = 1 if row.get("reviewed_real") else 0
+        twins = self.find_twins(event_id, window_seconds=window_seconds)
+        moved = []
+        with self._connect() as conn:
+            for tw in twins:
+                conn.execute("UPDATE events SET false_trigger=?, reviewed_real=? WHERE id=?",
+                             (ft, real, tw["id"]))
+                moved.append(tw["id"])
+        return moved
+
+    def set_false_trigger(self, event_id: str, value: bool) -> bool:
+        """
+        Set or clear the false_trigger flag on an event. Returns True if found.
+
+        Mirrors the 3-state exclusivity enforced by `update_event_review`:
+        setting false_trigger true also clears `reviewed_real` (a false
+        trigger can't also be a confirmed-real event). Clearing false_trigger
+        leaves `reviewed_real` untouched.
+        """
+        with self._connect() as conn:
+            if value:
                 cur = conn.execute(
-                "UPDATE events SET false_trigger=? WHERE id=?",
+                    "UPDATE events SET false_trigger=1, reviewed_real=0 WHERE id=?",
-                (1 if value else 0, event_id),
+                    (event_id,),
+                )
+            else:
+                cur = conn.execute(
+                    "UPDATE events SET false_trigger=0 WHERE id=?",
+                    (event_id,),
                 )
         return cur.rowcount > 0
 
@@ -661,17 +756,23 @@ class SeismoDb:
         """
         Sync derived index columns from a sidecar's `review` block.
 
-        Currently the only derived index is `events.false_trigger` — kept
+        The derived indexes are `events.false_trigger` and
-        in sync so `/db/events?false_trigger=true` queries don't have to
+        `events.reviewed_real` — kept in sync so `/db/events` queries don't
-        scan every sidecar JSON on disk.  The sidecar JSON itself remains
+        have to scan every sidecar JSON on disk.  The sidecar JSON itself
-        the source of truth for the full review state.
+        remains the source of truth for the full review state.
+
+        The two columns are mutually exclusive: setting either one true
+        forces the other's column to 0.
 
         Returns True when the row exists, False otherwise.  No-op fields
-        (review without `false_trigger`) leave the column untouched.
+        (review without `false_trigger` or `reviewed_real`) leave both
+        columns untouched.
         """
         if not isinstance(review, dict):
             return False
-        if "false_trigger" not in review:
+        has_ft = "false_trigger" in review
+        has_real = "reviewed_real" in review
+        if not has_ft and not has_real:
             # Nothing derived to update; just confirm the row exists.
             with self._connect() as conn:
                 row = conn.execute(
@@ -679,12 +780,20 @@ class SeismoDb:
                 ).fetchone()
             return row is not None
 
-        flag = 1 if review.get("false_trigger") else 0
+        sets = {}
+        if has_ft:
+            sets["false_trigger"] = 1 if review.get("false_trigger") else 0
+        if has_real:
+            sets["reviewed_real"] = 1 if review.get("reviewed_real") else 0
+        # mutual exclusivity: a true in one forces the other column to 0
+        if sets.get("false_trigger") == 1:
+            sets["reviewed_real"] = 0
+        if sets.get("reviewed_real") == 1:
+            sets["false_trigger"] = 0
+        assign = ", ".join(f"{k}=?" for k in sets)
+        params = list(sets.values()) + [event_id]
         with self._connect() as conn:
-            cur = conn.execute(
+            cur = conn.execute(f"UPDATE events SET {assign} WHERE id=?", params)
-                "UPDATE events SET false_trigger=? WHERE id=?",
-                (flag, event_id),
-            )
         return cur.rowcount > 0
 
     # ── Monitor log ───────────────────────────────────────────────────────────

sfm/server.py

@@ -90,7 +90,7 @@ app = FastAPI(
         "Implements the minimateplus RS-232 protocol library.\n"
         "Proxied by terra-view at /api/sfm/*."
     ),
-    version="0.23.0",
+    version="0.25.0",
 )
 
 # Allow requests from the waveform viewer opened as a local file (file://)
@@ -1957,7 +1957,10 @@ def db_set_false_trigger(
     value:    bool = Query(..., description="True to flag as false trigger, False to clear"),
 ) -> dict:
     """
-    Set or clear the false_trigger flag on a single event.
+    Set or clear the false_trigger flag on a single event.  Enforces the
+    same 3-state exclusivity as the sidecar review PATCH (flagging false_trigger
+    clears reviewed_real) and propagates the resulting flags to the event's
+    histogram/waveform twin(s), same as the sidecar path.
 
     Used by the terra-view event review UI.
     Returns 404 if the event_id is not found.
@@ -1965,6 +1968,12 @@ def db_set_false_trigger(
     found = _get_db().set_false_trigger(event_id, value)
     if not found:
         raise HTTPException(status_code=404, detail=f"Event {event_id} not found")
+
+    try:
+        _get_db().propagate_review_to_twins(event_id)
+    except Exception as exc:
+        log.warning("twin review-propagation failed for %s: %s", event_id, exc)
+
     return {"status": "ok", "event_id": event_id, "false_trigger": value}
 
 
@@ -2481,6 +2490,13 @@ def db_event_sidecar_patch(event_id: str, body: SidecarPatchBody) -> dict:
     if body.review is not None:
         _get_db().update_event_review(event_id, new_sidecar.get("review", {}))
 
+        # Propagate the review to the event's histogram/waveform twin(s) so
+        # flagging one flags both (column-level; twins share serial+PVS+near time).
+        try:
+            _get_db().propagate_review_to_twins(event_id)
+        except Exception as exc:
+            log.warning("twin review-propagation failed for %s: %s", event_id, exc)
+
     return new_sidecar
 
 

sfm/shape_metrics.py

@@ -0,0 +1,58 @@
+"""Waveform-shape metrics for false-trigger detection.
+
+A false trigger is an isolated impulse (quiet → spike → quiet); a real event
+rings for many cycles. Two numbers separate them: crest factor (how far the
+peak stands above the typical sample) and how many samples sit near the peak.
+"""
+from __future__ import annotations
+import numpy as np
+
+_GEO_CHANNELS = ("Tran", "Vert", "Long")
+NEAR_PEAK_FRACTION = 0.5   # a sample "near the peak" is >= this * peak amplitude
+
+
+def channel_shape(x) -> dict | None:
+    x = np.asarray(x, dtype=float)
+    if x.size < 2:
+        return None
+    peak = float(np.max(np.abs(x)))
+    if peak <= 0:
+        return None
+    rms = float(np.sqrt(np.mean(x ** 2)))
+    if rms <= 0:
+        return None
+    near = int(np.sum(np.abs(x) >= NEAR_PEAK_FRACTION * peak))
+    return {"crest_factor": peak / rms, "near_peak_count": near,
+            "sample_count": int(x.size)}
+
+
+def shape_from_samples(chans: dict) -> dict | None:
+    best_axis, best_peak, best_x = None, -1.0, None
+    for ax in _GEO_CHANNELS:
+        x = chans.get(ax)
+        if x is None:
+            continue
+        x = np.asarray(x, dtype=float)
+        if x.size < 2:
+            continue
+        p = float(np.max(np.abs(x)))
+        if p > best_peak:
+            best_axis, best_peak, best_x = ax, p, x
+    if best_axis is None:
+        return None
+    s = channel_shape(best_x)
+    if s is None:
+        return None
+    s["axis"] = best_axis
+    return s
+
+
+def shape_from_h5(path) -> dict | None:
+    import h5py
+    try:
+        with h5py.File(path, "r") as f:
+            chans = {ax: f[f"samples/{ax}"][:] for ax in _GEO_CHANNELS
+                     if f"samples/{ax}" in f}
+    except Exception:
+        return None
+    return shape_from_samples(chans)

sfm/waveform_store.py

@@ -41,6 +41,7 @@ from minimateplus.blastware_file import blastware_filename, write_blastware_file
 from minimateplus.framing import S3Frame
 from minimateplus.models import Event
 from sfm import event_hdf5
+from sfm.shape_metrics import shape_from_h5
 
 log = logging.getLogger("sfm.waveform_store")
 
@@ -262,6 +263,13 @@ class WaveformStore:
             serial, filename, filesize, len(a5_frames),
             hdf5_filename or "(skipped)", sidecar_path.name,
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return {
             "filename":            filename,
             "filesize":            filesize,
@@ -269,6 +277,7 @@ class WaveformStore:
             "a5_pickle_filename":  a5_path.name,
             "hdf5_filename":       hdf5_filename,
             "sidecar_filename":    sidecar_path.name,
+            **_shape_rec,
         }
 
     def save_imported_bw(
@@ -445,6 +454,13 @@ class WaveformStore:
             "h5=%s (no .a5.pkl — A5 source unavailable for BW-imported files)",
             serial, filename, filesize, hdf5_filename or "(skipped)",
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return ev, {
             "filename":           filename,
             "filesize":           filesize,
@@ -453,6 +469,7 @@ class WaveformStore:
             "hdf5_filename":      hdf5_filename,
             "sidecar_filename":   sidecar_path.name,
             "serial":             serial,
+            **_shape_rec,
         }
 
     def save_imported_idf(
@@ -727,6 +744,13 @@ class WaveformStore:
             hdf5_filename or "(skipped)",
             len(idf_intervals) if idf_intervals else 0,
         )
+        _shape = shape_from_h5(hdf5_path) if hdf5_filename else None
+        _shape_rec = {
+            "shape_crest_factor":    _shape["crest_factor"],
+            "shape_near_peak_count": _shape["near_peak_count"],
+            "shape_sample_count":    _shape["sample_count"],
+            "shape_axis":            _shape["axis"],
+        } if _shape else {}
         return ev, {
             "filename":           filename,
             "filesize":           filesize,
@@ -735,6 +759,7 @@ class WaveformStore:
             "hdf5_filename":      hdf5_filename,
             "sidecar_filename":   sidecar_path.name,
             "serial":             serial,
+            **_shape_rec,
         }
 
     def load_a5(self, serial: str, filename: str) -> Optional[list[S3Frame]]:

tests/test_backfill_event_shape.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+import numpy as np, h5py
+
+from sfm.database import SeismoDb
+from sfm.waveform_store import WaveformStore
+from scripts.backfill_event_shape import backfill_shape
+from minimateplus.models import Event
+
+
+def _h5(path, long):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k in ("Tran", "Vert"):
+            g.create_dataset(k, data=np.zeros(1024, "float32"))
+        g.create_dataset("Long", data=np.asarray(long, "float32"))
+
+
+def test_backfill_updates_shape_and_is_idempotent(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    store = WaveformStore(tmp_path / "waveforms")
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex("0111abcd")
+    db.insert_events([ev], serial="BE1",
+                     waveform_records={ev._waveform_key.hex():
+                                       {"filename": "F.CE0W", "filesize": 10}})
+    # place the .h5 where store.paths_for expects it
+    long = np.zeros(1024); long[100] = 0.48
+    _h5(store.hdf5_path_for("BE1", "F.CE0W"), long)
+
+    c1 = backfill_shape(db, store)
+    assert c1["updated"] == 1
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_axis"] == "Long" and row["shape_near_peak_count"] <= 3
+    c2 = backfill_shape(db, store)                  # idempotent: re-run overwrites same values
+    assert db.query_events(serial="BE1")[0]["shape_crest_factor"] == row["shape_crest_factor"]

tests/test_db_events_exposes_shape.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+import os, sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+
+SHAPE_COLS = (
+    "shape_crest_factor",
+    "shape_near_peak_count",
+    "shape_sample_count",
+    "shape_axis",
+)
+
+
+def test_query_events_row_includes_shape_keys(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex("0111abcd")
+
+    db.insert_events([ev], serial="BE1")
+    row = db.query_events(serial="BE1")[0]
+
+    for k in SHAPE_COLS:
+        assert k in row
+        assert row[k] is None

tests/test_find_twins.py

@@ -0,0 +1,33 @@
+import datetime
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp
+
+
+def _ins(db, key, serial, pvs, ts):
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex(key)
+    ev.timestamp = ts
+    # peak_vector_sum comes from peak_values; simplest: insert then UPDATE pvs directly
+    db.insert_events([ev], serial=serial)
+    row = [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]
+    import sqlite3
+    with sqlite3.connect(db.db_path) as c:
+        c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"]))
+    return row["id"]
+
+
+def test_find_twins_matches_same_serial_pvs_near_time(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    base = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=5)
+    twin = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=45)
+    far = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=21, minute=0, second=0)
+    # d needs a timestamp distinct from `twin` (UNIQUE(serial, timestamp) would
+    # otherwise collide with b and UPSERT onto its row instead of inserting a
+    # new one) while staying near `base` in time.
+    near = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=44)
+    a = _ins(db, "01110001", "BE1", 0.4763, base)
+    b = _ins(db, "01110002", "BE1", 0.4763, twin)   # twin: same serial+pvs, 40s apart
+    c = _ins(db, "01110003", "BE1", 0.4763, far)    # same pvs but >window away
+    d = _ins(db, "01110004", "BE1", 0.9999, near)   # near time but different pvs
+    ids = {r["id"] for r in db.find_twins(a, window_seconds=300)}
+    assert ids == {b}

tests/test_insert_events_shape.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+from pathlib import Path
+
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp, PeakValues
+
+
+def _event(waveform_key="0111abcd"):
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex(waveform_key)
+    ev.timestamp = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0,
+                             month=6, day=25, hour=8, minute=50, second=0)
+    ev.record_type = "Waveform"
+    ev.peak_values = PeakValues(tran=0.075, vert=0.220, long=0.045,
+                                 peak_vector_sum=0.231, micl=0.01)
+    return ev
+
+
+def test_insert_stores_shape_from_record(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec = {ev._waveform_key.hex(): {
+        "filename": "F.CE0W", "filesize": 10,
+        "shape_crest_factor": 34.0, "shape_near_peak_count": 3,
+        "shape_sample_count": 1024, "shape_axis": "Long"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec)
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 34.0
+    assert row["shape_near_peak_count"] == 3
+    assert row["shape_axis"] == "Long"
+
+
+def test_upsert_refreshes_shape_from_record(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec1 = {ev._waveform_key.hex(): {
+        "shape_crest_factor": 10.0, "shape_near_peak_count": 1,
+        "shape_sample_count": 512, "shape_axis": "Tran"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec1)
+
+    ev2 = _event()
+    rec2 = {ev2._waveform_key.hex(): {
+        "shape_crest_factor": 22.5, "shape_near_peak_count": 7,
+        "shape_sample_count": 2048, "shape_axis": "Vert"}}
+    db.insert_events([ev2], serial="BE1", waveform_records=rec2)
+
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 22.5
+    assert row["shape_near_peak_count"] == 7
+    assert row["shape_sample_count"] == 2048
+    assert row["shape_axis"] == "Vert"
+
+
+def test_upsert_preserves_shape_when_record_missing(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = _event()
+    rec1 = {ev._waveform_key.hex(): {
+        "shape_crest_factor": 10.0, "shape_near_peak_count": 1,
+        "shape_sample_count": 512, "shape_axis": "Tran"}}
+    db.insert_events([ev], serial="BE1", waveform_records=rec1)
+
+    ev2 = _event()  # re-import with no waveform_records (e.g. sample missing)
+    db.insert_events([ev2], serial="BE1")
+
+    row = db.query_events(serial="BE1")[0]
+    assert row["shape_crest_factor"] == 10.0
+    assert row["shape_near_peak_count"] == 1
+    assert row["shape_sample_count"] == 512
+    assert row["shape_axis"] == "Tran"

tests/test_reviewed_real_column.py

@@ -0,0 +1,58 @@
+import sqlite3
+from sfm.database import SeismoDb
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+def test_fresh_db_has_reviewed_real(tmp_path):
+    assert "reviewed_real" in _cols(SeismoDb(tmp_path/"s.db"))
+def test_existing_db_migrates_reviewed_real(tmp_path):
+    p = tmp_path/"s.db"; db = SeismoDb(p)
+    with sqlite3.connect(p) as c:
+        c.execute("ALTER TABLE events DROP COLUMN reviewed_real")
+    assert "reviewed_real" not in _cols(db)      # dropped (read via existing handle/connection)
+    SeismoDb(p)                                   # re-open migrates
+    assert "reviewed_real" in _cols(SeismoDb(p))
+
+def test_legacy_pre_migration1_db_rebuild_does_not_crash(tmp_path):
+    # Genuinely legacy (pre-Migration-1) events table: no UNIQUE(serial, timestamp)
+    # (still on the old UNIQUE(serial, waveform_key)) and columns only through
+    # false_trigger/created_at — i.e. none of the columns added later by the
+    # Migration-1b ADD COLUMN loop (blastware_filename, device_family,
+    # tran_zc_freq, shape_*, reviewed_real, ...). Opening this DB triggers the
+    # Migration-1 rebuild (`events_old` -> `events` via positional
+    # `INSERT OR IGNORE INTO events SELECT * FROM events_old`), which requires
+    # the rebuild's CREATE TABLE to have exactly the legacy column count.
+    p = tmp_path / "legacy.db"
+    with sqlite3.connect(p) as c:
+        c.execute("""
+            CREATE TABLE events (
+                id                      TEXT PRIMARY KEY,
+                serial                  TEXT NOT NULL,
+                waveform_key            TEXT NOT NULL,
+                session_id              TEXT,
+                timestamp               TEXT,
+                tran_ppv                REAL,
+                vert_ppv                REAL,
+                long_ppv                REAL,
+                peak_vector_sum         REAL,
+                mic_ppv                 REAL,
+                project                 TEXT,
+                client                  TEXT,
+                operator                TEXT,
+                sensor_location         TEXT,
+                sample_rate             INTEGER,
+                record_type             TEXT,
+                false_trigger           INTEGER NOT NULL DEFAULT 0,
+                created_at              TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
+                UNIQUE(serial, waveform_key)
+            )
+        """)
+        c.execute(
+            "INSERT INTO events (id, serial, waveform_key, timestamp) VALUES (?, ?, ?, ?)",
+            ("evt-1", "BE11529", "01110000", "2026-01-01T00:00:00"),
+        )
+    # Pre-fix this raised: OperationalError: table events has 19 columns but
+    # 18 values were supplied (reviewed_real had leaked into the rebuild's
+    # CREATE TABLE, but events_old — and the positional SELECT * — only had 18).
+    db = SeismoDb(p)
+    assert "reviewed_real" in _cols(db)

tests/test_reviewed_real_in_feed.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+import os, sys
+from pathlib import Path
+
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+
+
+def test_query_events_includes_reviewed_real(tmp_path: Path):
+    db = SeismoDb(tmp_path / "s.db")
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex("01110000")
+    db.insert_events([ev], serial="BE1")
+    assert "reviewed_real" in db.query_events(serial="BE1")[0]

tests/test_save_shape.py

@@ -0,0 +1,13 @@
+from pathlib import Path
+
+from sfm.waveform_store import WaveformStore
+
+_FIX = Path(__file__).parent / "fixtures/histogram-extension-re/events-5-21-26/K558LL8B.7I0W"
+
+
+def test_save_imported_bw_attaches_shape(tmp_path):
+    store = WaveformStore(tmp_path / "waveforms")
+    ev, rec = store.save_imported_bw(_FIX.read_bytes(), source_path=_FIX, serial_hint="BE9558")
+    assert rec.get("shape_axis") in ("Tran", "Vert", "Long")
+    assert rec["shape_crest_factor"] > 0
+    assert rec["shape_sample_count"] > 200

tests/test_set_false_trigger_exclusivity.py

@@ -0,0 +1,29 @@
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+
+
+def _ins(db, eid_key="01110000", serial="BE1"):
+    ev = Event(index=0); ev._waveform_key = bytes.fromhex(eid_key)
+    db.insert_events([ev], serial=serial)
+    return db.query_events(serial=serial)[0]["id"]
+
+
+def test_set_false_trigger_clears_reviewed_real(tmp_path):
+    db = SeismoDb(tmp_path / "s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"reviewed_real": True})
+    assert db.get_event(eid)["reviewed_real"] == 1
+
+    db.set_false_trigger(eid, True)
+    row = db.get_event(eid)
+    assert row["false_trigger"] == 1 and row["reviewed_real"] == 0
+
+
+def test_clearing_false_trigger_leaves_reviewed_real_untouched(tmp_path):
+    db = SeismoDb(tmp_path / "s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"false_trigger": True})
+    db.set_false_trigger(eid, False)
+    row = db.get_event(eid)
+    assert row["false_trigger"] == 0
+    # reviewed_real was never set true, so it stays 0 either way; the point
+    # is set_false_trigger(False) does not touch the column at all.
+    assert row["reviewed_real"] == 0

tests/test_shape_columns.py

@@ -0,0 +1,27 @@
+import sqlite3
+from sfm.database import SeismoDb
+
+_SHAPE_COLS = {"shape_crest_factor", "shape_near_peak_count",
+               "shape_sample_count", "shape_axis"}
+
+def _cols(db):
+    with sqlite3.connect(db.db_path) as c:
+        return {r[1] for r in c.execute("PRAGMA table_info(events)")}
+
+def test_fresh_db_has_shape_columns(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    assert _SHAPE_COLS <= _cols(db)
+
+def test_existing_db_migrates_shape_columns(tmp_path):
+    p = tmp_path / "s.db"
+    db = SeismoDb(p)
+    with sqlite3.connect(p) as c:      # simulate an older DB missing the columns
+        for col in _SHAPE_COLS:
+            c.execute(f"ALTER TABLE events DROP COLUMN {col}")
+    # sanity: dropped. Read via the already-constructed `db` (a raw PRAGMA
+    # read against db.db_path) rather than a fresh SeismoDb(p) — the latter
+    # would immediately re-trigger _migrate's self-healing ADD COLUMN loop
+    # and mask the gap we're trying to confirm.
+    assert not (_SHAPE_COLS <= _cols(db))
+    SeismoDb(p)                        # re-open triggers _migrate
+    assert _SHAPE_COLS <= _cols(SeismoDb(p))

tests/test_shape_metrics.py

@@ -0,0 +1,31 @@
+import numpy as np
+from sfm.shape_metrics import channel_shape, shape_from_samples
+
+def test_needle_spike_high_crest_few_near_peak():
+    x = np.zeros(1024); x[500] = 1.0          # one isolated spike
+    s = channel_shape(x)
+    assert s["sample_count"] == 1024
+    assert s["crest_factor"] > 15             # peak towers over rms
+    assert s["near_peak_count"] <= 3          # almost nothing near the peak
+
+def test_ringing_low_crest_many_near_peak():
+    t = np.arange(1024)
+    x = np.sin(2*np.pi*t/32) * np.exp(-t/4000)  # decaying oscillation
+    s = channel_shape(x)
+    assert s["crest_factor"] < 6
+    assert s["near_peak_count"] > 30          # many samples near the peak
+
+def test_channel_shape_none_for_unusable():
+    assert channel_shape(np.zeros(1024)) is None   # flat / all-zero
+    assert channel_shape(np.array([1.0])) is None  # too short
+
+def test_shape_from_samples_picks_max_peak_axis():
+    chans = {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": np.zeros(1024)}
+    chans["Long"][10] = 0.5
+    chans["Vert"] = np.sin(np.arange(1024)/5) * 0.01
+    s = shape_from_samples(chans)
+    assert s["axis"] == "Long"                # Long has the biggest peak
+    assert s["near_peak_count"] <= 3
+
+def test_shape_from_samples_none_when_no_geo():
+    assert shape_from_samples({"MicL": np.ones(1024)}) is None

tests/test_shape_metrics_h5.py

@@ -0,0 +1,22 @@
+import numpy as np, h5py
+from sfm.shape_metrics import shape_from_h5
+
+def _write_h5(path, chans):
+    with h5py.File(path, "w") as f:
+        g = f.create_group("samples")
+        for k, v in chans.items():
+            g.create_dataset(k, data=np.asarray(v, dtype="float32"))
+
+def test_shape_from_h5_reads_dominant_axis(tmp_path):
+    p = tmp_path / "ev.h5"
+    long = np.zeros(1024, dtype="float32"); long[100] = 0.48
+    _write_h5(p, {"Tran": np.zeros(1024), "Vert": np.zeros(1024), "Long": long,
+                  "MicL": np.ones(1024)})
+    s = shape_from_h5(str(p))
+    assert s["axis"] == "Long" and s["near_peak_count"] <= 3
+
+def test_shape_from_h5_none_on_missing_or_degenerate(tmp_path):
+    assert shape_from_h5(str(tmp_path / "nope.h5")) is None
+    p = tmp_path / "degen.h5"
+    _write_h5(p, {"Tran": np.zeros(1), "Vert": np.zeros(1), "Long": np.zeros(1)})
+    assert shape_from_h5(str(p)) is None

tests/test_twin_propagation.py

@@ -0,0 +1,33 @@
+import sqlite3
+from sfm.database import SeismoDb
+from minimateplus.models import Event, Timestamp
+
+
+def _ins(db, key, serial, pvs, ts):
+    ev = Event(index=0)
+    ev._waveform_key = bytes.fromhex(key)
+    ev.timestamp = ts
+    # peak_vector_sum comes from peak_values; simplest: insert then UPDATE pvs directly
+    db.insert_events([ev], serial=serial)
+    row = [r for r in db.query_events(serial=serial) if r["waveform_key"] == key][0]
+    with sqlite3.connect(db.db_path) as c:
+        c.execute("UPDATE events SET peak_vector_sum=? WHERE id=?", (pvs, row["id"]))
+    return row["id"]
+
+
+def test_propagate_copies_flags_to_twins(tmp_path):
+    db = SeismoDb(tmp_path / "s.db")
+    base = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=5)
+    twin = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=45)
+    other = Timestamp(raw=b"", flag=0x10, year=2026, unknown_byte=0, month=2, day=25, hour=20, minute=19, second=44)
+
+    primary_id = _ins(db, "01110001", "BE1", 0.4763, base)
+    twin_id = _ins(db, "01110002", "BE1", 0.4763, twin)        # twin: same serial+pvs, 40s apart
+    non_twin_id = _ins(db, "01110003", "BE1", 0.9999, other)   # near time but different pvs
+
+    db.update_event_review(primary_id, {"false_trigger": True})
+    moved = db.propagate_review_to_twins(primary_id)
+
+    assert twin_id in moved
+    assert db.get_event(twin_id)["false_trigger"] == 1
+    assert db.get_event(non_twin_id)["false_trigger"] == 0

tests/test_update_event_review_reviewed_real.py

@@ -0,0 +1,25 @@
+from sfm.database import SeismoDb
+from minimateplus.models import Event
+
+
+def _ins(db, eid_key="01110000", serial="BE1"):
+    ev = Event(index=0); ev._waveform_key = bytes.fromhex(eid_key)
+    db.insert_events([ev], serial=serial)
+    return db.query_events(serial=serial)[0]["id"]
+
+
+def test_confirm_real_sets_and_clears_ft(tmp_path):
+    db = SeismoDb(tmp_path/"s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"false_trigger": True})
+    assert db.get_event(eid)["false_trigger"] == 1
+    db.update_event_review(eid, {"reviewed_real": True})   # confirming real clears FT
+    row = db.get_event(eid)
+    assert row["reviewed_real"] == 1 and row["false_trigger"] == 0
+
+
+def test_flag_ft_clears_reviewed_real(tmp_path):
+    db = SeismoDb(tmp_path/"s.db"); eid = _ins(db)
+    db.update_event_review(eid, {"reviewed_real": True})
+    db.update_event_review(eid, {"false_trigger": True})
+    row = db.get_event(eid)
+    assert row["false_trigger"] == 1 and row["reviewed_real"] == 0