Fix bugs from the repo audit: event order, stale data after untrack, reset/export gaps

app/db.py

@@ -468,6 +468,9 @@ def latest_project(sid):
 # (table, channel) pairs whose row-level INSERT/UPDATE/DELETE bump a counter.
 _REV_TRIGGERS = [
     ("student_state", "states"),
+    # Goal profiles are written after the student_state row, so they signal the
+    # same channel; else an open sheet refetches before the newest run's evidence.
+    ("goal_profile", "states"),
     ("trigger_event", "triggers"),
     ("switch_event", "switches"),
     ("tracked_student", "roster"),
@@ -822,6 +825,7 @@ def _workspace_export_queries(workspace_id):
         "student_state": (f"SELECT * FROM student_state WHERE {roster}", (ws,)),
         "trigger_event": (f"SELECT * FROM trigger_event WHERE {roster}", (ws,)),
         "switch_event": (f"SELECT * FROM switch_event WHERE {roster}", (ws,)),
+        "goal_profile": (f"SELECT * FROM goal_profile WHERE {roster}", (ws,)),
         "vex_log": (f"SELECT * FROM vex_log WHERE {vex_roster}", (ws,)),
     }
 
@@ -995,7 +999,9 @@ def triggers_feed(cutoff, limit=100, workspace_id=None, since=None):
         params.append(workspace_id)
     after = ""
     if since is not None:
-        after = "AND te.started_at >= ? "
+        # A reset hides earlier alerts, except one still open: that is the
+        # student's current state (e.g. idle right now), not history.
+        after = "AND (te.resolved_at IS NULL OR te.started_at >= ?) "
         params.append(dt_to_db(since))
     params += [dt_to_db(cutoff), limit]
     rows = _query(
@@ -1255,6 +1261,15 @@ def tracked_add(sid, workspace_id=None):
     )
 
 
+def mark_workspace_for_backfill(workspace_id=None):
+    """Ask the daemon to backfill every student on this workspace's roster again.
+    Used when a paused board resumes: while it was paused its students weren't
+    fetched, but the shared cursor moved on if another board was live."""
+    _execute(
+        "UPDATE tracked_student SET backfilled = 0 WHERE workspace_id = ?", (_ws(workspace_id),)
+    )
+
+
 def mark_backfilled(sid):
     # Backfill pulls a student's shared history once, so mark every workspace's
     # roster row for that student (the daemon ingests the union, per student).
@@ -1290,6 +1305,10 @@ def tracked_remove(sid, workspace_id=None):
             con.executemany("DELETE FROM message WHERE id = ?", [(m,) for m in msg_ids])
         con.execute("DELETE FROM student_state WHERE studentID = ?", (sid,))
         con.execute("DELETE FROM trigger_event WHERE studentID = ?", (sid,))
+        # Everything else derived from their events goes too, or a re-add would
+        # mix these stale goal runs and switches into the rebuilt ones.
+        con.execute("DELETE FROM goal_profile WHERE studentID = ?", (sid,))
+        con.execute("DELETE FROM switch_event WHERE studentID = ?", (sid,))
 
 
 # ==========================================================================

app/main.py

@@ -893,7 +893,12 @@ def set_polling(body: PollingBody, wsid: int = Depends(current_workspace_id)):
     paused the daemon makes no prod requests on this board's behalf; it keeps
     running and picks back up within about a second of being re-enabled. Purely a
     control flag, prod is untouched either way."""
+    was_enabled = _polling_enabled(wsid)
     db.set_workspace_setting(wsid, "polling_enabled", "1" if body.enabled else "0")
+    if body.enabled and not was_enabled:
+        # Refill the pause gap: backfill is idempotent, and the daemon rebuilds
+        # each student's worker from the log afterwards.
+        db.mark_workspace_for_backfill(wsid)
     return {"enabled": body.enabled}
 
 

app/pipeline/poller.py

@@ -105,6 +105,11 @@ def drain(client, cursor, limit=500, overlap_seconds=2, tracked=None, since=None
     max_et = cursor.last_event_time
     max_id = cursor.last_source_id or 0
 
+    # Collect every page first. Prod pages newest-first, and the workers must see
+    # each student's events in the order they happened (a run's outcome follows
+    # its run; run numbers count up), so events are persisted and routed only
+    # after sorting them oldest-first.
+    wanted = []
     while True:
         results = client.page_by_time(date_from, limit, offset)
         if not results:
@@ -120,28 +125,33 @@ def drain(client, cursor, limit=500, overlap_seconds=2, tracked=None, since=None
                 continue
             if since is not None and et is not None and et < since:
                 continue  # before the session cutoff -- skip, but the cursor still advanced above
-            inserted, norm = persist(ev)
-            if inserted:
-                try:
-                    route(norm)
-                    new_count += 1
-                except Exception:
-                    # The row is already in vex_log, but routing it into the
-                    # in-memory worker failed. Discard that worker so the next
-                    # tick rebuilds it from the DB (which now has this event),
-                    # rather than keep serving a buffer that's silently missing it.
-                    from app.pipeline import workers
-
-                    workers._workers.pop(db.canon_id(norm["studentID"]), None)
-                    logger.exception(
-                        "route failed for %s after persist; dropped worker for rehydrate",
-                        norm["studentID"],
-                    )
-                    raise
+            wanted.append((et, sid or 0, ev))
         if len(results) < limit:
             break
         offset += limit
 
+    _EPOCH = datetime.min.replace(tzinfo=UTC)
+    wanted.sort(key=lambda w: (w[0] or _EPOCH, w[1]))
+    for _, _, ev in wanted:
+        inserted, norm = persist(ev)
+        if inserted:
+            try:
+                route(norm)
+                new_count += 1
+            except Exception:
+                # The row is already in vex_log, but routing it into the
+                # in-memory worker failed. Discard that worker so the next
+                # tick rebuilds it from the DB (which now has this event),
+                # rather than keep serving a buffer that's silently missing it.
+                from app.pipeline import workers
+
+                workers._workers.pop(db.canon_id(norm["studentID"]), None)
+                logger.exception(
+                    "route failed for %s after persist; dropped worker for rehydrate",
+                    norm["studentID"],
+                )
+                raise
+
     # Drain finished and everything is durably written, now advance the cursor.
     if max_et != cursor.last_event_time or max_id != (cursor.last_source_id or 0):
         cursor.last_event_time = max_et

docs/concepts/write-path.md

@@ -70,6 +70,10 @@ correctness machinery in the whole thing.
 - Each drain pages prod with `dateFrom = last_event_time - overlap`, where overlap is
   a 2-second window, so events sitting right on a timestamp boundary don't slip
   through.
+- Prod returns pages newest-first, so a drain collects every page, sorts the events
+  oldest-first, and only then persists and routes them. Workers must see a student's
+  events in the order they happened: a run's outcome follows its run, and run
+  numbers count up.
 - It persists, then advances. The cursor only moves after a full drain is safely
   written.
 - Inserts are idempotent. Every event has a unique `source_event_id`, so re-fetched

frontend/src/CohortDashboard.jsx

@@ -2245,17 +2245,21 @@ const CohortDashboard = () => {
     try {
       setNotes((await api.get("/api/notes/", { params: { studentID: sid } })).data.notes || []);
     } catch {
-      setNotes([]);
+      /* keep what's shown: a blip must not look like "no notes" */
     }
   }, []);
   React.useEffect(() => {
+    setNotes([]); // a different student: never show the previous one's notes
     fetchNotes(selected);
   }, [selected, fetchNotes]);
 
   // Trigger history for the open student's grid. Depending on `triggers` too
   // keeps the grid current while the modal is open: whenever the live feed
   // refetches (a new alert, an ack), the history follows on the same beat.
   const [history, setHistory] = React.useState([]);
+  React.useEffect(() => {
+    setHistory([]); // a different student: never show the previous one's history
+  }, [selected]);
   React.useEffect(() => {
     if (!selected) {
       setHistory([]);
@@ -2268,7 +2272,7 @@ const CohortDashboard = () => {
         if (alive) setHistory(r.data.history || []);
       })
       .catch(() => {
-        if (alive) setHistory([]);
+        /* keep what's shown: a blip must not empty the grid */
       });
     return () => {
       alive = false;
@@ -2301,7 +2305,8 @@ const CohortDashboard = () => {
         const d = (await api.get(`/api/student_states/${encodeURIComponent(selected)}/`)).data;
         if (alive) setDetailFull(d);
       } catch {
-        if (alive) setDetailFull(null);
+        /* keep the last good payload; the selection effect above already
+           cleared it if this is a different student */
       }
       if (alive) setDetailFor(selected);
     })();

tests/test_audit_fixes.py

@@ -0,0 +1,94 @@
+"""Regression tests for the 2026-09-25 bug audit."""
+
+import io
+import zipfile
+from datetime import timedelta
+
+from app import db
+from app.pipeline import poller, workers
+
+
+class _TimeClient:
+    """Serves page_by_time pages the way prod does: newest-first."""
+
+    def __init__(self, pages):
+        self.pages = pages
+
+    def page_by_time(self, date_from, limit, offset):
+        idx = offset // limit
+        return self.pages[idx] if idx < len(self.pages) else []
+
+
+def _ev(eid, sid, ts):
+    return {
+        "id": eid,
+        "studentID": sid,
+        "classCode": "C1",
+        "eventType": "runProject",
+        "project": "{}",
+        "raw_message": "{}",
+        "recieved_at": ts,
+    }
+
+
+def test_drain_feeds_a_newest_first_page_oldest_first():
+    """Prod pages newest-first. Routing in page order fed the worker backwards,
+    so runs in one tick were numbered newest-first and an outcome could land on
+    the wrong run."""
+    db.tracked_add("d1")
+    db.mark_backfilled("d1")
+    stamps = ["2026-09-25T10:00:00Z", "2026-09-25T10:00:05Z", "2026-09-25T10:00:10Z"]
+    page = [_ev(i + 1, "d1", t) for i, t in enumerate(stamps)][::-1]  # newest-first
+    cursor = poller.get_cursor()
+    poller.drain(_TimeClient([page]), cursor, limit=500, tracked={"d1"})
+    w = workers.get_worker("d1")
+    w.recompute_and_write()
+    ts = [r["ts"] for r in db.list_student_states(["d1"])[0]["runs"]["runs"]]
+    assert len(ts) == 3 and ts == sorted(ts)
+
+
+def test_untracking_purges_goal_evidence_and_switches():
+    """Removing a student from the last board must purge everything derived from
+    their events, or a re-add mixes stale goal runs into the rebuilt ones."""
+    db.tracked_add("u1")
+    db.upsert_goal_profile("u1", 7, {"index": 7, "status": "profiled", "ts": 1.0})
+    db.record_switch("u1", "casing", "U1", "u1")
+    db.tracked_remove("u1")
+    assert db.list_goal_profiles("u1", session=None) == []
+    assert db.list_switches() == []
+
+
+def test_goal_evidence_writes_signal_the_states_channel():
+    """Goal profiles are written after student_state, so without their own signal
+    an open sheet refetches before the newest run's evidence exists."""
+    before = db.data_fingerprint()["states"]
+    db.upsert_goal_profile("g1", 0, {"index": 0, "status": "profiled", "ts": 1.0})
+    assert db.data_fingerprint()["states"] != before
+
+
+def test_reset_keeps_alerts_that_are_still_open(client, seed_state):
+    """Reset hides history, but an alert still holding (an idle student) is the
+    student's current state and must stay in the feed."""
+    seed_state("o1")
+    started = db.now() - timedelta(minutes=5)
+    db.create_trigger("o1", "inactive", started, started, None, {"label": "Inactive"})
+    client.post("/api/reset/")
+    feed = client.get("/api/triggers/").json()["triggers"]
+    assert [t["trigger_type"] for t in feed] == ["inactive"]
+
+
+def test_export_includes_goal_evidence(seed_state):
+    seed_state("e1")
+    db.upsert_goal_profile("e1", 0, {"index": 0, "status": "profiled", "ts": 1.0})
+    names = zipfile.ZipFile(io.BytesIO(db.export_zip_bytes())).namelist()
+    assert any(n.endswith("goal_profile.csv") for n in names)
+
+
+def test_resuming_polling_rebackfills_the_boards_students(client, seed_state):
+    """While a board is paused its students aren't fetched, but the shared cursor
+    moves on if another board is live. Resuming must pull the gap back in."""
+    seed_state("p1")
+    db.mark_backfilled("p1")
+    client.post("/api/polling/", json={"enabled": False})
+    client.post("/api/polling/", json={"enabled": True})
+    assert db.tracked_list()[0]["backfilled"] is False