diff --git a/delphi/polismath/conversation/conversation.py b/delphi/polismath/conversation/conversation.py index 478229d3b0..dc2a65a1e0 100644 --- a/delphi/polismath/conversation/conversation.py +++ b/delphi/polismath/conversation/conversation.py @@ -172,6 +172,16 @@ def __init__(self, self.base_clusters = [] self.group_clusters = [] self.subgroup_clusters = {} + + # Warm-start state threaded across ticks in 'clojure-legacy' engine + # mode (see polismath.utils.engine_mode). Clojure carries these on the + # conv (conversation.clj:433-484): the per-k group clusterings and the + # group-k-smoother state {last_k, last_k_count, smoothed_k}. Cold + # default is empty (first tick); NOT persisted to/from dynamo — they + # thread in-memory only, exactly as Clojure's math_main whitelist omits + # them (conv_man.clj:52-74). Unused in the default 'improved' mode. + self.group_clusterings: Dict[Any, Any] = {} # k -> (labels, centers, member_lists, silhouette) + self.group_k_smoother: Dict[str, Any] = {} # {last_k, last_k_count, smoothed_k} self.proj = {} self.repness = None self.consensus = [] @@ -569,12 +579,17 @@ def update_moderation(self, return result - def _compute_pca(self, n_components: int = 2) -> None: + def _compute_pca(self, n_components: int = 2, + prev_pca: Optional[Dict[str, Any]] = None) -> None: """ Compute PCA on the vote matrix. Args: n_components: Number of principal components + prev_pca: The previous tick's PCA result ({'center', 'comps'}) or + None. Consumed ONLY in 'clojure-legacy' engine mode as the + power-iteration warm start (Clojure :start-vectors, + conversation.clj:385). Ignored in the default 'improved' mode. """ import time start_time = time.time() @@ -665,12 +680,28 @@ def _get_clean_matrix(self, raw: bool = False) -> pd.DataFrame: return pd.DataFrame(matrix_data, index=source.index, columns=source.columns) - def _compute_clusters(self) -> None: + def _compute_clusters(self, + prev_group_clusterings: Optional[Dict[Any, Any]] = None, + prev_group_k_smoother: Optional[Dict[str, Any]] = None) -> None: """ Compute two-level hierarchical clustering matching Clojure architecture. Level 1: Base clusters (participants → ~100 clusters) Level 2: Group clusters (base clusters → 2-5 groups with silhouette-based k selection) + + Args: + prev_group_clusterings: The previous tick's per-k group clusterings + dict (k -> clustering tuple), or None. Captured and threaded by + recompute() but NOT consumed yet: the per-k k-means warm start + that will use it (Clojure :group-clusterings → :last-clusters, + conversation.clj:441-442) lands with the Clojure-exact k-means + lineage port (PR-C; see SEQUENTIAL_BITS_PORT_SPEC.md §2.3). + Until then, legacy-mode group clusterings are recomputed cold + each tick. Ignored in the default 'improved' mode. + prev_group_k_smoother: The previous tick's group-k-smoother state + {last_k, last_k_count, smoothed_k}, or None. Consumed ONLY in + 'clojure-legacy' mode (conversation.clj:457). Ignored in + 'improved' mode. """ import time start_time = time.time() @@ -1100,12 +1131,27 @@ def recompute(self) -> 'Conversation': if result.rating_mat.size == 0: # Not enough data, return early return result - + + # Capture the PREVIOUS tick's warm-start state BEFORE the compute steps + # overwrite it. `result` is a deepcopy of self, so result.pca / + # result.group_clusterings / result.group_k_smoother currently hold the + # prior tick's values (deepcopied snapshots). This mirrors Clojure, + # whose fnks read the incoming `conv` for :start-vectors + # (conversation.clj:385) and :group-k-smoother (conversation.clj:457). + # In 'improved' mode (default) these are IGNORED and behavior is + # unchanged; only 'clojure-legacy' mode consumes them. + prev_pca = result.pca + prev_group_clusterings = getattr(result, 'group_clusterings', {}) + prev_group_k_smoother = getattr(result, 'group_k_smoother', {}) + # Compute PCA and projections - result._compute_pca() - + result._compute_pca(prev_pca=prev_pca) + # Compute clusters - result._compute_clusters() + result._compute_clusters( + prev_group_clusterings=prev_group_clusterings, + prev_group_k_smoother=prev_group_k_smoother, + ) # Compute representativeness result._compute_repness() diff --git a/delphi/polismath/utils/engine_mode.py b/delphi/polismath/utils/engine_mode.py new file mode 100644 index 0000000000..9ad70b64c3 --- /dev/null +++ b/delphi/polismath/utils/engine_mode.py @@ -0,0 +1,50 @@ +""" +Engine-mode switch: Clojure-parity warm-start vs improved cold-recompute. + +Python's delphi engine does a full COLD recompute on every conv-update tick. +Clojure instead THREADS warm-start state across ticks: + + - PCA :start-vectors — the previous tick's post-normalization unit + components are fed back in as the power-iteration starting vectors + (math/src/polismath/math/conversation.clj:381-387 -> pca.clj:86-105). + - group-k-smoother — {:last-k :last-k-count :smoothed-k} state carried in + the conv, so the group count K only changes after `:group-k-buffer` (4) + consecutive ticks agree on a new K (conversation.clj:454-478). + +`POLISMATH_ENGINE_MODE` selects between the two families: + + - 'improved' (default): today's cold-recompute behavior, byte-for-byte. + - 'clojure-legacy' : threads the warm-start state described above. + +The flag is resolved AT CALL TIME (never cached at import), reusing the exact +idiom of `pca._resolve_impl_flag` (pca.py:37-57): unknown values fall back to +the default with a warning so a typo in a deployment env cannot crash the math +worker. This lives in a shared spot (polismath.utils) because the mode +cross-cuts both PCA (conversation._compute_pca) and clustering +(conversation._compute_clusters). +""" + +from typing import Sequence + +from polismath.pca_kmeans_rep.pca import _resolve_impl_flag + +ENGINE_MODE_ENV_VAR = 'POLISMATH_ENGINE_MODE' +ENGINE_MODE_LEGACY = 'clojure-legacy' # warm-start parity with Clojure +ENGINE_MODE_IMPROVED = 'improved' # cold recompute every tick (default) +ENGINE_MODE_DEFAULT = ENGINE_MODE_IMPROVED +ENGINE_MODE_CHOICES: Sequence[str] = (ENGINE_MODE_LEGACY, ENGINE_MODE_IMPROVED) + + +def resolve_engine_mode() -> str: + """ + Resolve `POLISMATH_ENGINE_MODE` from the environment, at call time. + + Reuses `pca._resolve_impl_flag` (pca.py:37-57) so the resolution rules + (strip + lowercase, unknown -> default with a warning) are identical to + the PCA-solver switch. + + Returns: + Either 'improved' (default) or 'clojure-legacy'. + """ + return _resolve_impl_flag( + ENGINE_MODE_ENV_VAR, ENGINE_MODE_DEFAULT, ENGINE_MODE_CHOICES) diff --git a/delphi/tests/test_engine_mode.py b/delphi/tests/test_engine_mode.py new file mode 100644 index 0000000000..2600d38237 --- /dev/null +++ b/delphi/tests/test_engine_mode.py @@ -0,0 +1,114 @@ +#!/usr/bin/env python3 +""" +Tests for the POLISMATH_ENGINE_MODE switch (Clojure-parity warm-start vs +improved cold-recompute) and the cold-start invariance guard. + +The engine mode selects between two families of behavior: + - 'improved' (default): full cold recompute every tick — today's behavior. + - 'clojure-legacy' : threads warm-start state across ticks, matching + Clojure (PCA :start-vectors, group-k-smoother). + +On the FIRST tick (cold start) the two modes MUST coincide bit-for-bit, because +Clojure's warm-start state is empty on the first tick (no previous comps, no +smoother state). This module asserts: + 1. Flag resolution semantics (default, valid, invalid, case/whitespace, + read-at-call-time) — mirrors tests/test_powerit_pca.py::TestPcaImplFlag. + 2. Cold-start invariance: a single-shot vw pipeline run is identical under + both modes (guards commits 2 and 3 from diverging on the first tick). +""" + +import os +import sys + +import pytest + +sys.path.append(os.path.abspath(os.path.dirname(__file__))) + +from polismath.utils.engine_mode import ( + ENGINE_MODE_ENV_VAR, + ENGINE_MODE_LEGACY, + ENGINE_MODE_IMPROVED, + ENGINE_MODE_DEFAULT, + ENGINE_MODE_CHOICES, + resolve_engine_mode, +) + + +# --------------------------------------------------------------------------- +# POLISMATH_ENGINE_MODE flag resolution +# --------------------------------------------------------------------------- + +class TestEngineModeFlag: + + def test_default_is_improved(self, monkeypatch): + """Unset env var -> 'improved' (today's behavior is the default).""" + monkeypatch.delenv(ENGINE_MODE_ENV_VAR, raising=False) + assert resolve_engine_mode() == ENGINE_MODE_IMPROVED + assert ENGINE_MODE_DEFAULT == ENGINE_MODE_IMPROVED + + def test_legacy_flag_selects_legacy(self, monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ENGINE_MODE_LEGACY) + assert resolve_engine_mode() == ENGINE_MODE_LEGACY + + def test_improved_flag_selects_improved(self, monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ENGINE_MODE_IMPROVED) + assert resolve_engine_mode() == ENGINE_MODE_IMPROVED + + def test_invalid_flag_value_falls_back_to_default(self, monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, 'not-a-mode') + assert resolve_engine_mode() == ENGINE_MODE_DEFAULT + + def test_case_and_whitespace_insensitive(self, monkeypatch): + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ' Clojure-Legacy ') + assert resolve_engine_mode() == ENGINE_MODE_LEGACY + + def test_flag_read_at_call_time(self, monkeypatch): + """Env var read per call, not cached at import time.""" + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ENGINE_MODE_LEGACY) + assert resolve_engine_mode() == ENGINE_MODE_LEGACY + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, ENGINE_MODE_IMPROVED) + assert resolve_engine_mode() == ENGINE_MODE_IMPROVED + + def test_choices_are_exactly_the_two_modes(self): + assert set(ENGINE_MODE_CHOICES) == {ENGINE_MODE_LEGACY, ENGINE_MODE_IMPROVED} + + +# --------------------------------------------------------------------------- +# Cold-start invariance: improved vs clojure-legacy must coincide on tick 1 +# --------------------------------------------------------------------------- + +def _strip_volatile(d): + """Remove wall-clock fields that legitimately differ between two runs. + + `math_tick` is `25000 + (time_ms % 10000)` (conversation.py:1985) and + `last_updated` is a wall-clock timestamp — both are version counters, not + math output, so they are expected to differ between two independent runs. + """ + d = dict(d) + d.pop('last_updated', None) + d.pop('math_tick', None) + return d + + +class TestColdStartInvariance: + """A single-shot (first-tick) pipeline run must be byte-identical under + both engine modes. This is the hard gate protecting 'improved' mode from + any drift introduced by the legacy warm-start plumbing.""" + + def _recompute_to_dict(self, monkeypatch, mode): + from common_utils import create_test_conversation + monkeypatch.setenv(ENGINE_MODE_ENV_VAR, mode) + conv = create_test_conversation('vw') + # Pin last_updated so the two runs share a deterministic value. + conv.last_updated = 0 + result = conv.recompute() + return _strip_volatile(result.to_dict()) + + def test_vw_cold_run_identical_across_modes(self, monkeypatch): + improved = self._recompute_to_dict(monkeypatch, ENGINE_MODE_IMPROVED) + legacy = self._recompute_to_dict(monkeypatch, ENGINE_MODE_LEGACY) + assert improved == legacy, ( + "Cold-start (first-tick) vw run diverged between 'improved' and " + "'clojure-legacy' engine modes; warm-start state must be empty on " + "tick 1 so the two modes must coincide bit-for-bit." + )