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