diff --git a/delphi/polismath/replay/__init__.py b/delphi/polismath/replay/__init__.py new file mode 100644 index 000000000..6c4c8eb7d --- /dev/null +++ b/delphi/polismath/replay/__init__.py @@ -0,0 +1,35 @@ +"""Replay harness + R2 schedule inference (``polismath.replay``). + +Two overlapping efforts share this package: + +- **Replay harness (Phase H-A, this branch)** — replay a conversation's vote + history through the math engine at explicitly-chosen recompute points + ("schedules"), recording full per-step state for step-by-step comparison. + See ``delphi/docs/REPLAY_HARNESS_DESIGN.md``. Modules: :mod:`schedule`, + :mod:`driver`, :mod:`store`, :mod:`stepcompare`. +- **R2 schedule inference** — posterior inference of the *latent* recompute + schedule of a historic conversation. Modules (added on the R2 branch): dp, + emission, correction, scan, physics, weights, synthetic, experiments. + +The shared foundation is :mod:`types` (event/dataset types) and +:mod:`real_data` (export-CSV loader), lifted verbatim from the R2 branch so a +later rebase dedups cleanly. +""" + +from polismath.replay.types import ( + CommentMeta, + ModEvent, + ReplayDataset, + Schedule, + Vote, + VoteEvent, +) + +__all__ = [ + "CommentMeta", + "ModEvent", + "ReplayDataset", + "Schedule", + "Vote", + "VoteEvent", +] diff --git a/delphi/polismath/replay/driver.py b/delphi/polismath/replay/driver.py new file mode 100644 index 000000000..2f556f95b --- /dev/null +++ b/delphi/polismath/replay/driver.py @@ -0,0 +1,195 @@ +"""Python replay driver — replay harness Phase H-A (design §6). + +Chains :meth:`polismath.conversation.conversation.Conversation.update_votes` +over the schedule's batches, recomputing at each cut point, and records full +per-step state. ``update_votes`` is pure-functional (deepcopy → new object, +conversation.py:255-485, deepcopy at :269), so the driver is a straight fold over +batches — this +same driver is the future R2 forward model (design §1.3). + +Verified API facts (read from conversation.py, NOT guessed): + +- ``Conversation(zid, last_updated=…)`` — NB ``self.last_updated = last_updated + or int(time.time()*1000)``: **0 is falsy**, so a 0 base silently falls back to + wall-clock. The driver seeds a non-zero base (first vote's t_ms) to stay + deterministic. +- ``update_votes({'votes': [{pid,tid,vote,created}], 'lastVoteTimestamp': ms}, + recompute=bool)`` → new Conversation. Within a batch it keeps the LAST vote + per (pid,tid); across batches the reindex+where merge overwrites cells, so + feeding sorted votes gives later-vote-wins. ``last_updated`` becomes + ``max(lastVoteTimestamp, prev)`` — deterministic given the batch max. +- ``update_moderation({'mod_out_tids','mod_in_tids','meta_tids','mod_out_ptpts'}, + recompute=bool)`` → new Conversation. Quirk: each set is replaced only when + its list is truthy (conversation.py:616-626), so an EMPTY list cannot clear a + previously-set set. This seam is BROADER than "all moderation removed": ANY + single set emptying is silently retained — e.g. un-moderating the LAST mod_out + tid while mod_in is still active leaves that tid zeroed. The driver passes full + cumulative (latest-wins) sets and now DETECTS an emptying transition + (:func:`_guard_moderation_clear`), failing loudly rather than recording a stale + state. ``meta_tids`` / ``mod_out_ptpts`` are out of H-A scope (never wired by + this driver); the real clear-semantics fix is a conversation.py change tracked + on the seam wishlist. +- ``recompute()`` → new Conversation recomputing PCA→clusters→repness→ + priorities→participant-info on the moderation-applied matrix. Standalone + after an ``update_votes(recompute=False)``. + +Vote-sign convention (design §5, D1b journal): export CSVs are ALREADY in +Delphi convention (AGREE=+1) — the raw-DB→export flip lives in +server/src/report.ts. ``VoteEvent.sign`` carries that export sign and +``update_votes`` consumes it AS-IS (no flip). The FUTURE Clojure driver (H-B) +must feed raw-DB signs (flipped); the store records which convention was used. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Any, Callable + +from polismath.conversation.conversation import Conversation +from polismath.replay.schedule import ReplayStep, ScheduleSpec, slice_schedule +from polismath.replay.types import ReplayDataset + +# Vote sign convention recorded in provenance; the future Clojure driver flips. +VOTE_SIGN_CONVENTION = "delphi" # AGREE=+1 (export convention, no re-flip) + + +@dataclass +class StepRecord: + """Everything recorded at one replay step (design §6).""" + + index: int + prev_slot: int + cut_slot: int + batch_size: int + cut_time_ms: int + blob: dict[str, Any] # Conversation.to_dict() — the cross-language surface + extras: dict[str, Any] = field(default_factory=dict) # cheap diagnostics + + +def run_replay( + dataset: ReplayDataset, + spec: ScheduleSpec, + *, + progress: Callable[[int, int], None] | None = None, +) -> list[StepRecord]: + """Replay ``dataset`` through the math engine on ``spec``'s schedule. + + Returns one :class:`StepRecord` per cut slot, in order. Deterministic given + (dataset, spec): PCA power-iteration and k-means are fixed-seeded, and all + timestamps are data-derived — the only wall-clock field is the blob's + ``math_tick`` (conversation.py:2226). ``progress(i, total)`` is called + before each step if provided. + """ + steps = slice_schedule(dataset, spec) + total = len(steps) + + # `or 1`: a first vote at t_ms==0 would seed last_updated=0, which + # Conversation's `last_updated or now` footgun (conversation.py:205) turns + # into wall-clock — breaking determinism. Floor to 1 (nonzero). + base_last_updated = (dataset.votes[0].t_ms or 1) if dataset.votes else 1 + conv = Conversation(spec.dataset, last_updated=base_last_updated) + + # Cumulative latest-wins moderation value per tid across the whole replay. + mod_state: dict[int, int] = {} + + records: list[StepRecord] = [] + for step in steps: + if progress is not None: + progress(step.index, total) + + conv = conv.update_votes(_votes_dict(step), recompute=False) + + if step.mod_events: + for m in step.mod_events: + mod_state[m.tid] = m.mod + mod = _mod_dict(mod_state) + _guard_moderation_clear(conv, mod) + conv = conv.update_moderation(mod, recompute=True) + else: + conv = conv.recompute() + + records.append( + StepRecord( + index=step.index, + prev_slot=step.prev_slot, + cut_slot=step.cut_slot, + batch_size=len(step.vote_events), + cut_time_ms=step.cut_time_ms, + blob=conv.to_dict(), + extras=_step_extras(conv), + ) + ) + return records + + +def _votes_dict(step: ReplayStep) -> dict[str, Any]: + """Map a batch of VoteEvents to update_votes' expected payload. + + ``created`` carries each vote's own timestamp; ``lastVoteTimestamp`` is the + batch max (== the sorted batch's last vote) so ``last_updated`` advances + deterministically. Signs pass through in Delphi convention (see module doc). + """ + votes = [ + {"pid": v.pid, "tid": v.tid, "vote": v.sign, "created": v.t_ms} + for v in step.vote_events + ] + return {"votes": votes, "lastVoteTimestamp": step.cut_time_ms} + + +def _guard_moderation_clear(conv: Conversation, mod: dict[str, list[int]]) -> None: + """Fail loudly on a moderation-set emptying transition the engine can't apply. + + ``Conversation.update_moderation`` replaces ``mod_out_tids`` / ``mod_in_tids`` + only when the incoming list is TRUTHY (conversation.py:616-626), so an EMPTY + list can NOT clear a previously-applied set. If the schedule un-moderates the + LAST tid of a set while the conversation still holds it non-empty, the driver + would silently record the stale (still-zeroed) tids. Rather than emit a wrong + recording, raise — this is the H-A moderation-clear seam; the real fix is a + ``conversation.py`` change (clear on empty), tracked on the seam wishlist. + (``meta_tids`` / ``mod_out_ptpts`` are out of H-A scope: the driver never + wires them, so they are not guarded here.) + """ + stale: list[str] = [] + if not mod["mod_out_tids"] and getattr(conv, "mod_out_tids", None): + stale.append("mod_out_tids") + if not mod["mod_in_tids"] and getattr(conv, "mod_in_tids", None): + stale.append("mod_in_tids") + if stale: + raise NotImplementedError( + "replay driver cannot represent clearing " + f"{', '.join(stale)}: Conversation.update_moderation ignores an empty " + "list, so the previously-moderated tids would silently persist. Wire " + "the real clear semantics (conversation.py update_moderation seam) " + "before replaying a schedule that empties a moderation set." + ) + + +def _mod_dict(mod_state: dict[int, int]) -> dict[str, list[int]]: + """Cumulative moderation sets from latest-wins per-tid mod values. + + -1 → moderated-out, 1 → moderated-in, 0 → unmoderated (absent from both). + """ + return { + "mod_out_tids": sorted(t for t, v in mod_state.items() if v == -1), + "mod_in_tids": sorted(t for t, v in mod_state.items() if v == 1), + } + + +def _step_extras(conv: Conversation) -> dict[str, Any]: + """Cheap, read-only diagnostics (design §6) — NO production-code changes. + + All fields are derived from already-computed conversation state; none add a + computation seam. ``n_in_conv`` is the clustering in-conv count (may differ + from the blob's ``in-conv``, which uses the to_dict threshold) — a useful + localization signal when steps diverge. + """ + return { + "n_participants": int(conv.participant_count), + "n_comments": int(conv.comment_count), + "n_votes": int(conv.vote_stats.get("n_votes", 0)), + "n_base_clusters": len(conv.base_clusters or []), + "n_group_clusters": len(conv.group_clusters or []), + "n_in_conv": len(conv._get_in_conv_participants()), + "n_mod_out": len(conv.mod_out_tids), + "pca_present": conv.pca is not None, + } diff --git a/delphi/polismath/replay/real_data.py b/delphi/polismath/replay/real_data.py new file mode 100644 index 000000000..1202f4c47 --- /dev/null +++ b/delphi/polismath/replay/real_data.py @@ -0,0 +1,54 @@ +"""Loaders for real exported datasets (public, under ``delphi/real_data``). + +Export vote CSVs have columns ``timestamp,datetime,comment-id,voter-id,vote`` +with **second**-resolution timestamps. + +Sign caveat (era A only): export CSVs carry *flipped* signs relative to the +raw DB votes the math consumed (the flip is export-only — +math/src/polismath/darwin/export.clj:106-113). Era-B inference ignores signs +entirely (dom membership only). Before using era-A weights on export data, +audit the mapping; until then :func:`load_export_votes` stores the export +sign verbatim and era-A runs on export data are marked diagnostic-only. + +Datasets are located by slug glob (``real_data/*-``) so report-id +directory names never appear in code. +""" + +import csv +from pathlib import Path + +from polismath.replay.types import ReplayDataset + +REAL_DATA_ROOT = Path(__file__).resolve().parents[2] / "real_data" + + +def dataset_dir(slug: str) -> Path | None: + hits = sorted(REAL_DATA_ROOT.glob(f"*-{slug}")) + return hits[0] if hits else None + + +def load_export_votes(slug: str) -> ReplayDataset: + """Load an exported votes CSV into a ReplayDataset. + + Comment creation times are inferred as first-vote times (lower bound on + availability; adequate because a comment is unobservable in the mark + likelihood before its first vote anyway). + """ + d = dataset_dir(slug) + if d is None: + raise FileNotFoundError(f"no dataset directory matching *-{slug}") + votes_csvs = sorted(d.glob("*-votes.csv")) + if not votes_csvs: + raise FileNotFoundError(f"no *-votes.csv in {d.name}") + raw: list[tuple[int, int, int, int]] = [] + with open(votes_csvs[0], newline="") as fh: + for row in csv.DictReader(fh): + raw.append( + ( + int(row["timestamp"]) * 1000, + int(row["voter-id"]), + int(row["comment-id"]), + int(row["vote"]), + ) + ) + return ReplayDataset.build(raw) diff --git a/delphi/polismath/replay/schedule.py b/delphi/polismath/replay/schedule.py new file mode 100644 index 000000000..11a0c901e --- /dev/null +++ b/delphi/polismath/replay/schedule.py @@ -0,0 +1,326 @@ +"""Schedule spec (JSON) + slicer + presets — replay harness Phase H-A. + +A *schedule* is a first-class INPUT to the harness (REPLAY_HARNESS_DESIGN.md +§4): it declares WHERE a recompute fires along a conversation's **sorted** +event stream. R2 (schedule inference) consumes the same spec object, so the +resolution of cut modes into 1-based vote *slots* is kept consistent with +:meth:`polismath.replay.types.ReplayDataset.validate_schedule` / +:meth:`~polismath.replay.types.ReplayDataset.segments` — a cut slot ``s`` means +"a recompute fired after ingesting votes ``1..s``". + +Why sort first (design §5): the export CSVs are NOT pre-sorted (vw has 2136 +out-of-order rows). :meth:`ReplayDataset.build` sorts stably by +``(t_ms, input order)`` and flags revotes; this module operates on that sorted +stream so it does NOT inherit ``prepare_votes_data``'s unsorted-file-order +quirk. Revotes are KEPT (no dedup) — later-vote-wins is resolved inside the +engine, not at the source. + +Cut modes (design §4): +- ``vote-count`` : ``at`` are absolute vote counts (== slots). ``"end"`` → n. +- ``explicit-event-index``: ``at`` are 1-based sorted vote indices (== slots). +- ``timestamp`` : ``at`` are t_ms values; slot = #votes with ``t_ms <= T``. +- ``fraction`` : ``at`` are fractions in (0, 1]; slot = ``round(f * n)``. + +Presets: ``uniform-N``, ``front-loaded``, ``back-loaded``, ``every-vote`` +(small datasets only), ``single-cut`` (== today's cold-start), ``per-day`` +(day boundaries from the real timestamps). +""" + +from __future__ import annotations + +import json +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any + +from polismath.replay.types import ModEvent, ReplayDataset, Schedule, VoteEvent + +_END = "end" +_VALID_MODES = frozenset( + {"vote-count", "explicit-event-index", "timestamp", "fraction"} +) + + +# --------------------------------------------------------------------------- +# Schedule spec (the JSON input, preserved verbatim for the store). +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class ScheduleSpec: + """A (dataset, schedule) declaration — the verbatim §4 JSON input. + + ``to_dict`` returns exactly the mapping the spec was built from so the store + can persist ``schedule.json`` byte-faithfully (design §7): any replay is + re-derivable from (schedule, dataset, commit). + """ + + dataset: str + schedule_id: str + cuts: dict[str, Any] + source: str = "votes-csv" + # "none" | "interleave-by-timestamp" | explicit list of ModEvent-shaped dicts + moderation: Any = "none" + clojure: dict[str, Any] = field(default_factory=lambda: {"warm_start": "chain"}) + notes: str = "" + # Verbatim mapping this spec was loaded from (None → reconstruct on demand). + _raw: dict[str, Any] | None = field(default=None, repr=False, compare=False) + + @classmethod + def from_dict(cls, d: dict[str, Any]) -> "ScheduleSpec": + """Build from a §4 mapping, retaining it verbatim for round-tripping.""" + return cls( + dataset=d["dataset"], + schedule_id=d["schedule_id"], + cuts=d["cuts"], + source=d.get("source", "votes-csv"), + moderation=d.get("moderation", "none"), + clojure=d.get("clojure", {"warm_start": "chain"}), + notes=d.get("notes", ""), + _raw=dict(d), + ) + + @classmethod + def from_json_file(cls, path: str | Path) -> "ScheduleSpec": + with open(path) as fh: + return cls.from_dict(json.load(fh)) + + def to_dict(self) -> dict[str, Any]: + """Return the verbatim input mapping (or reconstruct a canonical one).""" + if self._raw is not None: + return dict(self._raw) + return { + "dataset": self.dataset, + "schedule_id": self.schedule_id, + "source": self.source, + "cuts": self.cuts, + "moderation": self.moderation, + "clojure": self.clojure, + "notes": self.notes, + } + + def write_json(self, path: str | Path) -> None: + with open(path, "w") as fh: + json.dump(self.to_dict(), fh, indent=2) + + +# --------------------------------------------------------------------------- +# One replay step: a batch of votes + newly-active moderation, and its cut. +# --------------------------------------------------------------------------- +@dataclass(frozen=True) +class ReplayStep: + """A single (batch-ingest → recompute → record) unit (design §2). + + ``prev_slot``/``cut_slot`` are the half-open ``(prev_slot, cut_slot]`` vote + range (1-based, inclusive right). ``vote_events`` is that batch in sorted + order; ``mod_events`` are moderation changes that become active in this + segment (interleave); ``cut_time_ms`` is the wall-clock of the batch's last + vote (the cut's clock, used to interleave moderation). + """ + + index: int + prev_slot: int + cut_slot: int + vote_events: tuple[VoteEvent, ...] + mod_events: tuple[ModEvent, ...] + cut_time_ms: int + + +# --------------------------------------------------------------------------- +# Cut-mode resolution → strictly-increasing 1-based slots. +# --------------------------------------------------------------------------- +def resolve_cut_slots(dataset: ReplayDataset, cuts: dict[str, Any]) -> Schedule: + """Resolve a §4 ``cuts`` spec into a validated :data:`Schedule`. + + Returns a strictly-increasing tuple of 1-based slots in ``1..n``. Slots + that resolve to 0 (e.g. a timestamp before the first vote) are dropped as + degenerate — recomputing an empty conversation is a no-op. Duplicate slots + are collapsed; out-of-range slots raise via ``validate_schedule``. + """ + mode = cuts.get("mode") + if mode not in _VALID_MODES: + raise ValueError( + f"unknown cut mode {mode!r}; expected one of {sorted(_VALID_MODES)}" + ) + at = cuts.get("at", []) + n = dataset.n + + raw_slots: list[int] = [] + for a in at: + if a == _END: + raw_slots.append(n) + elif mode in ("vote-count", "explicit-event-index"): + raw_slots.append(int(a)) + elif mode == "fraction": + f = float(a) + if not 0.0 < f <= 1.0: + raise ValueError(f"fraction cut {f} outside (0, 1]") + raw_slots.append(int(round(f * n))) + elif mode == "timestamp": + # slot = number of votes with t_ms <= T (votes are time-sorted). + raw_slots.append(_count_votes_up_to(dataset.votes, int(a))) + + # Drop degenerate 0-slots, dedupe, sort. + slots = tuple(sorted({s for s in raw_slots if s > 0})) + schedule: Schedule = slots + # validate_schedule enforces 1<=s<=n and strict monotonicity. + dataset.validate_schedule(schedule) + return schedule + + +def _count_votes_up_to(votes: list[VoteEvent], t_ms: int) -> int: + """#votes with ``t_ms <= T`` in a time-sorted list (linear; n is small).""" + count = 0 + for v in votes: + if v.t_ms <= t_ms: + count += 1 + else: + break + return count + + +# --------------------------------------------------------------------------- +# Slicer: schedule spec + dataset → ordered replay steps. +# --------------------------------------------------------------------------- +def slice_schedule(dataset: ReplayDataset, spec: ScheduleSpec) -> list[ReplayStep]: + """Partition the sorted event stream into :class:`ReplayStep` batches. + + One step per cut slot. The tail after the last cut is intentionally NOT a + step (recompute fires only at cut points; include ``"end"`` to recompute + the tail). Moderation events are woven in per ``spec.moderation``: + ``"none"`` ignores them; ``"interleave-by-timestamp"`` uses the dataset's + ``mod_events``; an explicit list of ModEvent-shaped dicts overrides. Each + mod event is attached to the FIRST cut whose ``cut_time_ms`` reaches its + ``t_ms``; events after the last cut are dropped (like tail votes). + """ + slots = resolve_cut_slots(dataset, spec.cuts) + if not slots: + return [] + + mod_events = _resolve_mod_events(dataset, spec) + + steps: list[ReplayStep] = [] + prev = 0 + for i, cut in enumerate(slots): + batch = tuple(dataset.votes[prev:cut]) # 1-based (prev, cut] → 0-based slice + cut_time_ms = dataset.votes[cut - 1].t_ms + prev_time = dataset.votes[prev - 1].t_ms if prev > 0 else None + step_mods = tuple( + m + for m in mod_events + if m.t_ms <= cut_time_ms and (prev_time is None or m.t_ms > prev_time) + ) + steps.append( + ReplayStep( + index=i, + prev_slot=prev, + cut_slot=cut, + vote_events=batch, + mod_events=step_mods, + cut_time_ms=cut_time_ms, + ) + ) + prev = cut + return steps + + +def _resolve_mod_events(dataset: ReplayDataset, spec: ScheduleSpec) -> list[ModEvent]: + mode = spec.moderation + if mode == "none": + return [] + if mode == "interleave-by-timestamp": + return sorted(dataset.mod_events, key=lambda m: m.t_ms) + if isinstance(mode, (list, tuple)): + parsed = [ + m if isinstance(m, ModEvent) else ModEvent(t_ms=int(m["t_ms"]), + tid=int(m["tid"]), + mod=int(m["mod"])) + for m in mode + ] + return sorted(parsed, key=lambda m: m.t_ms) + raise ValueError(f"unknown moderation spec {mode!r}") + + +# --------------------------------------------------------------------------- +# Presets — each returns a ready-to-slice ScheduleSpec (design §4). +# --------------------------------------------------------------------------- +def _spec(dataset_name: str, schedule_id: str, cuts: dict[str, Any], + notes: str = "", moderation: Any = "none") -> ScheduleSpec: + return ScheduleSpec.from_dict( + { + "dataset": dataset_name, + "schedule_id": schedule_id, + "source": "votes-csv", + "cuts": cuts, + "moderation": moderation, + "clojure": {"warm_start": "chain"}, + "notes": notes, + } + ) + + +def preset_single_cut(dataset_name: str, n: int, *, schedule_id: str = "single-cut") -> ScheduleSpec: + """One recompute over the whole stream — equivalent to today's cold start.""" + return _spec(dataset_name, schedule_id, {"mode": "vote-count", "at": [_END]}, + notes="single cold-start recompute over all votes") + + +def preset_every_vote(dataset_name: str, n: int, *, schedule_id: str = "every-vote") -> ScheduleSpec: + """Recompute after every vote (small datasets only — n recomputes).""" + return _spec(dataset_name, schedule_id, + {"mode": "vote-count", "at": list(range(1, n + 1))}, + notes="recompute after every vote (small datasets only)") + + +def preset_uniform(dataset_name: str, n: int, n_cuts: int, *, + schedule_id: str | None = None) -> ScheduleSpec: + """``n_cuts`` evenly-spaced recomputes; the last lands on ``n``.""" + slots = _dedupe_slots([round(n * i / n_cuts) for i in range(1, n_cuts + 1)], n) + return _spec(dataset_name, schedule_id or f"uniform-{n_cuts}", + {"mode": "vote-count", "at": slots}, + notes=f"{n_cuts} evenly-spaced recomputes") + + +def preset_front_loaded(dataset_name: str, n: int, *, n_cuts: int = 6, + schedule_id: str = "front-loaded") -> ScheduleSpec: + """Recomputes concentrated EARLY (quadratic spacing, denser at the start).""" + slots = _dedupe_slots([round(n * (i / n_cuts) ** 2) for i in range(1, n_cuts + 1)], n) + return _spec(dataset_name, schedule_id, {"mode": "vote-count", "at": slots}, + notes="front-loads recomputes into the early conversation") + + +def preset_back_loaded(dataset_name: str, n: int, *, n_cuts: int = 6, + schedule_id: str = "back-loaded") -> ScheduleSpec: + """Recomputes concentrated LATE (mirror of front-loaded).""" + slots = _dedupe_slots( + [round(n * (1 - (1 - i / n_cuts) ** 2)) for i in range(1, n_cuts + 1)], n + ) + return _spec(dataset_name, schedule_id, {"mode": "vote-count", "at": slots}, + notes="back-loads recomputes into the late conversation") + + +def preset_per_day(dataset_name: str, dataset: ReplayDataset, *, + schedule_id: str = "per-day") -> ScheduleSpec: + """One recompute at each UTC-day boundary derived from real timestamps. + + Cut slots are the cumulative vote counts at the end of each day that has + votes; encoded as explicit event indices so the schedule is stable even if + the dataset is re-derived. Requires a time-sorted dataset (build() sorts). + """ + day_ms = 24 * 3600 * 1000 + slots: list[int] = [] + prev_day: int | None = None + for idx, v in enumerate(dataset.votes, start=1): + d = v.t_ms // day_ms + if prev_day is not None and d != prev_day: + slots.append(idx - 1) # last vote of the previous day + prev_day = d + if dataset.n: + slots.append(dataset.n) # close the final day + slots = _dedupe_slots(slots, dataset.n) + return _spec(dataset_name, schedule_id, + {"mode": "explicit-event-index", "at": slots}, + notes="one recompute per UTC day (from real timestamps)") + + +def _dedupe_slots(slots: list[int], n: int) -> list[int]: + """Clamp to ``1..n``, drop 0/dupes, keep sorted — as a plain JSON list.""" + return sorted({max(1, min(int(s), n)) for s in slots if s > 0}) diff --git a/delphi/polismath/replay/stepcompare.py b/delphi/polismath/replay/stepcompare.py new file mode 100644 index 000000000..6e7df8dd5 --- /dev/null +++ b/delphi/polismath/replay/stepcompare.py @@ -0,0 +1,205 @@ +"""Step comparer — replay harness Phase H-A (design §8). + +A THIN repointing layer over +:class:`polismath.regression.comparer.ConversationComparer`. That comparer +already does recursive tolerant diffing, PCA sign-flip / scaling detection, and +outlier handling on arbitrary nested ``{key: blob}`` maps — so comparing two +recordings is just: reset it, run ``_compare_dicts`` on each pair of step blobs, +and classify the surviving divergences by FIELD FAMILY. We do NOT rewrite the +comparer. + +Tolerance classes (design §8): +- **exact** — counts, in-conv, moderation sets, selections/ids. These are + integers/lists in the blob; ``_compare_dicts`` already compares them exactly + (tolerance never applies to ints), so any difference is a hard divergence. +- **tolerant** — PCA comps/proj, cluster centers (PCA-derived), silhouettes, + repness/consensus/priority stats. These are floats compared within tolerance, + with PCA sign-flips absorbed (``ignore_pca_sign_flip=True``). + +The family tag is a reporting overlay: a divergence is 'tolerant' when its path +is PCA-related OR it is a numeric mismatch under a known stat container; else +'exact'. Because the underlying comparer already applies zero tolerance to +ints and numeric tolerance to floats, this classification faithfully realises +the per-family tolerance without a second comparison pass. + +``math_tick`` (wall-clock, conversation.py:2226) is ignored by the underlying +comparer, so it never shows up as a divergence. +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any + +from polismath.regression.comparer import ConversationComparer +from polismath.replay.store import _safe_path_component, load_step_blobs + +# Numeric-stat containers whose float leaves are the 'tolerant' family. +# `consensus` (top-level) joins group-aware-consensus: both are float consensus +# stats. `silhouette` scores (e.g. within group-clusters) are handled by path in +# _family so a group-clusters ID/member mismatch stays EXACT. +DEFAULT_TOLERANT_STAT_KEYS = frozenset( + {"repness", "participant_info", "comment_priorities", "group-aware-consensus", + "consensus"} +) + +_TOP_KEY_RE = re.compile(r"^step_\d+\.([^.\[]+)") + + +class StepComparer: + """Compare two step blobs and class divergences by field family.""" + + def __init__( + self, + *, + abs_tolerance: float = 1e-6, + rel_tolerance: float = 0.01, + outlier_fraction: float = 0.01, + ignore_pca_sign_flip: bool = True, + tolerant_stat_keys: frozenset[str] = DEFAULT_TOLERANT_STAT_KEYS, + ): + # One tolerant-configured comparer; PCA sign/scale handled internally. + self._cmp = ConversationComparer( + abs_tolerance=abs_tolerance, + rel_tolerance=rel_tolerance, + ignore_pca_sign_flip=ignore_pca_sign_flip, + outlier_fraction=outlier_fraction, + ) + self._tolerant_keys = tolerant_stat_keys + + def compare_step(self, blob_a: dict, blob_b: dict, index: int) -> dict[str, Any]: + """Diff one pair of step blobs → a per-step, per-family report.""" + c = self._cmp + # Reset the comparer's accumulators for a clean per-step run. + c.all_differences = [] + c.sign_flip_warnings = [] + c.outlier_warnings = [] + c._pca_sign_flips = {} + + path = f"step_{index}" + c._compare_dicts(blob_a, blob_b, path=path, stage_name=path) + + exact: list[dict] = [] + tolerant: list[dict] = [] + for diff in c.all_differences: + entry = { + "path": diff.get("path"), + "reason": diff.get("reason"), + "a": diff.get("golden_value"), + "b": diff.get("current_value"), + } + (tolerant if self._family(diff) == "tolerant" else exact).append(entry) + + return { + "step": index, + "match": len(c.all_differences) == 0, + "n_divergences": len(c.all_differences), + "families": {"exact": exact, "tolerant": tolerant}, + "sign_flips": [ + {"path": w.get("path"), "message": w.get("message")} + for w in c.sign_flip_warnings + ], + } + + def _family(self, diff: dict) -> str: + path = diff.get("path", "") or "" + reason = diff.get("reason", "") or "" + if self._cmp._is_pca_related_path(path): + return "tolerant" + # Silhouette scores are float cluster-quality stats (e.g. the + # group-clusters silhouette) — tolerant, not a hard structural mismatch. + # Keyed on the path (not the top key) so a group-clusters ID/member + # divergence stays EXACT. + if reason.startswith("Numeric mismatch") and "silhouette" in path: + return "tolerant" + m = _TOP_KEY_RE.match(path) + top = m.group(1) if m else "" + if top in self._tolerant_keys and reason.startswith("Numeric mismatch"): + return "tolerant" + return "exact" + + +def compare_recordings( + dir_a: str | Path, + dir_b: str | Path, + *, + engine: str = "py", + comparer: StepComparer | None = None, +) -> dict[str, Any]: + """Compare two recordings step-by-step (design §8). + + Aligns ``py/step-NNN.json`` blobs by index. A step-count mismatch is + reported (never silently truncated): only the overlapping prefix is + diffed, and ``overall_match`` is False whenever counts differ or any + aligned step diverges. + """ + dir_a, dir_b = Path(dir_a), Path(dir_b) + # engine joins the paths as a single component — reject traversal values. + engine = _safe_path_component(engine, label="engine") + blobs_a = load_step_blobs(dir_a / engine) + blobs_b = load_step_blobs(dir_b / engine) + cmp = comparer or StepComparer() + + aligned = min(len(blobs_a), len(blobs_b)) + per_step = [cmp.compare_step(blobs_a[i], blobs_b[i], i) for i in range(aligned)] + + count_mismatch = len(blobs_a) != len(blobs_b) + steps_match = all(s["match"] for s in per_step) + + total_exact = sum(len(s["families"]["exact"]) for s in per_step) + total_tolerant = sum(len(s["families"]["tolerant"]) for s in per_step) + + return { + "recording_a": str(dir_a), + "recording_b": str(dir_b), + "engine": engine, + "n_steps_a": len(blobs_a), + "n_steps_b": len(blobs_b), + "aligned_steps": aligned, + "step_count_mismatch": count_mismatch, + "overall_match": steps_match and not count_mismatch, + "summary": { + "diverging_steps": sum(1 for s in per_step if not s["match"]), + "total_exact_divergences": total_exact, + "total_tolerant_divergences": total_tolerant, + }, + "per_step": per_step, + } + + +def write_report(report: dict[str, Any], path: str | Path) -> None: + Path(path).parent.mkdir(parents=True, exist_ok=True) + with open(path, "w") as fh: + json.dump(report, fh, indent=2, default=str) + + +def format_report(report: dict[str, Any]) -> str: + """Render a compact human summary of a compare_recordings report.""" + lines: list[str] = [] + verdict = "MATCH" if report["overall_match"] else "DIVERGENCE" + lines.append(f"Replay comparison: {verdict}") + lines.append(f" A: {report['recording_a']} ({report['n_steps_a']} steps)") + lines.append(f" B: {report['recording_b']} ({report['n_steps_b']} steps)") + if report["step_count_mismatch"]: + lines.append( + f" ! step-count mismatch — comparing first {report['aligned_steps']}" + ) + s = report["summary"] + lines.append( + f" diverging steps: {s['diverging_steps']}/{report['aligned_steps']}" + f" (exact={s['total_exact_divergences']}," + f" tolerant={s['total_tolerant_divergences']})" + ) + for step in report["per_step"]: + if step["match"]: + continue + ex = len(step["families"]["exact"]) + tol = len(step["families"]["tolerant"]) + lines.append(f" step {step['step']}: exact={ex} tolerant={tol}") + for d in step["families"]["exact"][:3]: + lines.append(f" [exact] {d['path']}: {d['reason']}") + for d in step["families"]["tolerant"][:3]: + lines.append(f" [tolerant] {d['path']}: {d['reason']}") + return "\n".join(lines) diff --git a/delphi/polismath/replay/store.py b/delphi/polismath/replay/store.py new file mode 100644 index 000000000..0bd537a51 --- /dev/null +++ b/delphi/polismath/replay/store.py @@ -0,0 +1,286 @@ +"""Recording store + provenance — replay harness Phase H-A (design §7). + +Layout (one directory per (dataset, schedule)):: + + real_data/.local/replays/// + schedule.json # the §4 input, VERBATIM + provenance.json # delphi commit, dataset sha256, versions, flags + py/step-000.json # {index, cut_slot, …, blob: to_dict, extras} + py/step-001.json + … + +Everything lives under ``real_data/.local/`` which is gitignored +(``delphi/.gitignore:219``; verified with ``git check-ignore``), so replays of +private datasets never leak into the repo — and neither do the absolute paths a +provenance file may contain. Directory creation is lazy: :func:`recording_dir` +only computes a path; :func:`write_recording` creates it. + +Provenance satisfies the reproducible-traces requirement: a replay is +re-derivable from (schedule.json, dataset file, delphi commit). We additionally +pin the runtime that MATTERS for the numbers — the vote-sign convention +(``delphi``; the future Clojure driver needs raw-DB/flipped signs), the +``POLISMATH_PCA_IMPL`` engine flag, and numpy/sklearn/pandas versions. +""" + +from __future__ import annotations + +import hashlib +import json +import os +import platform +import subprocess +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from polismath.replay.driver import VOTE_SIGN_CONVENTION, StepRecord +from polismath.replay.schedule import ScheduleSpec + +# Default engine tag for the store subdirectory (design reserves clj/ for H-B). +PY_ENGINE = "py" + + +def replays_root() -> Path: + """Default store root: ``delphi/real_data/.local/replays`` (gitignored).""" + # store.py → replay → polismath → delphi + delphi = Path(__file__).resolve().parents[2] + return delphi / "real_data" / ".local" / "replays" + + +def _safe_path_component(name: str, *, label: str) -> str: + """Reject a path component that could escape the store root (path traversal). + + ``dataset`` / ``schedule_id`` flow straight into the on-disk path, so a value + like ``..`` or ``/etc`` (or one containing a separator) would write OUTSIDE + ``replays_root()``. Slugs are simple identifiers — reject anything else. + """ + if ( + not isinstance(name, str) + or not name + or name in (".", "..") + or os.path.isabs(name) + or "/" in name + or "\\" in name + or os.sep in name + or (os.altsep and os.altsep in name) + ): + raise ValueError(f"unsafe {label} for recording path: {name!r}") + return name + + +def recording_dir(dataset: str, schedule_id: str, *, root: Path | None = None) -> Path: + """Compute (do NOT create) the recording directory for (dataset, schedule).""" + dataset = _safe_path_component(dataset, label="dataset") + schedule_id = _safe_path_component(schedule_id, label="schedule_id") + return (root or replays_root()) / dataset / schedule_id + + +def write_recording( + records: list[StepRecord], + spec: ScheduleSpec, + *, + root: Path | None = None, + engine: str = PY_ENGINE, + extra_provenance: dict[str, Any] | None = None, +) -> Path: + """Write schedule.json (verbatim), provenance.json and per-step blobs. + + Returns the recording directory. Steps are written to ``/step-NNN`` + (``py/`` for the Python driver); the H-B Clojure driver will populate + ``clj/`` under the same layout. + """ + out = recording_dir(spec.dataset, spec.schedule_id, root=root) + engine = _safe_path_component(engine, label="engine") + step_dir = out / engine + step_dir.mkdir(parents=True, exist_ok=True) # lazy: created only on write + + # Clear any step-*.json left by a PRIOR recording of this (dataset, schedule, + # engine) before writing the fresh set. Loaders glob EVERY step-*.json (see + # _load_step_payloads) and preset schedule_ids don't encode n_cuts, so a + # re-run producing FEWER steps would otherwise leave stale higher-index files + # that silently mix into the loaded recording. This makes each write the + # authoritative step set. + for stale in step_dir.glob("step-*.json"): + stale.unlink() + + spec.write_json(out / "schedule.json") + + prov = build_provenance(spec, records, engine=engine, extra=extra_provenance) + _write_json(out / "provenance.json", prov) + + for r in records: + _write_json(step_dir / f"step-{r.index:03d}.json", _step_payload(r)) + + return out + + +def _step_payload(r: StepRecord) -> dict[str, Any]: + return { + "index": r.index, + "prev_slot": r.prev_slot, + "cut_slot": r.cut_slot, + "batch_size": r.batch_size, + "cut_time_ms": r.cut_time_ms, + "blob": r.blob, + "extras": r.extras, + } + + +# --------------------------------------------------------------------------- +# Provenance. +# --------------------------------------------------------------------------- +def build_provenance( + spec: ScheduleSpec, + records: list[StepRecord], + *, + engine: str = PY_ENGINE, + extra: dict[str, Any] | None = None, +) -> dict[str, Any]: + """Assemble the provenance record (design §7).""" + prov: dict[str, Any] = { + "schedule_id": spec.schedule_id, + "source": spec.source, + "engine": engine, + "n_steps": len(records), + "created_at": datetime.now(timezone.utc).isoformat(), + "delphi_git_commit": _git_commit(), + "python_version": platform.python_version(), + "packages": _package_versions(), + # Design §5 / D1b: the export CSVs are already in Delphi convention + # (AGREE=+1); the Python engine consumes them AS-IS. The future Clojure + # driver (H-B) must feed raw-DB signs (flipped, export.clj:106-113). + "vote_sign_convention": VOTE_SIGN_CONVENTION, + "vote_sign_note": ( + "export CSVs already Delphi convention (AGREE=+1); Clojure driver " + "needs raw-DB/flipped signs" + ), + # Engine impl flags that change the numbers (design §7). + "engine_flags": { + "POLISMATH_PCA_IMPL": os.environ.get("POLISMATH_PCA_IMPL", "powerit"), + # The engine mode changes warm-start behavior across steps — two + # recordings of the same schedule under different modes are + # materially different trajectories. + "POLISMATH_ENGINE_MODE": os.environ.get( + "POLISMATH_ENGINE_MODE", "improved"), + "OMP_NUM_THREADS": os.environ.get("OMP_NUM_THREADS"), + "OPENBLAS_NUM_THREADS": os.environ.get("OPENBLAS_NUM_THREADS"), + }, + "dataset": _dataset_provenance(spec.dataset), + } + if extra: + prov.update(extra) + return prov + + +def _dataset_provenance(dataset_slug: str) -> dict[str, Any]: + """Dataset name + votes/comments basenames and sha256 (best-effort).""" + info: dict[str, Any] = {"name": dataset_slug} + try: + # Reuse regression dataset discovery (report_id-based) to locate files. + from polismath.regression.datasets import get_dataset_files + + files = get_dataset_files(dataset_slug) + except Exception as exc: # dataset not locatable (e.g. not on this checkout) + info["resolution_error"] = str(exc) + return info + votes = files.get("votes") + comments = files.get("comments") + if votes: + info["votes_file"] = Path(votes).name + info["votes_sha256"] = _sha256(votes) + if comments: + info["comments_file"] = Path(comments).name + info["comments_sha256"] = _sha256(comments) + return info + + +def _sha256(path: str | Path) -> str | None: + h = hashlib.sha256() + try: + with open(path, "rb") as fh: + for chunk in iter(lambda: fh.read(65536), b""): + h.update(chunk) + except OSError: + return None + return h.hexdigest() + + +def _git_commit() -> str: + repo = Path(__file__).resolve().parents[3] # replay→polismath→delphi→repo + try: + out = subprocess.run( + ["git", "-C", str(repo), "rev-parse", "HEAD"], + capture_output=True, text=True, timeout=10, + ) + if out.returncode == 0: + return out.stdout.strip() + except (OSError, subprocess.SubprocessError): + pass + return "unknown" + + +def _package_versions() -> dict[str, str | None]: + versions: dict[str, str | None] = {} + for name in ("numpy", "pandas", "scipy", "sklearn"): + try: + mod = __import__(name) + versions[name] = getattr(mod, "__version__", None) + except Exception: + versions[name] = None + return versions + + +# --------------------------------------------------------------------------- +# Numpy-aware JSON. +# --------------------------------------------------------------------------- +def _json_default(obj: Any) -> Any: + """Encode numpy scalars/arrays (mirrors utils.save_golden_snapshot).""" + import numpy as np + + if isinstance(obj, np.integer): + return int(obj) + if isinstance(obj, np.floating): + return float(obj) + if isinstance(obj, np.ndarray): + return obj.tolist() + if isinstance(obj, (set, frozenset)): + return sorted(obj) + raise TypeError(f"Object of type {type(obj).__name__} is not JSON serializable") + + +def _write_json(path: Path, data: Any) -> None: + with open(path, "w") as fh: + json.dump(data, fh, indent=2, default=_json_default) + + +# --------------------------------------------------------------------------- +# Load side. +# --------------------------------------------------------------------------- +@dataclass +class Recording: + """A loaded recording: verbatim schedule, provenance, and ordered steps.""" + + path: Path + schedule: dict[str, Any] + provenance: dict[str, Any] + steps: list[dict[str, Any]] # full per-step payloads (blob + extras + meta) + + +def load_recording(path: str | Path, *, engine: str = PY_ENGINE) -> Recording: + path = Path(path) + schedule = json.loads((path / "schedule.json").read_text()) + prov_file = path / "provenance.json" + provenance = json.loads(prov_file.read_text()) if prov_file.exists() else {} + steps = _load_step_payloads(path / engine) + return Recording(path=path, schedule=schedule, provenance=provenance, steps=steps) + + +def _load_step_payloads(step_dir: Path) -> list[dict[str, Any]]: + files = sorted(step_dir.glob("step-*.json")) + return [json.loads(f.read_text()) for f in files] + + +def load_step_blobs(step_dir: str | Path) -> list[dict[str, Any]]: + """Return just the to_dict blobs, in step order — the comparison surface.""" + return [p["blob"] for p in _load_step_payloads(Path(step_dir))] diff --git a/delphi/polismath/replay/types.py b/delphi/polismath/replay/types.py new file mode 100644 index 000000000..3f85cc277 --- /dev/null +++ b/delphi/polismath/replay/types.py @@ -0,0 +1,154 @@ +"""Core event and dataset types for replay schedule inference. + +Normative conventions (docs/plans/2026-07-06-r2-schedule-inference.md): + +- Votes are sorted by ``(t_ms, input order)`` and 1-indexed by ``k``. +- A *revote* is a later occurrence of an already-seen ``(pid, tid)`` pair in + sorted order. Revotes are excluded from the mark likelihood (they cannot be + serve-generated) but still update prefix statistics (latest-vote-wins). +- A *cut slot* ``i`` in ``1..n`` means "a recompute fired after ingesting + votes ``1..i``". A :data:`Schedule` is a strictly increasing tuple of slots. +- ``segments(schedule)`` partitions the vote index range into half-open + segments ``(left, right]`` where ``left`` is the previous cut slot (sentinel + ``-1`` before any cut — production serves with all-default weights until the + first recompute lands) and the final segment runs to ``n`` (the tail after + the last cut; possibly empty). +""" + +from dataclasses import dataclass, field +from enum import IntEnum + + +class Vote(IntEnum): + """Semantic vote signs. + + These are *internal* semantics. Adapters that ingest external data own + the mapping: the polis DB stores agree as -1, and export CSVs flip signs + relative to the DB (see math/src/polismath/darwin/export.clj:106-113). + """ + + AGREE = 1 + DISAGREE = -1 + PASS = 0 + + +@dataclass(frozen=True) +class VoteEvent: + """One vote, in sorted order. ``k`` is its 1-based index.""" + + k: int + t_ms: int + pid: int + tid: int + sign: int + is_revote: bool + + +@dataclass(frozen=True) +class CommentMeta: + """Static comment metadata relevant to routing.""" + + tid: int + created_ms: int + is_meta: bool = False + + +@dataclass(frozen=True) +class ModEvent: + """A moderation change at ``t_ms`` setting ``comments.mod`` for ``tid``. + + ``mod`` uses the production convention: -1 moderated-out, 0 unmoderated, + 1 moderated-in. + """ + + t_ms: int + tid: int + mod: int + + +Schedule = tuple[int, ...] +"""Strictly increasing tuple of cut slots in ``1..n``.""" + + +@dataclass +class ReplayDataset: + """A conversation's event stream, prepared for schedule inference.""" + + votes: list[VoteEvent] + comments: dict[int, CommentMeta] + mod_events: list[ModEvent] = field(default_factory=list) + strict_moderation: bool = False + + @property + def n(self) -> int: + return len(self.votes) + + @classmethod + def build( + cls, + raw_votes: list[tuple[int, int, int, int]], + comments: dict[int, CommentMeta] | None = None, + mod_events: list[ModEvent] | tuple[ModEvent, ...] = (), + strict_moderation: bool = False, + ) -> "ReplayDataset": + """Build a dataset from raw ``(t_ms, pid, tid, sign)`` rows. + + Sorts votes stably by time, assigns 1-based ``k``, flags revotes. + When ``comments`` is None, each comment's creation time is inferred + as its first (sorted) vote time — a lower bound adequate for + availability modelling when the comments table is absent. + """ + indexed = sorted(enumerate(raw_votes), key=lambda p: (p[1][0], p[0])) + votes: list[VoteEvent] = [] + seen: set[tuple[int, int]] = set() + first_vote_ms: dict[int, int] = {} + for k, (_, (t_ms, pid, tid, sign)) in enumerate(indexed, start=1): + pair = (pid, tid) + votes.append( + VoteEvent( + k=k, + t_ms=t_ms, + pid=pid, + tid=tid, + sign=sign, + is_revote=pair in seen, + ) + ) + seen.add(pair) + first_vote_ms.setdefault(tid, t_ms) + + if comments is None: + comments = { + tid: CommentMeta(tid=tid, created_ms=t) + for tid, t in first_vote_ms.items() + } + else: + missing = sorted(set(first_vote_ms) - set(comments)) + if missing: + raise ValueError( + f"comments table missing voted tid {missing[0]}" + + (f" (+{len(missing) - 1} more)" if len(missing) > 1 else "") + ) + + return cls( + votes=votes, + comments=dict(comments), + mod_events=sorted(mod_events, key=lambda m: m.t_ms), + strict_moderation=strict_moderation, + ) + + def validate_schedule(self, schedule: Schedule) -> None: + prev = 0 + for s in schedule: + if not 1 <= s <= self.n: + raise ValueError(f"cut slot {s} outside 1..{self.n}") + if s <= prev: + raise ValueError(f"schedule not strictly increasing at slot {s}") + prev = s + + def segments(self, schedule: Schedule) -> list[tuple[int, int]]: + """Partition vote indices into ``(left, right]`` scoring segments.""" + self.validate_schedule(schedule) + lefts = [-1, *schedule] + rights = [*schedule, self.n] + return list(zip(lefts, rights)) diff --git a/delphi/scripts/replay_driver.py b/delphi/scripts/replay_driver.py new file mode 100644 index 000000000..cbadaffce --- /dev/null +++ b/delphi/scripts/replay_driver.py @@ -0,0 +1,146 @@ +#!/usr/bin/env python3 +"""Replay-harness CLI (Phase H-A) — run a (dataset, schedule) replay and +compare two recordings. + +Runs the Python driver end-to-end and writes a recording store +(``real_data/.local/replays///`` by default), and offers +a compare entry point for two recordings. The Clojure driver (H-B) and +cross-language comparison land later; this CLI already exercises the full +Python spine. + +Usage (from delphi/):: + + # Run a preset schedule on the public vw dataset: + uv run python scripts/replay_driver.py run --dataset vw --preset front-loaded + uv run python scripts/replay_driver.py run --dataset vw --preset uniform --n-cuts 8 + uv run python scripts/replay_driver.py run --dataset vw --preset per-day + + # Run an explicit schedule JSON (§4): + uv run python scripts/replay_driver.py run --schedule my_schedule.json + + # Compare two recordings step-by-step: + uv run python scripts/replay_driver.py compare --report out.json +""" + +from __future__ import annotations + +import logging +import sys +from pathlib import Path + +import click + +from polismath.replay import schedule as sched +from polismath.replay import store as st +from polismath.replay import stepcompare as sc +from polismath.replay.driver import run_replay +from polismath.replay.real_data import load_export_votes +from polismath.replay.types import ReplayDataset + +_PRESETS = ( + "uniform", "front-loaded", "back-loaded", "every-vote", "single-cut", "per-day" +) + + +def _spec_from_preset( + preset: str, dataset: str, ds: ReplayDataset, *, n_cuts: int, + schedule_id: str | None, +) -> sched.ScheduleSpec: + n = ds.n + if preset == "uniform": + return sched.preset_uniform(dataset, n, n_cuts=n_cuts, schedule_id=schedule_id) + if preset == "front-loaded": + return sched.preset_front_loaded( + dataset, n, n_cuts=n_cuts, schedule_id=schedule_id or "front-loaded") + if preset == "back-loaded": + return sched.preset_back_loaded( + dataset, n, n_cuts=n_cuts, schedule_id=schedule_id or "back-loaded") + if preset == "every-vote": + return sched.preset_every_vote(dataset, n, schedule_id=schedule_id or "every-vote") + if preset == "single-cut": + return sched.preset_single_cut(dataset, n, schedule_id=schedule_id or "single-cut") + if preset == "per-day": + return sched.preset_per_day(dataset, ds, schedule_id=schedule_id or "per-day") + raise click.BadParameter(f"unknown preset {preset!r}") + + +@click.group() +def cli() -> None: + """Replay-harness driver + comparer.""" + + +@cli.command() +@click.option("--dataset", help="Dataset slug (e.g. vw). Required unless --schedule sets it.") +@click.option("--schedule", "schedule_path", type=click.Path(exists=True, path_type=Path), + help="Path to a schedule.json (§4). Overrides --preset.") +@click.option("--preset", type=click.Choice(_PRESETS), default=None, + help="Built-in schedule preset.") +@click.option("--n-cuts", type=int, default=6, show_default=True, + help="Number of cuts for uniform/front/back presets.") +@click.option("--schedule-id", default=None, help="Override the schedule id.") +@click.option("--out", "out_root", type=click.Path(path_type=Path), default=None, + help="Store root (default: real_data/.local/replays).") +@click.option("--verbose", is_flag=True, help="Show driver progress logging.") +def run(dataset, schedule_path, preset, n_cuts, schedule_id, out_root, verbose): + """Run a (dataset, schedule) replay and write the recording store.""" + if not verbose: + # logging.disable is process-global: restore it in _run's finally so an + # in-process caller (CliRunner tests) isn't silenced past this command. + logging.disable(logging.CRITICAL) + try: + _run_impl(dataset, schedule_path, preset, n_cuts, schedule_id, out_root, verbose) + finally: + logging.disable(logging.NOTSET) + + +def _run_impl(dataset, schedule_path, preset, n_cuts, schedule_id, out_root, verbose): + ds: ReplayDataset | None = None + loaded_slug: str | None = None + if schedule_path is not None: + spec = sched.ScheduleSpec.from_json_file(schedule_path) + dataset = dataset or spec.dataset + elif preset is not None: + if not dataset: + raise click.UsageError("--dataset is required with --preset") + ds = load_export_votes(dataset) + loaded_slug = dataset + spec = _spec_from_preset(preset, dataset, ds, n_cuts=n_cuts, + schedule_id=schedule_id) + else: + raise click.UsageError("provide either --schedule or --preset") + + # Reuse the dataset already loaded to build a preset spec instead of loading + # it a second time; only the --schedule path (or a slug mismatch) needs a load. + if ds is None or loaded_slug != spec.dataset: + ds = load_export_votes(spec.dataset) + click.echo(f"dataset={spec.dataset} n_votes={ds.n} schedule={spec.schedule_id}", err=True) + + def _progress(i: int, total: int) -> None: + click.echo(f" step {i + 1}/{total} …", err=True) + + records = run_replay(ds, spec, progress=_progress if verbose else None) + out_dir = st.write_recording(records, spec, root=out_root) + click.echo(f"wrote {len(records)} steps → {out_dir}") + + +@cli.command() +@click.argument("dir_a", type=click.Path(exists=True, path_type=Path)) +@click.argument("dir_b", type=click.Path(exists=True, path_type=Path)) +@click.option("--engine", default="py", show_default=True) +@click.option("--report", "report_path", type=click.Path(path_type=Path), default=None, + help="Write the full per-step JSON report here.") +@click.option("--abs-tol", type=float, default=1e-6, show_default=True) +@click.option("--rel-tol", type=float, default=0.01, show_default=True) +def compare(dir_a, dir_b, engine, report_path, abs_tol, rel_tol): + """Compare two recordings step-by-step and print a divergence summary.""" + comparer = sc.StepComparer(abs_tolerance=abs_tol, rel_tolerance=rel_tol) + report = sc.compare_recordings(dir_a, dir_b, engine=engine, comparer=comparer) + click.echo(sc.format_report(report)) + if report_path is not None: + sc.write_report(report, report_path) + click.echo(f"report → {report_path}", err=True) + sys.exit(0 if report["overall_match"] else 1) + + +if __name__ == "__main__": + cli() diff --git a/delphi/tests/replay_harness/__init__.py b/delphi/tests/replay_harness/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/delphi/tests/replay_harness/test_cli.py b/delphi/tests/replay_harness/test_cli.py new file mode 100644 index 000000000..5394c0294 --- /dev/null +++ b/delphi/tests/replay_harness/test_cli.py @@ -0,0 +1,135 @@ +"""End-to-end CLI smoke test for scripts/replay_driver.py (Phase H-A). + +Drives the real CLI (via click's runner) on the public vw dataset with a tiny +explicit schedule so it stays fast, then compares the recording against itself +(→ match, exit 0). This exercises the whole spine: load → slice → drive → +store → compare. +""" + +import importlib.util +import json +from pathlib import Path + +import pytest +from click.testing import CliRunner + +_CLI_PATH = Path(__file__).resolve().parents[2] / "scripts" / "replay_driver.py" + + +def _module(): + spec = importlib.util.spec_from_file_location("replay_driver_cli", _CLI_PATH) + mod = importlib.util.module_from_spec(spec) + spec.loader.exec_module(mod) + return mod + + +def _load_cli(): + return _module().cli + + +@pytest.fixture(scope="module") +def cli(): + return _load_cli() + + +def _write_schedule(tmp_path): + # Small vote-count cuts (no "end") → fast prefix recomputes only. + d = { + "dataset": "vw", + "schedule_id": "cli-smoke", + "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [300, 700]}, + "moderation": "none", + "clojure": {"warm_start": "chain"}, + "notes": "cli smoke", + } + p = tmp_path / "sched.json" + p.write_text(json.dumps(d)) + return p + + +def test_run_then_compare_self(cli, tmp_path): + runner = CliRunner() + sched_path = _write_schedule(tmp_path) + store_root = tmp_path / "store" + + res = runner.invoke( + cli, ["run", "--schedule", str(sched_path), "--out", str(store_root)] + ) + assert res.exit_code == 0, res.output + rec_dir = store_root / "vw" / "cli-smoke" + assert (rec_dir / "schedule.json").exists() + assert (rec_dir / "provenance.json").exists() + assert sorted((rec_dir / "py").glob("step-*.json")) + + # schedule.json is verbatim. + assert json.loads((rec_dir / "schedule.json").read_text())["schedule_id"] == "cli-smoke" + + # Compare the recording against itself → match, exit 0. + report_path = tmp_path / "report.json" + res2 = runner.invoke( + cli, ["compare", str(rec_dir), str(rec_dir), "--report", str(report_path)] + ) + assert res2.exit_code == 0, res2.output + assert "MATCH" in res2.output + report = json.loads(report_path.read_text()) + assert report["overall_match"] is True + assert report["aligned_steps"] == 2 + + +def test_run_requires_schedule_or_preset(cli): + runner = CliRunner() + res = runner.invoke(cli, ["run", "--dataset", "vw"]) + assert res.exit_code != 0 + assert "either --schedule or --preset" in res.output + + +def test_run_preset_loads_dataset_once(tmp_path, monkeypatch): + """P6h: the --preset path must not load the dataset twice (once to build the + preset spec, once to run) — reuse the already-loaded dataset.""" + from polismath.replay.types import ReplayDataset + + mod = _module() + ds = ReplayDataset.build([(10, 0, 1, 1), (11, 1, 1, -1), (12, 0, 2, 1), (13, 1, 2, -1)]) + calls = {"n": 0} + + def _counting_load(slug): + calls["n"] += 1 + return ds + + monkeypatch.setattr(mod, "load_export_votes", _counting_load) + monkeypatch.setattr(mod, "run_replay", lambda ds, spec, progress=None: []) + # write_recording is accessed as st.write_recording in the CLI. + monkeypatch.setattr(mod.st, "write_recording", lambda records, spec, root=None: tmp_path) + + res = CliRunner().invoke( + mod.cli, + ["run", "--dataset", "vw", "--preset", "single-cut", "--out", str(tmp_path)], + ) + assert res.exit_code == 0, res.output + assert calls["n"] == 1, f"dataset loaded {calls['n']} times, expected 1" + + +def test_run_restores_global_logging(tmp_path, monkeypatch): + """A non-verbose `run` silences logging with logging.disable — it must + restore it on exit, or one CLI call permanently mutes logging for the rest + of the process (notably in-process CliRunner test suites).""" + import logging + + from polismath.replay.types import ReplayDataset + + mod = _module() + ds = ReplayDataset.build([(10, 0, 1, 1), (11, 1, 1, -1), (12, 0, 2, 1), (13, 1, 2, -1)]) + monkeypatch.setattr(mod, "load_export_votes", lambda slug: ds) + monkeypatch.setattr(mod, "run_replay", lambda ds, spec, progress=None: []) + monkeypatch.setattr(mod.st, "write_recording", lambda records, spec, root=None: tmp_path) + + assert logging.root.manager.disable == logging.NOTSET + res = CliRunner().invoke( + mod.cli, + ["run", "--dataset", "vw", "--preset", "single-cut", "--out", str(tmp_path)], + ) + assert res.exit_code == 0, res.output + assert logging.root.manager.disable == logging.NOTSET, ( + "logging.disable level leaked past the CLI invocation" + ) diff --git a/delphi/tests/replay_harness/test_driver.py b/delphi/tests/replay_harness/test_driver.py new file mode 100644 index 000000000..ab39336be --- /dev/null +++ b/delphi/tests/replay_harness/test_driver.py @@ -0,0 +1,196 @@ +"""Python replay-driver tests (Phase H-A) — runs on the public ``vw`` dataset. + +Covers: a small 3-cut replay end-to-end; monotonic growth of counts across +steps; blob + diagnostic-extras completeness; and DETERMINISM (same schedule +twice → bit-identical step blobs except the wall-clock ``math_tick``, empirically +the ONLY nondeterministic field — see the module note below). + +Nondeterminism (verified 2026-07-18 by running the driver twice and diffing all +blob paths): the sole differing field is ``math_tick`` (conversation.py:2226, +``25000 + (int(time.time()) % 10000)``). Everything else — PCA (fixed-seed +power iteration), k-means (``random_state=42``), all counts/ids, and every +timestamp (data-derived, seeded from the first vote) — is bit-identical. +""" + +import logging + +import pytest + +from polismath.replay.real_data import load_export_votes +from polismath.replay import schedule as sched +from polismath.replay.driver import run_replay, VOTE_SIGN_CONVENTION +from polismath.replay.types import ModEvent, ReplayDataset + +# Blob fields that are wall-clock dependent and therefore excluded from the +# determinism assertion. Discovered empirically (see module docstring). +WALL_CLOCK_FIELDS = {"math_tick"} + +# A small, fast 3-cut schedule on vw. +_CUTS = {"mode": "fraction", "at": [0.34, 0.67, 1.0]} + + +@pytest.fixture(scope="module") +def vw_dataset(): + return load_export_votes("vw") + + +@pytest.fixture(scope="module") +def spec(): + return sched.ScheduleSpec.from_dict( + { + "dataset": "vw", + "schedule_id": "harness-3cut", + "source": "votes-csv", + "cuts": _CUTS, + "moderation": "none", + "clojure": {"warm_start": "chain"}, + "notes": "H-A driver smoke", + } + ) + + +@pytest.fixture(scope="module") +def run1(vw_dataset, spec): + logging.disable(logging.CRITICAL) + try: + return run_replay(vw_dataset, spec) + finally: + logging.disable(logging.NOTSET) + + +def test_produces_one_record_per_cut(run1, vw_dataset): + expected_slots = list(sched.resolve_cut_slots(vw_dataset, _CUTS)) + assert [r.cut_slot for r in run1] == expected_slots + assert len(run1) == 3 + assert [r.index for r in run1] == [0, 1, 2] + # Batches partition the covered prefix with no gaps. + prev = 0 + for r in run1: + assert r.prev_slot == prev + assert r.batch_size == r.cut_slot - r.prev_slot + prev = r.cut_slot + + +def test_counts_grow_monotonically(run1): + p = [r.extras["n_participants"] for r in run1] + c = [r.extras["n_comments"] for r in run1] + v = [r.extras["n_votes"] for r in run1] + assert p == sorted(p) and c == sorted(c) and v == sorted(v) + assert v[0] < v[-1], "later steps ingest more votes" + + +def test_final_step_ingests_all_distinct_pairs(run1, vw_dataset): + # Final step covers all votes; n_votes counts filled (pid,tid) cells + # (revotes overwrite, so == number of distinct voted pairs). + distinct_pairs = len({(x.pid, x.tid) for x in vw_dataset.votes}) + assert run1[-1].cut_slot == vw_dataset.n + assert run1[-1].extras["n_votes"] == distinct_pairs + + +def test_blob_has_expected_fields(run1): + blob = run1[-1].blob + for key in [ + "pca", "proj", "base-clusters", "group-clusters", "repness", + "in-conv", "tids", "n", "n-cmts", "user-vote-counts", "votes-base", + "group-votes", "comment_priorities", "math_tick", "zid", + ]: + assert key in blob, f"missing blob field {key!r}" + assert blob["pca"]["comps"], "PCA components should be present on vw" + + +def test_extras_complete(run1): + ex = run1[-1].extras + for key in [ + "n_participants", "n_comments", "n_votes", "n_base_clusters", + "n_group_clusters", "n_in_conv", "n_mod_out", "pca_present", + ]: + assert key in ex + assert ex["pca_present"] is True + assert ex["n_base_clusters"] > 0 + assert ex["n_group_clusters"] >= 2, "vw resolves into multiple groups" + + +def test_sign_convention_is_delphi(): + assert VOTE_SIGN_CONVENTION == "delphi" + + +# --- T5: moderation-clear seam guard -------------------------------------- +# update_moderation replaces mod_out/mod_in only when the incoming list is +# truthy, so an empty list cannot clear a previously-applied set. The driver +# must DETECT an emptying transition and fail loudly rather than silently record +# a stale (still-moderated) state. +_MOD_RAW_VOTES = [ + (10, 0, 100, 1), (20, 1, 100, -1), (30, 0, 101, 1), + (40, 1, 101, -1), (50, 2, 100, 1), (60, 2, 101, -1), +] +_MOD_CUTS = {"mode": "vote-count", "at": [4, 6]} + + +def _mod_spec(mod_events): + return sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "t5-clear", "source": "votes-csv", + "cuts": _MOD_CUTS, "moderation": "interleave-by-timestamp", + "clojure": {"warm_start": "chain"}, "notes": "t5 moderation-clear guard", + }) + + +def test_driver_fails_loudly_on_moderation_set_emptying(): + # Step 0 moderates tid 100 OUT and tid 101 IN (both sets non-empty). Step 1 + # un-moderates tid 100 (mod=0) -> mod_out_tids goes empty while mod_in stays + # active: an emptying transition update_moderation cannot represent. + mods = [ModEvent(35, 100, -1), ModEvent(35, 101, 1), ModEvent(55, 100, 0)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + logging.disable(logging.CRITICAL) + try: + with pytest.raises(NotImplementedError, match="clearing mod_out_tids"): + run_replay(ds, _mod_spec(mods)) + finally: + logging.disable(logging.NOTSET) + + +def test_driver_allows_non_emptying_moderation_sequence(): + # tid 100 OUT at t1, tid 101 IN at t2: both sets stay non-empty across steps, + # so the guard must NOT fire and the replay records both steps. + mods = [ModEvent(35, 100, -1), ModEvent(55, 101, 1)] + ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) + logging.disable(logging.CRITICAL) + try: + records = run_replay(ds, _mod_spec(mods)) + finally: + logging.disable(logging.NOTSET) + assert len(records) == 2 + + +def test_first_vote_at_tms_zero_stays_deterministic(): + """P6c: a first vote at t_ms==0 must not seed last_updated=0 (which the + `last_updated or now` footgun turns into wall-clock, breaking determinism).""" + raw = [(0, 0, 100, 1), (1, 1, 100, -1), (2, 0, 101, 1), (3, 1, 101, -1)] + ds = ReplayDataset.build(raw) + spec = sched.ScheduleSpec.from_dict({ + "dataset": "vw", "schedule_id": "tms0", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [4]}, "moderation": "none", + "clojure": {"warm_start": "chain"}, "notes": "t_ms==0 seed guard", + }) + logging.disable(logging.CRITICAL) + try: + records = run_replay(ds, spec) + finally: + logging.disable(logging.NOTSET) + lvt = records[-1].blob["lastVoteTimestamp"] + # Data-derived (== cut_time_ms == 3), NOT a wall-clock timestamp (~1e12). + assert lvt == records[-1].cut_time_ms == 3 + + +def test_determinism_bit_identical_except_wall_clock(vw_dataset, spec, run1): + logging.disable(logging.CRITICAL) + try: + run2 = run_replay(vw_dataset, spec) + finally: + logging.disable(logging.NOTSET) + + assert len(run2) == len(run1) + for a, b in zip(run1, run2): + assert a.extras == b.extras, f"extras differ at step {a.index}" + blob_a = {k: v for k, v in a.blob.items() if k not in WALL_CLOCK_FIELDS} + blob_b = {k: v for k, v in b.blob.items() if k not in WALL_CLOCK_FIELDS} + assert blob_a == blob_b, f"non-wall-clock blob differs at step {a.index}" diff --git a/delphi/tests/replay_harness/test_schedule.py b/delphi/tests/replay_harness/test_schedule.py new file mode 100644 index 000000000..dc35a4304 --- /dev/null +++ b/delphi/tests/replay_harness/test_schedule.py @@ -0,0 +1,287 @@ +"""Unit tests for the replay-harness schedule spec + slicer (Phase H-A). + +Covers, per REPLAY_HARNESS_DESIGN.md §4/§5: +- timestamp sort with input-order tiebreak (via lifted ReplayDataset.build) +- revotes preserved (no dedup at source) +- every cut mode (vote-count / timestamp / fraction / explicit-event-index) +- per-day preset from real timestamps +- empty / degenerate schedules +- moderation interleave +- schedule-spec JSON round-trip (verbatim) +""" + +import json + +import pytest + +from polismath.replay.types import ReplayDataset, ModEvent +from polismath.replay import schedule as sched + + +# -------------------------------------------------------------------------- +# Fixtures: tiny hand-built datasets with known ordering / revotes. +# -------------------------------------------------------------------------- +def _raw(rows): + """rows: list of (t_ms, pid, tid, sign).""" + return list(rows) + + +@pytest.fixture +def ds_unsorted_with_ties(): + # File order deliberately out-of-order, with two rows sharing t_ms=100. + # (pid, tid, sign). Input order is the tuple order below. + raw = _raw([ + (300, 1, 10, 1), # input idx 0 + (100, 2, 10, -1), # input idx 1 (t=100, tie A) + (200, 3, 11, 1), # input idx 2 + (100, 4, 11, 0), # input idx 3 (t=100, tie B, later input order) + (150, 5, 12, 1), # input idx 4 + ]) + return ReplayDataset.build(raw) + + +@pytest.fixture +def ds_with_revote(): + # (pid=1, tid=10) votes twice: once at t=100 (agree), again at t=400 (disagree). + raw = _raw([ + (100, 1, 10, 1), + (200, 2, 10, 1), + (300, 3, 11, -1), + (400, 1, 10, -1), # revote by pid 1 on tid 10 (later-vote-wins is engine's job) + ]) + return ReplayDataset.build(raw) + + +@pytest.fixture +def ds8(): + # 8 votes, strictly increasing timestamps 100..800. + raw = _raw([(100 * (i + 1), i + 1, (i % 3) + 10, 1) for i in range(8)]) + return ReplayDataset.build(raw) + + +# -------------------------------------------------------------------------- +# Sorting + revote invariants (contract the slicer relies on). +# -------------------------------------------------------------------------- +def test_votes_sorted_by_timestamp_with_input_order_tiebreak(ds_unsorted_with_ties): + ds = ds_unsorted_with_ties + times = [v.t_ms for v in ds.votes] + assert times == sorted(times), "votes must be time-sorted" + # The two t=100 rows must keep input order (pid 2 before pid 4). + at_100 = [v.pid for v in ds.votes if v.t_ms == 100] + assert at_100 == [2, 4], "same-timestamp rows must preserve input order" + # k is 1-based and contiguous in sorted order. + assert [v.k for v in ds.votes] == [1, 2, 3, 4, 5] + + +def test_revotes_preserved(ds_with_revote): + ds = ds_with_revote + assert ds.n == 4, "revotes must NOT be deduped at source" + # The later (pid1,tid10) occurrence is flagged as a revote. + revote_ks = [v.k for v in ds.votes if v.is_revote] + assert len(revote_ks) == 1 + revote = ds.votes[revote_ks[0] - 1] + assert (revote.pid, revote.tid, revote.sign) == (1, 10, -1) + + +# -------------------------------------------------------------------------- +# Cut-mode resolution. +# -------------------------------------------------------------------------- +def test_resolve_vote_count_mode(ds8): + slots = sched.resolve_cut_slots(ds8, {"mode": "vote-count", "at": [2, 5, "end"]}) + assert slots == (2, 5, 8) + + +def test_resolve_explicit_index_mode(ds8): + slots = sched.resolve_cut_slots(ds8, {"mode": "explicit-event-index", "at": [3, 6]}) + assert slots == (3, 6) + + +def test_resolve_fraction_mode(ds8): + slots = sched.resolve_cut_slots(ds8, {"mode": "fraction", "at": [0.25, 0.5, 1.0]}) + assert slots == (2, 4, 8) + + +def test_resolve_timestamp_mode(ds8): + # Timestamps are 100..800 (ms). A cut at t=250 covers votes 1,2 (t=100,200). + slots = sched.resolve_cut_slots( + ds8, {"mode": "timestamp", "at": [250, 550, "end"]} + ) + assert slots == (2, 5, 8) + + +def test_resolve_dedupes_and_sorts(ds8): + slots = sched.resolve_cut_slots(ds8, {"mode": "vote-count", "at": [5, 2, 5, "end"]}) + assert slots == (2, 5, 8) + + +def test_cut_slot_out_of_range_raises(ds8): + with pytest.raises(ValueError): + sched.resolve_cut_slots(ds8, {"mode": "vote-count", "at": [999]}) + + +def test_timestamp_before_first_vote_dropped(ds8): + # t=50 is before the first vote (t=100) → slot 0 → dropped (degenerate). + slots = sched.resolve_cut_slots(ds8, {"mode": "timestamp", "at": [50, "end"]}) + assert slots == (8,) + + +def test_unknown_mode_raises(ds8): + with pytest.raises(ValueError): + sched.resolve_cut_slots(ds8, {"mode": "no-such-mode", "at": [1]}) + + +# -------------------------------------------------------------------------- +# Slicer: batching partitions the vote stream with no loss / duplication. +# -------------------------------------------------------------------------- +def test_slice_partitions_votes(ds8): + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": [2, 5, "end"]} + ) + steps = sched.slice_schedule(ds8, spec) + assert [s.cut_slot for s in steps] == [2, 5, 8] + assert [len(s.vote_events) for s in steps] == [2, 3, 3] + # Concatenated batches == the full sorted vote stream (order + identity). + flat = [v.k for s in steps for v in s.vote_events] + assert flat == list(range(1, 9)) + # cut_time_ms is the last vote's timestamp in each batch. + assert [s.cut_time_ms for s in steps] == [200, 500, 800] + + +def test_slice_preserves_revotes_in_batches(ds_with_revote): + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": [2, "end"]} + ) + steps = sched.slice_schedule(ds_with_revote, spec) + all_pairs = [(v.pid, v.tid, v.sign) for s in steps for v in s.vote_events] + assert (1, 10, 1) in all_pairs and (1, 10, -1) in all_pairs + assert len(all_pairs) == 4 # nothing deduped + + +def test_empty_schedule_yields_no_steps(ds8): + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": []} + ) + assert sched.slice_schedule(ds8, spec) == [] + + +def test_slice_drops_tail_after_last_cut(ds8): + # Last cut at 5 (< n=8): votes 6..8 are NOT recomputed → no trailing step. + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", cuts={"mode": "vote-count", "at": [2, 5]} + ) + steps = sched.slice_schedule(ds8, spec) + assert [s.cut_slot for s in steps] == [2, 5] + assert sum(len(s.vote_events) for s in steps) == 5 + + +# -------------------------------------------------------------------------- +# Moderation interleave. +# -------------------------------------------------------------------------- +def test_moderation_interleave_assigns_events_to_steps(ds8): + # Mod event at t=250 falls in the second segment (cut at slot 5, t=500); + # its first covering cut is the one whose cut_time_ms >= 250 → slot 5. + mods = [ModEvent(t_ms=250, tid=10, mod=-1)] + ds = ReplayDataset(votes=ds8.votes, comments=ds8.comments, mod_events=mods) + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", + cuts={"mode": "vote-count", "at": [2, 5, "end"]}, + moderation="interleave-by-timestamp", + ) + steps = sched.slice_schedule(ds, spec) + # step0 cut_time=200 (<250) → no mod; step1 cut_time=500 (>=250) → the event. + assert [len(s.mod_events) for s in steps] == [0, 1, 0] + assert steps[1].mod_events[0].tid == 10 + + +def test_moderation_none_ignores_events(ds8): + mods = [ModEvent(t_ms=250, tid=10, mod=-1)] + ds = ReplayDataset(votes=ds8.votes, comments=ds8.comments, mod_events=mods) + spec = sched.ScheduleSpec( + dataset="t", schedule_id="s", + cuts={"mode": "vote-count", "at": [2, 5, "end"]}, + moderation="none", + ) + steps = sched.slice_schedule(ds, spec) + assert all(len(s.mod_events) == 0 for s in steps) + + +# -------------------------------------------------------------------------- +# Presets. +# -------------------------------------------------------------------------- +def test_single_cut_preset(ds8): + spec = sched.preset_single_cut("t", ds8.n) + steps = sched.slice_schedule(ds8, spec) + assert len(steps) == 1 + assert steps[0].cut_slot == 8 + assert len(steps[0].vote_events) == 8 + + +def test_every_vote_preset(ds8): + spec = sched.preset_every_vote("t", ds8.n) + steps = sched.slice_schedule(ds8, spec) + assert len(steps) == 8 + assert all(len(s.vote_events) == 1 for s in steps) + + +def test_uniform_preset(ds8): + spec = sched.preset_uniform("t", ds8.n, n_cuts=4) + slots = sched.resolve_cut_slots(ds8, spec.cuts) + assert slots == (2, 4, 6, 8) + + +def test_front_and_back_loaded_density(): + # Use a bigger n so early-vs-late density differences are visible. + big = ReplayDataset.build([(100 * (i + 1), i + 1, 10, 1) for i in range(100)]) + front = sched.resolve_cut_slots(big, sched.preset_front_loaded("t", 100, n_cuts=5).cuts) + back = sched.resolve_cut_slots(big, sched.preset_back_loaded("t", 100, n_cuts=5).cuts) + # front-loaded: first gap smaller than last gap; back-loaded: reverse. + front_gaps = [b - a for a, b in zip((0,) + front, front)] + back_gaps = [b - a for a, b in zip((0,) + back, back)] + assert front_gaps[0] < front_gaps[-1], f"front-loaded should be denser early: {front}" + assert back_gaps[0] > back_gaps[-1], f"back-loaded should be denser late: {back}" + assert front[-1] == 100 and back[-1] == 100 + + +def test_per_day_preset_from_real_timestamps(): + # 3 UTC days: 2024-11-19, -20, -21. 2 votes/day, out of file order. + day = 24 * 3600 * 1000 + base = 1732000000000 # ~2024-11-19 + raw = [ + (base + 0 * day + 500, 1, 10, 1), + (base + 2 * day + 100, 2, 10, 1), # day 3 first in file + (base + 1 * day + 200, 3, 11, -1), + (base + 0 * day + 900, 4, 11, 1), + (base + 2 * day + 800, 5, 12, 1), + (base + 1 * day + 600, 6, 12, -1), + ] + ds = ReplayDataset.build(raw) + spec = sched.preset_per_day("t", ds) + steps = sched.slice_schedule(ds, spec) + # One recompute per day → 3 steps, each covering that day's 2 votes. + assert len(steps) == 3 + assert [len(s.vote_events) for s in steps] == [2, 2, 2] + + +# -------------------------------------------------------------------------- +# ScheduleSpec JSON round-trip (verbatim). +# -------------------------------------------------------------------------- +def test_schedule_spec_roundtrip(tmp_path): + d = { + "dataset": "vw", + "schedule_id": "front-loaded-01", + "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [50, 100, "end"]}, + "moderation": "interleave-by-timestamp", + "clojure": {"warm_start": "chain"}, + "notes": "front-loads recomputes early", + } + spec = sched.ScheduleSpec.from_dict(d) + assert spec.dataset == "vw" + assert spec.schedule_id == "front-loaded-01" + # to_dict returns the verbatim input dict. + assert spec.to_dict() == d + p = tmp_path / "schedule.json" + spec.write_json(p) + reloaded = sched.ScheduleSpec.from_json_file(p) + assert reloaded.to_dict() == d + assert json.loads(p.read_text()) == d diff --git a/delphi/tests/replay_harness/test_stepcompare.py b/delphi/tests/replay_harness/test_stepcompare.py new file mode 100644 index 000000000..f25c750dc --- /dev/null +++ b/delphi/tests/replay_harness/test_stepcompare.py @@ -0,0 +1,180 @@ +"""Step-comparer tests (Phase H-A, design §8). + +The comparer repoints ``ConversationComparer._compare_dicts`` from the fixed +6 golden stages onto ``{step_i: blob}`` maps. We verify: +- recording vs itself → zero divergence; +- an EXACT-family perturbation (a count) is reported and classed 'exact'; +- a TOLERANT-family perturbation beyond tolerance (a projection) is reported + and classed 'tolerant'; +- a tolerant-family perturbation WITHIN tolerance is NOT reported; +- step-count mismatch between recordings is reported. +""" + +import copy + +import pytest + +from polismath.replay import schedule as sched +from polismath.replay import store as st +from polismath.replay import stepcompare as sc +from polismath.replay.driver import StepRecord + + +def _blob(n=3): + return { + "zid": "t", + "math_tick": 30000, + "n": n, + "n-cmts": 2, + "in-conv": [1, 2, 3], + "pca": {"center": [0.1, 0.2], "comps": [[1.0, 0.0], [0.0, 1.0]]}, + "proj": {"1": [0.5, -0.3], "2": [-0.4, 0.2], "3": [0.1, 0.1]}, + "group-clusters": [{"id": 0, "center": [0.5, 0.5], "members": [1, 2]}], + "repness": {"group_repness": {"0": {"12": {"pat": 0.8, "tid": 12}}}}, + } + + +def _spec(): + return sched.ScheduleSpec.from_dict( + {"dataset": "t", "schedule_id": "cmp-01", "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [2, "end"]}, "moderation": "none"} + ) + + +def _records(blobs): + return [ + StepRecord(index=i, prev_slot=i, cut_slot=i + 1, batch_size=1, + cut_time_ms=100 * (i + 1), blob=b, extras={}) + for i, b in enumerate(blobs) + ] + + +# -------------------------------------------------------------------------- +# compare_step unit level. +# -------------------------------------------------------------------------- +def test_identical_blobs_have_zero_divergence(): + cmp = sc.StepComparer() + r = cmp.compare_step(_blob(), _blob(), 0) + assert r["match"] is True + assert r["n_divergences"] == 0 + + +def test_exact_family_count_divergence_reported(): + cmp = sc.StepComparer() + a = _blob(n=3) + b = _blob(n=4) # count differs + r = cmp.compare_step(a, b, 0) + assert r["match"] is False + assert len(r["families"]["exact"]) >= 1 + assert any(".n" in d["path"] for d in r["families"]["exact"]) + + +def test_tolerant_family_projection_divergence_reported(): + cmp = sc.StepComparer() + a = _blob() + b = copy.deepcopy(a) + b["proj"]["1"] = [9.9, -9.9] # way beyond tolerance + r = cmp.compare_step(a, b, 0) + assert r["match"] is False + assert len(r["families"]["tolerant"]) >= 1 + assert all("proj" in d["path"] for d in r["families"]["tolerant"]) + assert not r["families"]["exact"] + + +def test_consensus_and_silhouette_are_tolerant_family(): + # P6f: top-level consensus + group-clusters silhouette float divergences + # belong to the TOLERANT family (soft signals), not exact. + cmp = sc.StepComparer() + a = _blob() + a["consensus"] = {"agree": {"c0": 0.10}} + a["group-clusters"] = [ + {"id": 0, "center": [0.5, 0.5], "members": [1, 2], "silhouette": 0.70} + ] + b = copy.deepcopy(a) + b["consensus"]["agree"]["c0"] = 0.95 # top-level consensus float + b["group-clusters"][0]["silhouette"] = 0.10 # group-clusters silhouette + r = cmp.compare_step(a, b, 0) + tol_paths = [d["path"] for d in r["families"]["tolerant"]] + assert any(p.endswith(".consensus.agree.c0") for p in tol_paths), tol_paths + assert any("silhouette" in p for p in tol_paths), tol_paths + assert not r["families"]["exact"], r["families"]["exact"] + + +def test_group_clusters_id_mismatch_stays_exact(): + # The silhouette rule must NOT reclassify a structural group-clusters id diff. + cmp = sc.StepComparer() + a = _blob() + a["group-clusters"] = [ + {"id": 0, "center": [0.5, 0.5], "members": [1, 2], "silhouette": 0.7} + ] + b = copy.deepcopy(a) + b["group-clusters"][0]["id"] = 9 # structural (int) divergence + r = cmp.compare_step(a, b, 0) + exact_paths = [d["path"] for d in r["families"]["exact"]] + assert any(p.endswith(".id") for p in exact_paths), r["families"] + + +def test_tolerant_within_tolerance_not_reported(): + cmp = sc.StepComparer(abs_tolerance=1e-6, rel_tolerance=1e-3) + a = _blob() + b = copy.deepcopy(a) + b["repness"]["group_repness"]["0"]["12"]["pat"] = 0.8 + 1e-9 # within tol + r = cmp.compare_step(a, b, 0) + assert r["match"] is True + + +def test_math_tick_ignored(): + cmp = sc.StepComparer() + a = _blob() + b = copy.deepcopy(a) + b["math_tick"] = 99999 # wall-clock, must be ignored + r = cmp.compare_step(a, b, 0) + assert r["match"] is True + + +# -------------------------------------------------------------------------- +# compare_recordings (dir vs dir). +# -------------------------------------------------------------------------- +def test_recording_vs_itself_zero_divergence(tmp_path): + out = st.write_recording(_records([_blob(3), _blob(5)]), _spec(), root=tmp_path) + report = sc.compare_recordings(out, out) + assert report["overall_match"] is True + assert report["aligned_steps"] == 2 + assert not report["step_count_mismatch"] + + +def test_recording_vs_perturbed_reports_divergence(tmp_path): + rec_a = _records([_blob(3), _blob(5)]) + perturbed = [_blob(3), _blob(5)] + perturbed[1]["proj"]["2"] = [7.0, 7.0] # tolerant divergence + perturbed[1]["n"] = 99 # exact divergence + rec_b = _records(perturbed) + + a = st.write_recording(rec_a, _spec(), root=tmp_path / "a") + b = st.write_recording(rec_b, _spec(), root=tmp_path / "b") + report = sc.compare_recordings(a, b) + + assert report["overall_match"] is False + step1 = report["per_step"][1] + assert step1["match"] is False + assert step1["families"]["exact"] and step1["families"]["tolerant"] + # A human-readable summary renders without error. + text = sc.format_report(report) + assert "step 1" in text.lower() + + +def test_step_count_mismatch_reported(tmp_path): + a = st.write_recording(_records([_blob(3), _blob(5)]), _spec(), root=tmp_path / "a") + b = st.write_recording(_records([_blob(3)]), _spec(), root=tmp_path / "b") + report = sc.compare_recordings(a, b) + assert report["step_count_mismatch"] is True + assert report["overall_match"] is False + assert report["aligned_steps"] == 1 + + +@pytest.mark.parametrize("bad", ["..", "py/../..", "a/b", "/abs", ""]) +def test_compare_recordings_rejects_unsafe_engine(bad, tmp_path): + # engine joins the recording dirs as a path component — reject traversal + # values before reading anything. + with pytest.raises(ValueError): + sc.compare_recordings(tmp_path, tmp_path, engine=bad) diff --git a/delphi/tests/replay_harness/test_store.py b/delphi/tests/replay_harness/test_store.py new file mode 100644 index 000000000..72389344b --- /dev/null +++ b/delphi/tests/replay_harness/test_store.py @@ -0,0 +1,176 @@ +"""Recording store + provenance tests (Phase H-A, design §7). + +Covers: store layout under ``.local/replays`` (gitignored), lazy directory +creation, schedule.json written VERBATIM, provenance completeness, numpy-aware +JSON round-trip, and step ordering on load. +""" + +import json + +import numpy as np +import pytest + +from polismath.replay import schedule as sched +from polismath.replay import store as st +from polismath.replay.driver import StepRecord + + +def _spec(): + return sched.ScheduleSpec.from_dict( + { + "dataset": "vw", + "schedule_id": "unit-01", + "source": "votes-csv", + "cuts": {"mode": "vote-count", "at": [2, "end"]}, + "moderation": "none", + "clojure": {"warm_start": "chain"}, + "notes": "store unit test", + } + ) + + +def _records(): + # Two tiny step records with numpy content to exercise the encoder. + return [ + StepRecord( + index=0, prev_slot=0, cut_slot=2, batch_size=2, cut_time_ms=200, + blob={ + "zid": "vw", "math_tick": 30001, "n": 2, + "pca": {"center": np.array([0.1, 0.2]), + "comps": [np.array([1.0, 0.0]), np.array([0.0, 1.0])]}, + "in-conv": [1, 2], + }, + extras={"n_participants": 2, "n_base_clusters": np.int64(1)}, + ), + StepRecord( + index=1, prev_slot=2, cut_slot=4, batch_size=2, cut_time_ms=400, + blob={"zid": "vw", "math_tick": 30002, "n": 4, "in-conv": [1, 2, 3]}, + extras={"n_participants": 4, "n_base_clusters": np.int64(2)}, + ), + ] + + +def test_recording_dir_under_local_replays(tmp_path): + d = st.recording_dir("vw", "unit-01", root=tmp_path) + assert d == tmp_path / "vw" / "unit-01" + # The production default root lives under the gitignored .local/replays. + default = st.recording_dir("vw", "unit-01") + assert ".local" in default.parts and "replays" in default.parts + + +def test_recording_dir_is_lazy(tmp_path): + d = st.recording_dir("vw", "unit-01", root=tmp_path) + assert not d.exists(), "recording_dir must not create anything" + + +@pytest.mark.parametrize("bad", ["..", "a/b", "/abs", ".", "a\\b", ""]) +def test_recording_dir_rejects_unsafe_dataset(bad, tmp_path): + # P6d: dataset / schedule_id flow into the on-disk path — a '..' or absolute + # value would escape the store root. Reject rather than traverse. + with pytest.raises(ValueError): + st.recording_dir(bad, "ok", root=tmp_path) + + +@pytest.mark.parametrize("bad", ["..", "a/b", "/abs", "."]) +def test_recording_dir_rejects_unsafe_schedule_id(bad, tmp_path): + with pytest.raises(ValueError): + st.recording_dir("ok", bad, root=tmp_path) + + +def test_write_recording_rejects_traversal(tmp_path): + spec = _spec() + object.__setattr__(spec, "schedule_id", "../escape") + with pytest.raises(ValueError): + st.write_recording(_records(), spec, root=tmp_path) + + +def test_write_creates_layout(tmp_path): + out = st.write_recording(_records(), _spec(), root=tmp_path) + assert (out / "schedule.json").exists() + assert (out / "provenance.json").exists() + assert (out / "py" / "step-000.json").exists() + assert (out / "py" / "step-001.json").exists() + + +def test_schedule_json_is_verbatim(tmp_path): + spec = _spec() + out = st.write_recording(_records(), spec, root=tmp_path) + written = json.loads((out / "schedule.json").read_text()) + assert written == spec.to_dict() + + +def test_provenance_completeness(tmp_path): + out = st.write_recording(_records(), _spec(), root=tmp_path) + prov = json.loads((out / "provenance.json").read_text()) + for key in [ + "delphi_git_commit", "dataset", "created_at", "python_version", + "vote_sign_convention", "engine", "engine_flags", "n_steps", + "schedule_id", "source", "packages", + ]: + assert key in prov, f"provenance missing {key!r}" + assert prov["n_steps"] == 2 + assert prov["schedule_id"] == "unit-01" + assert prov["vote_sign_convention"] == "delphi" + assert "POLISMATH_PCA_IMPL" in prov["engine_flags"] + # Engine mode changes warm-start behavior across steps — it MUST be pinned + # in provenance (review finding B, 2026-07-18). + assert "POLISMATH_ENGINE_MODE" in prov["engine_flags"] + # Dataset sha256 recorded (public vw dataset is resolvable on this branch). + assert prov["dataset"]["name"] == "vw" + assert prov["dataset"].get("votes_sha256") + + +def test_numpy_roundtrip_and_step_order(tmp_path): + out = st.write_recording(_records(), _spec(), root=tmp_path) + rec = st.load_recording(out) + assert [s["index"] for s in rec.steps] == [0, 1] + b0 = rec.steps[0]["blob"] + # numpy arrays serialize to plain lists; numpy scalars to numbers. + assert b0["pca"]["center"] == [0.1, 0.2] + assert b0["pca"]["comps"] == [[1.0, 0.0], [0.0, 1.0]] + assert rec.steps[0]["extras"]["n_base_clusters"] == 1 + assert rec.schedule == _spec().to_dict() + assert rec.provenance["n_steps"] == 2 + + +def test_load_steps_returns_blobs_in_order(tmp_path): + out = st.write_recording(_records(), _spec(), root=tmp_path) + blobs = st.load_step_blobs(out / "py") + assert len(blobs) == 2 + assert [b["n"] for b in blobs] == [2, 4] + + +def _third_record(): + return StepRecord( + index=2, prev_slot=4, cut_slot=6, batch_size=2, cut_time_ms=600, + blob={"zid": "vw", "math_tick": 30003, "n": 6, "in-conv": [1, 2, 3, 4]}, + extras={"n_participants": 6, "n_base_clusters": np.int64(3)}, + ) + + +def test_rerun_with_fewer_steps_clears_stale(tmp_path): + """T4: a re-run producing FEWER steps must not leave stale step files that + loaders (which glob every step-*.json) would silently mix into the result.""" + spec = _spec() + # First recording: 3 steps. + st.write_recording(_records() + [_third_record()], spec, root=tmp_path) + step_dir = st.recording_dir("vw", "unit-01", root=tmp_path) / "py" + assert (step_dir / "step-002.json").exists() + + # Re-run: only 2 steps. The stale step-002 must be gone. + st.write_recording(_records(), spec, root=tmp_path) + assert (step_dir / "step-000.json").exists() + assert (step_dir / "step-001.json").exists() + assert not (step_dir / "step-002.json").exists(), "stale step file not cleared" + + rec = st.load_recording(st.recording_dir("vw", "unit-01", root=tmp_path)) + assert [s["index"] for s in rec.steps] == [0, 1] + assert rec.provenance["n_steps"] == 2 + + +@pytest.mark.parametrize("bad", ["..", "py/../..", "a/b", "/abs", ""]) +def test_write_recording_rejects_unsafe_engine(bad, tmp_path): + # engine is a path component too (py/, clj/) — same traversal rules as + # dataset / schedule_id. + with pytest.raises(ValueError): + st.write_recording(_records(), _spec(), root=tmp_path, engine=bad)