Praxis CI
c396ded395
TASK-07-01: server/cohort/aggregator.py — aggregate_session with k-anon write-time suppression (D-034, K_ANON_THRESHOLD=10), idempotent upsert, 7-day rolling window, multiple metrics (sessions_count, active_learners, gate_open_rate, median_mastery_score, rubric_criterion_means, failure_mode_frequency, branch distribution). No PII in aggregates (D-031). TASK-07-02: server/cohort/hook.py — on_session_end fire-and-forget (D-054), no-op when no Postgres, failures log + nightly reconciles. TASK-07-03: server/cohort/nightly.py — NightlyScheduler in-process asyncio loop, 03:00 CT (America/Winnipeg approx), reconcile from mastery_gate_events, R-DASH-04 failure handling. TASK-07-04: session_recorder.py — chain aggregation hook after mastery flow via asyncio.create_task (parallel, off voice path, D-054). TASK-07-05: tests/test_cohort_aggregation.py — k-anon threshold (9/10/11), idempotent, 7-day window, metrics, no PII. TASK-07-06: tests/test_cohort_nightly.py — scheduler timing, reconciliation, hook-failure+nightly recovery, R-DASH-04. G-038 (binding): differencing-attack test — 10 learners window A, 9 in B, verify dropped learner cannot be isolated (B suppressed, value=NULL). ---ci--- project: praxis phase: 2 milestone: v0.4 status: execute persona: backend-engineer task: 07-01..07-06 requirements: covered: [REQ-MT-02, REQ-NFR-DASH-02, REQ-NFR-DASH-01] ---/ci---
232 lines
9.1 KiB
Python
232 lines
9.1 KiB
Python
"""Nightly reconciliation scheduler (TASK-07-03, D-054, REQ-NFR-DASH-02).
|
|
|
|
In-process asyncio scheduler (no APScheduler — RESEARCH-v0.4 §3.4). Loops:
|
|
compute seconds until next 03:00 CT (America/Winnipeg — Canada pilot) →
|
|
asyncio.sleep → reconcile all 7-day windows → repeat. Resumes after restart.
|
|
Failures log + retry next night (R-DASH-04).
|
|
|
|
Reconciliation recomputes all (path, metric, window_start) cells from the
|
|
mastery_gate_events audit log + re-applies k-anonymity suppression. This
|
|
guarantees REQ-NFR-DASH-02 (freshness ≤ 24h — the nightly job runs at least
|
|
once/day) and reconciles any hook failures.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import datetime as _dt
|
|
import logging
|
|
import statistics
|
|
from collections import Counter, defaultdict
|
|
from typing import Any
|
|
|
|
from db.pg_store import PgStore
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
CT = _dt.timezone(_dt.timedelta(hours=-5), "CT")
|
|
NIGHTLY_HOUR = 3
|
|
NIGHTLY_MINUTE = 0
|
|
|
|
|
|
def seconds_until_next_03_ct(now: _dt.datetime | None = None) -> float:
|
|
"""Seconds from `now` until the next 03:00 America/Winnipeg (CT).
|
|
|
|
America/Winnipeg observes CST (UTC-6) in winter + CDT (UTC-5) in summer.
|
|
We approximate CT as a fixed UTC-5 offset (the pilot is in summer CDT
|
|
and the scheduler drift of ≤1h over DST boundaries is acceptable for a
|
|
nightly reconciliation job — the on-session-end hook keeps data fresh).
|
|
A future hardening would use zoneinfo.ZoneInfo("America/Winnipeg") with
|
|
proper DST handling.
|
|
"""
|
|
now = now or _dt.datetime.now(CT)
|
|
if now.tzinfo is None:
|
|
now = now.replace(tzinfo=CT)
|
|
next_run = now.replace(hour=NIGHTLY_HOUR, minute=NIGHTLY_MINUTE,
|
|
second=0, microsecond=0)
|
|
if next_run <= now:
|
|
next_run += _dt.timedelta(days=1)
|
|
return (next_run - now).total_seconds()
|
|
|
|
|
|
class NightlyScheduler:
|
|
"""In-process asyncio scheduler for nightly cohort reconciliation.
|
|
|
|
Started as an asyncio task in the app lifespan (TASK-10-02). Cancel on
|
|
shutdown. R-DASH-04: a reconciliation failure logs + retries the next
|
|
night (the loop continues).
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._task: asyncio.Task | None = None
|
|
self._stopped = False
|
|
|
|
async def start(self, pg_store: PgStore) -> asyncio.Task:
|
|
"""Begin the nightly loop. Returns the running task."""
|
|
self._stopped = False
|
|
self._task = asyncio.create_task(self._run_loop(pg_store))
|
|
return self._task
|
|
|
|
async def stop(self) -> None:
|
|
"""Cancel the running loop (graceful shutdown)."""
|
|
self._stopped = True
|
|
if self._task is not None:
|
|
self._task.cancel()
|
|
try:
|
|
await self._task
|
|
except (asyncio.CancelledError, Exception):
|
|
pass
|
|
self._task = None
|
|
|
|
async def _run_loop(self, pg_store: PgStore) -> None:
|
|
while not self._stopped:
|
|
try:
|
|
secs = seconds_until_next_03_ct()
|
|
log.info("nightly scheduler: next run in %.0fs (03:00 CT)", secs)
|
|
await asyncio.sleep(secs)
|
|
if self._stopped:
|
|
return
|
|
await self._reconcile(pg_store)
|
|
except asyncio.CancelledError:
|
|
return
|
|
except Exception:
|
|
log.exception("nightly reconciliation failed — retry next night (R-DASH-04)")
|
|
# brief sleep to avoid a tight error loop if the clock is broken
|
|
await asyncio.sleep(60)
|
|
|
|
async def _reconcile(self, pg_store: PgStore) -> None:
|
|
"""Recompute all 7-day windows for all paths from mastery_gate_events.
|
|
|
|
Reads recent gate events (the audit log, REQ-NFR-MAST-02), groups by
|
|
(path, window_start), recomputes each metric cell, applies k-anon
|
|
suppression, and upserts. Idempotent — re-running produces the same
|
|
aggregates (ON CONFLICT upsert).
|
|
"""
|
|
events = await _load_recent_events(pg_store)
|
|
if not events:
|
|
log.info("nightly reconcile: no recent gate events; nothing to recompute")
|
|
return
|
|
|
|
# Group by path → window_start → list[events]
|
|
by_path_window: dict[tuple[str, _dt.date], list[dict[str, Any]]] = defaultdict(list)
|
|
today = _dt.datetime.now(_dt.timezone.utc).date()
|
|
window_start = today - _dt.timedelta(days=6)
|
|
for ev in events:
|
|
ev_date = _coerce_date(ev.get("recorded_at"))
|
|
if ev_date is None or ev_date < window_start:
|
|
continue
|
|
path = ev.get("path_id") or "unknown"
|
|
by_path_window[(path, window_start)].append(ev)
|
|
|
|
from server.cohort.aggregator import K_ANON_THRESHOLD, _rolling_window
|
|
|
|
ws, we = _rolling_window()
|
|
for (path, _), evs in by_path_window.items():
|
|
learners = {e.get("learner_ref") for e in evs if e.get("learner_ref")}
|
|
active_count = len(learners)
|
|
suppressed = active_count < K_ANON_THRESHOLD
|
|
|
|
# sessions_count
|
|
await pg_store.upsert_cohort_aggregate(
|
|
path, "sessions_count", ws, we,
|
|
None if suppressed else float(len(evs)),
|
|
active_count, suppressed,
|
|
)
|
|
# active_learners_count
|
|
await pg_store.upsert_cohort_aggregate(
|
|
path, "active_learners_count", ws, we,
|
|
None if suppressed else float(active_count),
|
|
active_count, suppressed,
|
|
)
|
|
# gate_open_rate
|
|
gate_opens = sum(1 for e in evs if (e.get("gate_outcome") or "") == "open")
|
|
rate = gate_opens / len(evs) if evs else 0.0
|
|
await pg_store.upsert_cohort_aggregate(
|
|
path, "gate_open_rate", ws, we,
|
|
None if suppressed else rate,
|
|
active_count, suppressed,
|
|
)
|
|
# median_mastery_score + rubric_criterion_means from rubric_scores_jsonb
|
|
score_rows: list[float] = []
|
|
crit_scores: dict[str, list[float]] = defaultdict(list)
|
|
for e in evs:
|
|
scores = e.get("rubric_scores") or []
|
|
if isinstance(scores, str):
|
|
import json as _json
|
|
try:
|
|
scores = _json.loads(scores)
|
|
except Exception:
|
|
scores = []
|
|
for r in scores:
|
|
if isinstance(r, dict):
|
|
cid = r.get("criterion_id") or r.get("id") or "unknown"
|
|
s = r.get("score") or r.get("weighted_mean")
|
|
if s is not None:
|
|
crit_scores[cid].append(float(s))
|
|
score_rows.append(float(s))
|
|
if score_rows:
|
|
med = statistics.median(score_rows)
|
|
await pg_store.upsert_cohort_aggregate(
|
|
path, "median_mastery_score", ws, we,
|
|
None if suppressed else med,
|
|
active_count, suppressed,
|
|
)
|
|
for cid, vals in crit_scores.items():
|
|
mean_v = statistics.mean(vals) if vals else 0.0
|
|
await pg_store.upsert_cohort_aggregate(
|
|
path, f"rubric_criterion_mean:{cid}", ws, we,
|
|
None if suppressed else mean_v,
|
|
active_count, suppressed,
|
|
)
|
|
|
|
log.info("nightly reconcile: recomputed %d (path, window) cells", len(by_path_window))
|
|
|
|
async def reconcile_now(self, pg_store: PgStore) -> None:
|
|
"""Public hook for tests / ad-hoc reconciliation (no clock wait)."""
|
|
await self._reconcile(pg_store)
|
|
|
|
|
|
async def _load_recent_events(pg_store: PgStore) -> list[dict[str, Any]]:
|
|
"""Load mastery_gate_events from the last 7 days.
|
|
|
|
Uses the PgStore pool directly (no extra method on PgStore to keep the
|
|
surface minimal). Returns rows as dicts with decoded rubric_scores.
|
|
"""
|
|
async with pg_store.pool.acquire() as conn:
|
|
rows = await conn.fetch(
|
|
"SELECT learner_ref, scenario_id, path_id, gate_outcome, "
|
|
"rubric_scores_jsonb, recorded_at "
|
|
"FROM mastery_gate_events "
|
|
"WHERE recorded_at >= now() - interval '7 days' "
|
|
"ORDER BY recorded_at"
|
|
)
|
|
out: list[dict[str, Any]] = []
|
|
for r in rows:
|
|
d = dict(r)
|
|
scores = d.get("rubric_scores_jsonb")
|
|
if hasattr(scores, "resolve"):
|
|
try:
|
|
import json as _json
|
|
d["rubric_scores"] = _json.loads(scores.resolve()) if scores else []
|
|
except Exception:
|
|
d["rubric_scores"] = []
|
|
else:
|
|
d["rubric_scores"] = scores
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
def _coerce_date(val: Any) -> _dt.date | None:
|
|
if val is None:
|
|
return None
|
|
if isinstance(val, _dt.datetime):
|
|
return val.date()
|
|
if isinstance(val, _dt.date):
|
|
return val
|
|
try:
|
|
return _dt.datetime.fromisoformat(str(val)).date()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
__all__ = ["NightlyScheduler", "seconds_until_next_03_ct", "CT"] |