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

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

# 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

# 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:

    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

# 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

    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)

# 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:

    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:

        # 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).

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.