python-math #12: feat(delphi): Python math poller service (phase 1) — Clojure math-container replacement

delphi/polismath/poller/__init__.py

@@ -0,0 +1,153 @@
+"""polismath.poller — Python replacement for the Clojure math poller (phase 1).
+
+A service that polls Postgres for votes/moderation, maintains per-conversation
+math state in-memory, and writes the same Postgres tables the TS server + legacy
+clients consume.  See ``delphi/docs/MATH_POLLER_DESIGN.md`` for the full recon
+and cutover plan.
+
+Architecture
+------------
+::
+
+    scripts/math_poller.py (CLI)
+      └─ service.MathPollerService
+           ├─ vote loop   (thread): poll_votes_since(wm)      cadence VOTE_POLLING_INTERVAL
+           ├─ mod  loop   (thread): poll_moderation_since(wm)  cadence MOD_POLLING_INTERVAL
+           │     both group-by zid, allow/block filter, advance watermark to max(ts)
+           ├─ worker_pool.ConversationWorkerPool
+           │     one FIFO queue + single-owner flag per zid  -> strict per-zid
+           │     serialization; drains+coalesces queued batches (votes-before-
+           │     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)
+           ├─ 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
+           │     (derived from base_clusters), math_ptptstats — one shared math_tick
+           └─ error path: dump conv+batch JSON -> retry once -> park zid (breaker)
+
+Clojure provenance for every duty is cited inline in each module.
+
+bidToPid shape (VERIFIED against both ends of the contract)
+-----------------------------------------------------------
+``math_bidtopid.data`` = ``{"zid", "bidToPid", "lastVoteTimestamp"}`` where
+``bidToPid`` is a LIST OF PID-LISTS, positionally aligned with
+``math_main.base-clusters.id`` (ascending by base-cluster id):
+
+* Clojure ``prep-bidToPid`` (math/src/polismath/conv_man.clj:35-40) wraps
+  ``:bid-to-pid`` = ``(mapv :members (sort-by :id base-clusters))``
+  (math/src/polismath/math/conversation.clj:585-586) — "a vector of member
+  vectors, sorted by base cluster id".
+* The TS server (server/src/utils/participants.ts:33-51 with pca.ts:20-27
+  ``base-clusters.members: number[][]``) indexes ``data.bidToPid`` by the
+  position of a bid inside ``base-clusters.id``:
+  ``bidToIndex[base_clusters.id[i]] = i`` then ``indexToPids[bidToIndex[bid]]``.
+
+Python's ``Conversation.base_clusters`` is sorted ascending by ``id``
+(conversation.py:789) and ``_fold_base_clusters`` writes ``base-clusters.id`` /
+``.members`` in that order (conversation.py:1643-1649), so
+``[c['members'] for c in conv.base_clusters]`` is the exact alignment the server
+needs.  ``derive_bidtopid`` (math_writer.py) implements this.  Pids are strings
+Python-side (poll_votes casts ``str(pid)``) vs ints Clojure-side; the server
+parseInt()s them (participants.ts:53-55), so a parity comparer needs int/str
+tolerance on this one field.
+
+load-or-init finding (from_dict restoration is PARTIAL)
+-------------------------------------------------------
+``Conversation.from_dict`` (conversation.py:2249-2303) restores from a dict with
+underscore/nested keys: ``last_updated, participant_count, comment_count,
+vote_stats, moderation{...}, pca{center,comps}, proj, group_clusters, repness,
+participant_info, comment_priorities``.  ``Conversation.to_dict`` (used as the
+math_main ``data`` blob) is a SUPERSET that carries those same underscore keys
+alongside the hyphenated Clojure keys, so ``from_dict(to_dict(conv))`` round-trips
+the listed fields — notably the PCA warm-start vectors and prior moderation.
+
+But ``from_dict`` does NOT restore: ``raw_rating_mat`` / ``rating_mat`` (the vote
+matrices), ``base_clusters``, ``subgroup_clusters``, ``group_clusterings`` /
+``group_k_smoother`` (warm smoother state), ``consensus`` or ``group_votes``.
+Therefore load-or-init ALWAYS rebuilds the rating matrices from the full vote
+history (``poll_votes(zid)`` ordered by zid,tid,pid,created — parity with
+conv-poll offset 0) and recomputes base_clusters; the non-persisted smoother
+state cold-starts.  This is CLOSE TO — but not byte-identical with — a Clojure
+worker restart: on restart Clojure ``restructure-json-conv`` RESTORES
+``base-clusters`` (and the PCA) from the persisted blob before its ``:reboot``
+recompute (conv_man.clj:173 keeps ``:base-clusters`` in the subset, :180 unfolds
+them), whereas Python re-derives base_clusters cold
+from the vote matrices.  The rating-matrix rebuild itself matches
+(conv_man.clj:188-207 rebuilds ``raw-rating-mat`` the same way), and we
+opportunistically seed the warm PCA start from ``from_dict`` when a row exists
+(low-risk, literally what ``restructure-json-conv`` does).  The base-cluster
+lineage difference is a KNOWN divergence to trace against Clojure's ``:reboot``
+semantics before the parity gate; a full cold rebuild is otherwise correct —
+just without Clojure's restored-lineage warm start.
+
+Config var mapping (config.py names PREFERRED, design aliases accepted)
+-----------------------------------------------------------------------
+======================  ==============================================  =======
+PollerConfig field      Env var(s) (first set wins)                     Default
+======================  ==============================================  =======
+database_url            DATABASE_URL                                    —
+math_env                MATH_ENV                                        dev
+vote_interval_ms        POLL_VOTE_INTERVAL_MS | VOTE_POLLING_INTERVAL |
+                        POLL_INTERVAL_MS                                1000
+mod_interval_ms         POLL_MOD_INTERVAL_MS | MOD_POLLING_INTERVAL |
+                        POLL_INTERVAL_MS                                1000
+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
+======================  ==============================================  =======
+
+``POLL_VOTE_INTERVAL_MS`` / ``POLL_MOD_INTERVAL_MS`` / ``POLL_ALLOWLIST`` /
+``POLL_BLOCKLIST`` are the names already present in
+``polismath.components.config.py`` (:216-269, previously unwired);
+``VOTE_POLLING_INTERVAL`` / ``MOD_POLLING_INTERVAL`` / ``MATH_ZID_ALLOWLIST`` /
+``MATH_ZID_BLOCKLIST`` are the design-doc aliases.  (config.py's example default
+for the mod interval was 5000ms; the binding design §3 uses 1000ms, adopted here.)
+
+Usage
+-----
+::
+
+    # Run the service (blocks; SIGTERM/SIGINT -> graceful stop)
+    uv run python scripts/math_poller.py
+
+    # Single poll cycle then exit (smoke test / cron-style)
+    uv run python scripts/math_poller.py --once
+
+Shadow-mode deployment writes under a DISTINCT ``MATH_ENV`` (e.g. ``delphi``)
+next to the Clojure ``math`` container; ``UNIQUE(zid, math_env)`` keeps the rows
+invisible to the prod server until cutover.
+"""
+
+from polismath.poller.service import (
+    MathPollerService,
+    PollerConfig,
+    advance_watermark,
+    initial_watermark,
+    should_process_zid,
+)
+from polismath.poller.worker_pool import (
+    ConversationWorkerPool,
+    CoalescedBatch,
+    coalesce_messages,
+)
+from polismath.poller.math_writer import MathWriter, derive_bidtopid, dump_error
+
+__all__ = [
+    "MathPollerService",
+    "PollerConfig",
+    "advance_watermark",
+    "initial_watermark",
+    "should_process_zid",
+    "ConversationWorkerPool",
+    "CoalescedBatch",
+    "coalesce_messages",
+    "MathWriter",
+    "derive_bidtopid",
+    "dump_error",
+]

delphi/polismath/poller/math_writer.py

@@ -0,0 +1,172 @@
+"""Postgres writer for the math poller.
+
+Writes the three data tables the TS server + legacy clients consume, all under
+one math_env string and ONE shared math_tick per cycle, exactly like Clojure's
+write-conv-updates! (conv_man.clj:158-169):
+
+    math-tick = inc-math-tick(zid)          ; atomic, postgres.clj:292-295
+    upload-math-main      zid math-tick ...  ; postgres.clj:323-338
+    upload-math-bidtopid  zid math-tick ...  ; postgres.clj:369-380
+    upload-math-ptptstats zid math-tick ...  ; postgres.clj:350-361
+
+The Clojure-exact SQL (caching_tick = MAX+1 subquery, atomic tick upsert) lives
+in polismath.database.postgres.PostgresClient; this module orchestrates the
+per-cycle write and derives the bidToPid blob.
+"""
+
+import json
+import logging
+import os
+import time
+import traceback
+import uuid
+from typing import Any, Dict, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+def derive_bidtopid(conv: Any, zid: int) -> Dict[str, Any]:
+    """Derive the prep-bidToPid blob from a computed Conversation.
+
+    Shape (verified against BOTH sides of the contract):
+
+      * Clojure ``prep-bidToPid`` (conv_man.clj:35-40) emits
+        ``{:zid :bidToPid :lastVoteTimestamp}`` where ``:bidToPid`` is
+        ``(mapv :members (sort-by :id base-clusters))`` (conversation.clj:585-586)
+        — "a vector of member vectors, sorted by base cluster id".
+
+      * The TS server (server/src/utils/participants.ts:33-51,
+        pca.ts:20-27 ``members: number[][]``) indexes ``data.bidToPid`` by the
+        POSITION of a base-cluster id inside ``base-clusters.id``:
+        ``bidToIndex[base_clusters.id[i]] = i`` then ``bidToPid[i]``.
+
+    Python's ``Conversation.base_clusters`` is already sorted ascending by ``id``
+    (conversation.py:789) and ``_fold_base_clusters`` (conversation.py:1643-1649)
+    writes ``base-clusters.id`` / ``base-clusters.members`` in that same order, so
+    ``[c['members'] for c in conv.base_clusters]`` is positionally aligned with
+    ``base-clusters.id`` — the exact alignment the server relies on.
+
+    Note on element type: Python pids are strings (poll_votes casts ``str(pid)``),
+    whereas Clojure emits integer pids.  The server parseInt()s them
+    (participants.ts:53-55) so both work; a parity comparer needs int/str
+    tolerance on this field.  Members are left as-is so that
+    math_bidtopid.bidToPid and math_main.base-clusters.members stay identical.
+
+    Args:
+        conv: A computed Conversation (public attrs only).
+        zid: Conversation id.
+
+    Returns:
+        ``{"zid": int, "bidToPid": [[pid, ...], ...], "lastVoteTimestamp": int}``
+    """
+    base_clusters = getattr(conv, "base_clusters", None) or []
+    # Defensive: never rely on caller having sorted; sort by id here too
+    # (idempotent since conversation.py already keeps them sorted).
+    ordered = sorted(base_clusters, key=lambda c: c["id"])
+    bid_to_pid: List[List[Any]] = [list(c.get("members", [])) for c in ordered]
+    return {
+        "zid": zid,
+        "bidToPid": bid_to_pid,
+        "lastVoteTimestamp": getattr(conv, "last_updated", None),
+    }
+
+
+def derive_ptptstats(conv: Any, zid: int) -> Dict[str, Any]:
+    """Derive the prep-ptpt-stats blob (conv_man.clj:90-94).
+
+    ptptstats is a secondary consumer (scoped "replace", not fidelity-critical
+    like math_main / math_bidtopid).  We wrap the conversation's public
+    ``participant_info`` under the same envelope keys Clojure uses.
+    """
+    return {
+        "zid": zid,
+        "ptptstats": getattr(conv, "participant_info", {}) or {},
+        "lastVoteTimestamp": getattr(conv, "last_updated", None),
+    }
+
+
+class MathWriter:
+    """Writes a computed conversation's results to Postgres for one cycle."""
+
+    def __init__(self, pg_client: Any):
+        self._pg = pg_client
+
+    def write_conv_updates(self, zid: int, conv: Any) -> int:
+        """Mint one math_tick and write all three data tables with it.
+
+        Returns the math_tick used (handy for logging / tests).
+        """
+        math_tick = self._pg.increment_math_tick(zid)
+
+        data = conv.to_dict()
+        last_vote_timestamp = data.get("lastVoteTimestamp")
+        if last_vote_timestamp is None:
+            last_vote_timestamp = getattr(conv, "last_updated", None)
+
+        # 1. math_main — client-facing PCA/cluster/repness blob (fidelity-critical)
+        self._pg.write_math_main(
+            zid,
+            data,
+            last_vote_timestamp=last_vote_timestamp,
+            math_tick=math_tick,
+        )
+        # 2. math_bidtopid — server bid->pid mapping (fidelity-critical)
+        self._pg.write_math_bidtopid(
+            zid, data=derive_bidtopid(conv, zid), math_tick=math_tick
+        )
+        # 3. math_ptptstats — participant stats
+        self._pg.write_participant_stats(
+            zid, data=derive_ptptstats(conv, zid), math_tick=math_tick
+        )
+
+        logger.info(
+            "Wrote math results for zid=%s math_tick=%s (main+bidtopid+ptptstats)",
+            zid,
+            math_tick,
+        )
+        return math_tick
+
+
+def dump_error(
+    zid: int,
+    conv: Any,
+    coalesced: Any,
+    error: BaseException,
+    dump_dir: str,
+) -> str:
+    """Dump conversation state + failing batch + traceback to an errorconv JSON.
+
+    Mirrors Clojure's conv-update-dump on failure (conv_man.clj:319-323): a
+    debugging artefact written before the batch is retried / the zid is parked.
+
+    Returns the path written (best-effort; never raises).
+    """
+    try:
+        os.makedirs(dump_dir, exist_ok=True)
+        # ms + short uuid so back-to-back dumps within the same millisecond never
+        # collide (the retry-then-park path dumps twice in quick succession).
+        stamp = f"{int(time.time() * 1000)}-{uuid.uuid4().hex[:8]}"
+        path = os.path.join(dump_dir, f"errorconv-zid{zid}-{stamp}.json")
+        try:
+            conv_dump = conv.to_dict() if conv is not None else None
+        except Exception:  # pragma: no cover - defensive
+            conv_dump = {"_dump_error": "conv.to_dict() failed"}
+        payload = {
+            "zid": zid,
+            "error": str(error),
+            "traceback": "".join(
+                traceback.format_exception(type(error), error, error.__traceback__)
+            ),
+            "batch": {
+                "votes": getattr(coalesced, "votes", None),
+                "moderation": getattr(coalesced, "moderation", None),
+            },
+            "conv": conv_dump,
+        }
+        with open(path, "w") as fh:
+            json.dump(payload, fh, default=str)
+        logger.error("Dumped failed conversation state for zid=%s to %s", zid, path)
+        return path
+    except Exception:  # pragma: no cover - dump must never mask the real error
+        logger.exception("Unable to write errorconv dump for zid=%s", zid)
+        return ""