Skip to content

python-math #41: feat(math): mode collapse C5+C6 — driver mod semantics + poller flag plumbing deleted #2670

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
jucor wants to merge 1 commit into
base: spr/edge/fe2b0bb2
Choose a base branch
from
Draft

python-math #41: feat(math): mode collapse C5+C6 — driver mod semantics + poller flag plumbing deleted #2670

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 1 addition & 2 deletions delphi/polismath/poller/__init__.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
29 changes: 1 addition & 28 deletions delphi/polismath/poller/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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)
Expand All @@ -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
Expand Down Expand Up @@ -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")
),
Expand Down Expand Up @@ -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:
Expand All @@ -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 = [
Expand All @@ -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.
Expand Down Expand Up @@ -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()
Expand Down
107 changes: 16 additions & 91 deletions delphi/polismath/replay/driver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)``.
Expand All @@ -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)
Expand Down Expand Up @@ -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
Expand All @@ -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,
Expand Down Expand Up @@ -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)."""
Expand Down Expand Up @@ -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.

Expand Down
9 changes: 4 additions & 5 deletions delphi/polismath/utils/env_flags.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,18 +1,17 @@
"""
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
to the default with a warning (defensive: a typo in a deployment env must not
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
Expand Down
26 changes: 0 additions & 26 deletions delphi/tests/poller/test_service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"):
Expand Down Expand Up @@ -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
Expand Down
48 changes: 0 additions & 48 deletions delphi/tests/replay_harness/test_driver.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
Loading