From df93b0d52bfb5a7d4f78973e357cc525b0760da8 Mon Sep 17 00:00:00 2001 From: Julien Cornebise Date: Mon, 27 Jul 2026 17:52:54 +0200 Subject: [PATCH] =?UTF-8?q?python-math=20#41:=20feat(math):=20mode=20colla?= =?UTF-8?q?pse=20C5+C6=20=E2=80=94=20driver=20mod=20semantics=20+=20poller?= =?UTF-8?q?=20flag=20plumbing=20deleted?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Phase 2, chunks C5+C6 of `GOAL_CUTOVER_READY.md` (the standing goal document for the Python math cutover) — continuing the mode collapse, the removal of the engine-mode switch (`POLISMATH_ENGINE_MODE`, clojure-legacy vs improved) so only the Clojure-exact behavior remains. - Replay driver: the `mod_update` reducer semantics — a moderation update triggers a votes-recompute first and updates only the moderation sets and watermark, with the effect landing on the next tick — is now the only moderation path. Deleted: the improved-mode `update_moderation` path, its cumulative `mod_state`, the emptying-transition guard (`_guard_moderation_clear` / `_mod_dict`), and the `NotImplementedError` raised on improved mode + restart. - Poller service (the service that polls Postgres for new votes and runs math ticks): the `engine_mode` config field, the `apply_engine_mode` passthrough, and the env import and startup-log field are deleted — the poller no longer reads or writes `POLISMATH_ENGINE_MODE`. `poller/__init__` and `env_flags` docstrings updated. - `utils/engine_mode.py` itself survives until the harness purge in the next commit: `certify.py` / `store.py` / `poller_equiv.py` still import it. ## Testing Deleted the 3 driver tests and 2 poller tests that pinned the removed paths. 21 driver + 93 poller tests green. commit-id:a4c818c8 --- delphi/polismath/poller/__init__.py | 3 +- delphi/polismath/poller/service.py | 29 +----- delphi/polismath/replay/driver.py | 107 +++------------------ delphi/polismath/utils/env_flags.py | 9 +- delphi/tests/poller/test_service.py | 26 ----- delphi/tests/replay_harness/test_driver.py | 48 --------- 6 files changed, 22 insertions(+), 200 deletions(-) diff --git a/delphi/polismath/poller/__init__.py b/delphi/polismath/poller/__init__.py index ddc064a00..b293ebf5d 100644 --- a/delphi/polismath/poller/__init__.py +++ b/delphi/polismath/poller/__init__.py @@ -20,7 +20,7 @@ │ moderation); bounded concurrency across zids ├─ engine: Conversation held in memory per zid │ update_votes(recompute=False) -> update_moderation(recompute=False) - │ -> recompute() (POLISMATH_ENGINE_MODE honored) + │ -> recompute() ├─ load-or-init (first message per zid): from_dict(math_main) warm │ restore + full-history rating-matrix rebuild ├─ math_writer.MathWriter: math_main (caching_tick=MAX+1), math_bidtopid @@ -136,7 +136,6 @@ poll_from_days_ago POLL_FROM_DAYS_AGO 10 allowlist POLL_ALLOWLIST | MATH_ZID_ALLOWLIST [] blocklist POLL_BLOCKLIST | MATH_ZID_BLOCKLIST [] -engine_mode POLISMATH_ENGINE_MODE improved worker_pool_size MATH_WORKER_POOL_SIZE 4 dump_dir MATH_POLLER_DUMP_DIR scratch/errorconv retry_cap MATH_POLLER_RETRY_CAP 1 diff --git a/delphi/polismath/poller/service.py b/delphi/polismath/poller/service.py index 8f12de2fe..e021366b0 100644 --- a/delphi/polismath/poller/service.py +++ b/delphi/polismath/poller/service.py @@ -20,11 +20,6 @@ from typing import Any, Dict, Iterable, List, Optional from polismath.conversation.conversation import Conversation -from polismath.utils.engine_mode import ( - ENGINE_MODE_ENV_VAR, - ENGINE_MODE_CHOICES, - resolve_engine_mode, -) from polismath.poller.math_writer import MathWriter, dump_error from polismath.poller.worker_pool import ( ConversationWorkerPool, @@ -139,7 +134,6 @@ class PollerConfig: poll_from_days_ago POLL_FROM_DAYS_AGO (default 10) allowlist POLL_ALLOWLIST | MATH_ZID_ALLOWLIST (default []) blocklist POLL_BLOCKLIST | MATH_ZID_BLOCKLIST (default []) - engine_mode POLISMATH_ENGINE_MODE (default None -> compute's own default) shard_index POLL_SHARD_INDEX | MATH_SHARD_INDEX (default 0) shard_count POLL_SHARD_COUNT | MATH_SHARD_COUNT (default 1 = unsharded) worker_pool_size MATH_WORKER_POOL_SIZE (default 4) @@ -155,7 +149,6 @@ class PollerConfig: poll_from_days_ago: float = 10 allowlist: List[int] = field(default_factory=list) blocklist: List[int] = field(default_factory=list) - engine_mode: Optional[str] = None # zid-sharding: this process handles zids where zid % shard_count == # shard_index. shard_count=1 (the default) is unsharded -- every zid. # One shard = one PROCESS: threads do not parallelise this workload @@ -230,7 +223,6 @@ def from_env(cls) -> "PollerConfig": blocklist=_parse_int_list( _env_first("POLL_BLOCKLIST", "MATH_ZID_BLOCKLIST") ), - engine_mode=os.environ.get(ENGINE_MODE_ENV_VAR), shard_index=int( _env_first("POLL_SHARD_INDEX", "MATH_SHARD_INDEX", default="0") ), @@ -265,22 +257,6 @@ def __init__(self, pg_client: Any, config: PollerConfig): self._vote_wm: Optional[int] = None self._mod_wm: Optional[int] = None - # -- engine-mode passthrough ------------------------------------------- # - def apply_engine_mode(self) -> str: - """Propagate the configured engine mode into the process environment so - the in-process compute (conversation._compute_pca/_compute_clusters, - which read POLISMATH_ENGINE_MODE at call time) honors it. Returns the - resolved mode actually in effect.""" - if self.config.engine_mode: - if self.config.engine_mode not in ENGINE_MODE_CHOICES: - logger.warning( - "Unknown POLISMATH_ENGINE_MODE=%r; compute will fall back to " - "its default", - self.config.engine_mode, - ) - os.environ[ENGINE_MODE_ENV_VAR] = self.config.engine_mode - return resolve_engine_mode() - # -- lifecycle ---------------------------------------------------------- # def _ensure_runtime(self) -> None: if self._pool is None: @@ -293,7 +269,6 @@ def _ensure_runtime(self) -> None: self._mod_wm = initial_watermark(self.config.poll_from_days_ago) def start(self) -> None: - self.apply_engine_mode() self._ensure_runtime() self._stop.clear() self._threads = [ @@ -303,9 +278,8 @@ def start(self) -> None: for t in self._threads: t.start() logger.info( - "MathPollerService started (math_env=%s engine_mode=%s pool=%d shard=%s)", + "MathPollerService started (math_env=%s pool=%d shard=%s)", self.config.math_env, - resolve_engine_mode(), self.config.worker_pool_size, # Spelled out so a misconfigured fleet is visible in the logs rather # than silently leaving a slice of conversations unprocessed. @@ -337,7 +311,6 @@ def poll_once(self) -> None: Used by ``--once`` and the integration test. """ - self.apply_engine_mode() self._ensure_runtime() self._poll_votes_once() self._poll_moderation_once() diff --git a/delphi/polismath/replay/driver.py b/delphi/polismath/replay/driver.py index b966815a8..34e4929a5 100644 --- a/delphi/polismath/replay/driver.py +++ b/delphi/polismath/replay/driver.py @@ -18,17 +18,10 @@ 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. +- ``mod_update(rows)`` → new Conversation. Clojure reducer semantics: sets + and watermark only, NO recompute — a mod change's effect on the math lands + at the NEXT votes recompute. (The former improved-mode ``update_moderation`` + driver path and its clear-transition guard went with the mode collapse.) - ``recompute()`` → new Conversation recomputing PCA→clusters→repness→ priorities→participant-info on the moderation-applied matrix. Standalone after an ``update_votes(recompute=False)``. @@ -50,7 +43,6 @@ from polismath.conversation.conversation import Conversation from polismath.replay.schedule import ReplayStep, ScheduleSpec, slice_schedule from polismath.replay.types import ModEvent, ReplayDataset -from polismath.utils.engine_mode import ENGINE_MODE_LEGACY, resolve_engine_mode # Vote sign convention recorded in provenance; the future Clojure driver flips. VOTE_SIGN_CONVENTION = "delphi" # AGREE=+1 (export convention, no re-flip) @@ -114,24 +106,6 @@ def run_replay( # certify-cold-start-pca. conv.pca = {'center': np.zeros(1), 'comps': np.array([[1.0], [1.0]])} - # Cumulative latest-wins moderation value per tid across the whole replay - # (improved-mode path only; legacy mode carries its own mod state on - # `conv` via `mod_update` — see the branch below). - mod_state: dict[int, int] = {} - legacy = resolve_engine_mode() == ENGINE_MODE_LEGACY - - # The restart seam replays woven mods via mod_update — Clojure's (and - # legacy mode's) reducer semantics. Improved mode moderates through - # update_moderation (truthy-replace lists); silently applying mod_update - # at its restart seam would mix semantics (#2656 review, 2026-07-24). - if spec.restart_after is not None and spec.moderation != "none" and not legacy: - raise NotImplementedError( - "restart_after with a moderation-bearing schedule is only " - "implemented for clojure-legacy engine mode: the restart seam " - "replays woven mods via mod_update (legacy reducer semantics), " - "which does not mirror improved mode's update_moderation." - ) - records: list[StepRecord] = [] # Mods woven into steps so far — the restart seam replays exactly these # (clj restart-conv: (mapcat :mods steps-so-far)), NEVER dataset.mod_events @@ -144,28 +118,18 @@ def run_replay( conv = conv.update_votes(_votes_dict(step), recompute=False) - if legacy: - # Clojure batch order (:votes :moderation, conv_man.clj:361-371): - # the votes recompute runs FIRST, on the PRIOR step's mod state. - # mod_update then touches only sets/watermark for THIS step's - # blob — NO recompute — so a mod change's effect on the math - # lands at the NEXT votes recompute (module docstring / conv/ - # mod_update docstring). moderation="none" schedules never reach - # the `if step.mod_events` branch below, so this is bit-identical - # to the pre-existing (unconditional) `conv.recompute()` call for - # every schedule that doesn't request moderation. - conv = conv.recompute() - if step.mod_events: - conv = conv.mod_update(_mod_rows(step.mod_events)) - else: - 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() + # Clojure batch order (:votes :moderation, conv_man.clj:361-371): + # the votes recompute runs FIRST, on the PRIOR step's mod state. + # mod_update then touches only sets/watermark for THIS step's + # blob — NO recompute — so a mod change's effect on the math + # lands at the NEXT votes recompute (module docstring / conv/ + # mod_update docstring). moderation="none" schedules never reach + # the `if step.mod_events` branch below, so this is bit-identical + # to a plain unconditional `conv.recompute()` for every schedule + # that doesn't request moderation. + conv = conv.recompute() + if step.mod_events: + conv = conv.mod_update(_mod_rows(step.mod_events)) record = StepRecord( index=step.index, @@ -201,34 +165,6 @@ def _votes_dict(step: ReplayStep) -> dict[str, Any]: 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_rows(events: tuple[ModEvent, ...]) -> list[dict[str, Any]]: """Map a batch of ModEvents to ``Conversation.mod_update``'s row shape (``{tid, is_meta, mod, modified}`` — conversation.clj:846-884 parity).""" @@ -280,17 +216,6 @@ def _restart_conversation( return restored -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. diff --git a/delphi/polismath/utils/env_flags.py b/delphi/polismath/utils/env_flags.py index 3e0e2c911..5b7ffead6 100644 --- a/delphi/polismath/utils/env_flags.py +++ b/delphi/polismath/utils/env_flags.py @@ -1,8 +1,8 @@ """ Shared resolver for legacy-vs-improved implementation switches. -Pattern for env-var implementation switches (POLISMATH_PCA_IMPL, -POLISMATH_ENGINE_MODE, and future ones like a k-means solver switch): a +Pattern for env-var implementation switches (POLISMATH_PCA_IMPL, and +future ones like a k-means solver switch): a module-level env var name + default + allowed values, resolved by `resolve_impl_flag` AT CALL TIME (never at import time), so tests and operators can flip the env var without re-importing. Unknown values fall back @@ -10,9 +10,8 @@ crash the math worker). This lives in polismath.utils (not pca.py, where it originated) so that -lightweight consumers — e.g. `polismath.utils.engine_mode`, read on every -conv-update tick — do not drag in the numpy/pandas pca import chain, and resolution -warnings are logged under this module's logger rather than pca's. +lightweight consumers do not drag in the numpy/pandas pca import chain, and +resolution warnings are logged under this module's logger rather than pca's. """ import logging diff --git a/delphi/tests/poller/test_service.py b/delphi/tests/poller/test_service.py index cc2046c42..8ef88c880 100644 --- a/delphi/tests/poller/test_service.py +++ b/delphi/tests/poller/test_service.py @@ -5,7 +5,6 @@ from unittest.mock import MagicMock from polismath.poller.service import MathPollerService, PollerConfig -from polismath.utils.engine_mode import resolve_engine_mode def _vote_row(zid, created, pid="1", tid="1"): @@ -74,31 +73,6 @@ def test_moderation_dispatch_and_watermark(self): assert svc._mod_wm == 250 -class TestEngineModePassthrough: - def test_configured_mode_is_pushed_into_env(self, monkeypatch): - # apply_engine_mode() writes os.environ directly, which monkeypatch's - # delenv undo does NOT cover when the var was absent — restore by hand - # or the mode leaks into every later test in this worker. - monkeypatch.delenv("POLISMATH_ENGINE_MODE", raising=False) - try: - svc = MathPollerService( - MagicMock(), PollerConfig(engine_mode="clojure-legacy") - ) - resolved = svc.apply_engine_mode() - assert os.environ["POLISMATH_ENGINE_MODE"] == "clojure-legacy" - assert resolved == "clojure-legacy" - # The in-process compute resolves the SAME value at call time. - assert resolve_engine_mode() == "clojure-legacy" - finally: - os.environ.pop("POLISMATH_ENGINE_MODE", None) - - def test_no_configured_mode_leaves_compute_default(self, monkeypatch): - monkeypatch.delenv("POLISMATH_ENGINE_MODE", raising=False) - svc = MathPollerService(MagicMock(), PollerConfig(engine_mode=None)) - resolved = svc.apply_engine_mode() - assert resolved == "improved" # engine_mode.ENGINE_MODE_DEFAULT - - class TestShardedDispatch: """Two shard processes over the SAME polled rows must partition the work: disjoint (nothing double-processed, since per-zid serialisation does not diff --git a/delphi/tests/replay_harness/test_driver.py b/delphi/tests/replay_harness/test_driver.py index 85499afd8..dcb14ab5f 100644 --- a/delphi/tests/replay_harness/test_driver.py +++ b/delphi/tests/replay_harness/test_driver.py @@ -136,20 +136,6 @@ def _mod_spec(mod_events): }) -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. @@ -267,24 +253,6 @@ def _spy(self, mods): assert records[-1].blob["moderation"]["mod_out_tids"] == [] -def test_improved_mode_still_uses_update_moderation_and_guard(monkeypatch): - # Explicit control: 'improved' (default, no env override) keeps using - # update_moderation + _guard_moderation_clear, never mod_update. - calls = [] - original = Conversation.mod_update - - def _spy(self, mods): - calls.append(list(mods)) - return original(self, mods) - - monkeypatch.setattr(Conversation, "mod_update", _spy) - mods = [ModEvent(35, 100, -1), ModEvent(55, 101, 1)] - ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) - records = _run_legacy(ds, _mod_spec(mods)) - assert len(records) == 2 - assert calls == [] # mod_update never called on the improved path - - # --- restart_after: worker-restart seam ------------------------------------ # MOD_RESTART_PORT_SPEC.md "Replay-step semantics" / restart plumbing: after # recording the step at spec.restart_after, the driver rebuilds the @@ -416,22 +384,6 @@ def test_restart_replays_woven_mods_so_far(monkeypatch): assert records[1].blob["moderation"]["mod_out_tids"] == [100] -def test_restart_with_moderation_requires_legacy_mode(): - # #2656 review (2026-07-24): the restart seam replays woven mods via - # mod_update (legacy reducer semantics); combining restart_after with a - # moderation-bearing schedule in IMPROVED mode would silently apply the - # wrong moderation semantics — fail loudly instead. - mods = [ModEvent(35, 100, -1)] - ds = ReplayDataset.build(_MOD_RAW_VOTES, mod_events=mods) - spec = sched.ScheduleSpec.from_dict({ - "dataset": "vw", "schedule_id": "restart-improved", "source": "votes-csv", - "cuts": _MOD_CUTS, "moderation": "interleave-by-timestamp", - "clojure": {"warm_start": "chain"}, "notes": "", "restart_after": 0, - }) - with pytest.raises(NotImplementedError, match="clojure-legacy"): - _run_legacy(ds, spec) # improved mode: no env override set - - @pytest.mark.parametrize("bad", [-1, 3, 4]) def test_restart_after_out_of_range_raises(monkeypatch, bad): # replay.clj CLI parity: restart_after must be a step index with at least