python-math #5: feat(math): add POLISMATH_ENGINE_MODE flag + prev-tick warm-start scaffolding (PR-A)

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()

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)

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."
+        )