fix(homeostatic): the fold UPDATE runs in each backend's own dialect (#736)

mcp_server/handlers/consolidation/homeostatic_apply.py

@@ -178,19 +178,7 @@ def _apply_fold(store: MemoryStore, domain: str, factor: float, cls: str) -> int
             cls,
         )
         return 0
-    with store.acquire_batch() as conn:
-        result = conn.execute(
-            "UPDATE memories "
-            "SET heat_base = LEAST(1.0, GREATEST(0.0, heat_base * %s)), "
-            "    heat_base_set_at = NOW() "
-            "WHERE domain = %s "
-            "  AND write_class = %s "
-            "  AND NOT is_protected "
-            "  AND NOT no_decay "
-            "  AND NOT is_stale",
-            (float(factor), domain or "", cls),
-        )
-        rows = int(getattr(result, "rowcount", 0) or 0)
+    rows = store.fold_heat_base(domain, cls, float(factor))
     store.set_homeostatic_factor(domain, 1.0, write_class=cls)
     try:
         store.log_homeostatic_fold(domain, cls, factor, rows)

mcp_server/infrastructure/pg_store_heat.py

@@ -52,6 +52,28 @@ def bump_heat_raw(self, memory_id: int, new_heat_base: float) -> None:
         )
         self._conn.commit()
 
+    def fold_heat_base(self, domain: str, write_class: str, factor: float) -> int:
+        """Homeostatic fold: multiply heat_base by ``factor`` for one
+        (domain, write_class), clamped to [0, 1]. Returns rows written.
+
+        Skips protected, no-decay and stale rows. SQLite twin:
+        ``SqliteMemoryStore.fold_heat_base`` (no LEAST/GREATEST/NOW()).
+
+        source: ADR-0363"""
+        with self.acquire_batch() as conn:
+            result = conn.execute(
+                "UPDATE memories "
+                "SET heat_base = LEAST(1.0, GREATEST(0.0, heat_base * %s)), "
+                "    heat_base_set_at = NOW() "
+                "WHERE domain = %s "
+                "  AND write_class = %s "
+                "  AND NOT is_protected "
+                "  AND NOT no_decay "
+                "  AND NOT is_stale",
+                (float(factor), domain or "", write_class),
+            )
+            return int(getattr(result, "rowcount", 0) or 0)
+
     def get_homeostatic_factor(self, domain: str, write_class: str = "auto") -> float:
         """A3: fetch per-(domain, write_class) homeostatic factor, default 1.0.
 

mcp_server/infrastructure/pg_store_host.py

@@ -96,6 +96,10 @@ def acquire_interactive(
             self,
         ) -> AbstractContextManager[psycopg.Connection[DictRow]]: ...
 
+        def acquire_batch(
+            self,
+        ) -> AbstractContextManager[psycopg.Connection[DictRow]]: ...
+
         def _execute(
             self,
             query: str | sql.Composable,

mcp_server/infrastructure/sqlite_store.py

@@ -532,6 +532,26 @@ def bump_heat_raw(self, memory_id: int, new_heat_base: float) -> None:
         )
         self._conn.commit()
 
+    def fold_heat_base(self, domain: str, write_class: str, factor: float) -> int:
+        """Homeostatic fold, SQLite dialect: scalar ``MIN``/``MAX`` stand in
+        for ``LEAST``/``GREATEST`` and ``CURRENT_TIMESTAMP`` for ``NOW()``.
+        See ``PgHeatMixin.fold_heat_base``; returns rows written.
+
+        source: ADR-0363"""
+        cur = self._conn.execute(
+            "UPDATE memories "
+            "SET heat_base = MIN(1.0, MAX(0.0, heat_base * ?)), "
+            "    heat_base_set_at = CURRENT_TIMESTAMP "
+            "WHERE domain = ? "
+            "  AND write_class = ? "
+            "  AND NOT COALESCE(is_protected, 0) "
+            "  AND NOT COALESCE(no_decay, 0) "
+            "  AND NOT COALESCE(is_stale, 0)",
+            (float(factor), domain or "", write_class),
+        )
+        self._conn.commit()
+        return int(cur.rowcount or 0)
+
     def get_homeostatic_factor(self, domain: str, write_class: str = "auto") -> float:
         """SQLite parity: default 1.0 when no row exists.
 

tests_py/handlers/consolidation/test_homeostatic_stratification.py

@@ -64,17 +64,9 @@ def _stub_store(factor_return: float = 3.0) -> MagicMock:
     store.set_homeostatic_factor.return_value = None
     store.log_homeostatic_fold.return_value = 1
 
-    conn_result = MagicMock()
-    conn_result.rowcount = 40  # only the auto rows match the fold predicate
-    conn = MagicMock()
-    conn.execute.return_value = conn_result
-
-    batch_cm = MagicMock()
-    batch_cm.__enter__ = MagicMock(return_value=conn)
-    batch_cm.__exit__ = MagicMock(return_value=False)
-    store.acquire_batch.return_value = batch_cm
-
-    store._test_conn = conn
+    # only the auto rows match the fold predicate (#736: the fold UPDATE is
+    # the store's fold_heat_base, one per regulated class)
+    store.fold_heat_base.return_value = 40
     return store
 
 
@@ -149,14 +141,11 @@ def test_auto_folds_deliberate_untouched(self):
         assert by_class[DELIBERATE]["memories_scanned"] == 40
         assert by_class[DELIBERATE]["mean_heat"] == 0.45
 
-        # The fold UPDATE was issued exactly once (for the auto class),
-        # with a source predicate — not a bare domain predicate. We can't
-        # inspect the SQL string's bound params directly through the
-        # MagicMock's .execute call args without over-fitting to
-        # implementation, so we assert on the observable outcome instead:
-        # rowcount reported back matches only the auto population (stub
-        # configured to 40 — the auto count), not 80 (auto+deliberate).
-        store._test_conn.execute.assert_called_once()
+        # The fold was issued exactly once, for the auto class only — never
+        # for deliberate. The rowcount reported back matches only the auto
+        # population (stub configured to 40), not 80 (auto+deliberate).
+        store.fold_heat_base.assert_called_once()
+        assert store.fold_heat_base.call_args.args[1] == AUTO
         assert by_class[AUTO]["rows_folded"] == 40
 
     def test_deliberate_only_corpus_never_regulated(self):

tests_py/handlers/test_a3_homeostatic_scalar.py

@@ -133,9 +133,9 @@ def test_fold_triggered_writes_batch_and_resets_factor(self):
 
         assert result["scaling_kind"] == "fold"
         assert result["scaling_applied"] is True
-        # Phase 5: fold UPDATE runs on the batch pool's checked-out connection.
-        store.acquire_batch.assert_called_once()
-        store._test_conn.execute.assert_called_once()
+        # The fold UPDATE is the store's own, in its backend's dialect (#736).
+        store.fold_heat_base.assert_called_once()
+        assert store.fold_heat_base.call_args.args[1] == "auto"
         # Post-fold, factor reset to 1.0.
         args, _ = store.set_homeostatic_factor.call_args
         assert args[1] == pytest.approx(1.0)

tests_py/infrastructure/test_sqlite_homeostatic_fold_736.py

@@ -0,0 +1,72 @@
+"""The homeostatic fold runs on the SQLite backend (issue #736).
+
+``_apply_fold`` issued ``LEAST``/``GREATEST``, which SQLite does not have, so
+every fold cycle raised ``OperationalError`` on the default backend. The fold
+now goes through ``fold_heat_base`` on each store, in that backend's dialect.
+
+source: ADR-0363
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from mcp_server.handlers.consolidation import homeostatic_apply
+from mcp_server.infrastructure.sqlite_store import SqliteMemoryStore
+from mcp_server.shared import write_class
+
+_DOMAIN = "fold-domain"
+
+
+@pytest.fixture
+def store(tmp_path):
+    s = SqliteMemoryStore(db_path=str(tmp_path / "fold.db"))
+    yield s
+    s.close()
+
+
+def _insert(store, heat_base: float, **flags) -> int:
+    mid = store.insert_memory(
+        {
+            "content": f"row {heat_base} {sorted(flags)}",
+            "domain": _DOMAIN,
+            "heat": heat_base,
+            "heat_base": heat_base,
+            "write_class": write_class.AUTO,
+        }
+    )
+    for column in flags:
+        store._conn.execute(f"UPDATE memories SET {column} = 1 WHERE id = ?", (mid,))
+    store._conn.commit()
+    return mid
+
+
+def _heat(store, mid: int) -> float:
+    row = store._conn.execute(
+        "SELECT heat_base FROM memories WHERE id = ?", (mid,)
+    ).fetchone()
+    return float(row["heat_base"])
+
+
+def test_fold_scales_clamps_and_skips_exempt_rows(store):
+    mid_scaled = _insert(store, 0.2)
+    mid_high = _insert(store, 0.8)
+    mid_low = _insert(store, 0.2)
+    mid_protected = _insert(store, 0.3, is_protected=True)
+    mid_stale = _insert(store, 0.3, is_stale=True)
+
+    folded = homeostatic_apply._apply_fold(store, _DOMAIN, 2.5, write_class.AUTO)
+
+    assert folded == 3
+    assert _heat(store, mid_scaled) == pytest.approx(0.5)
+    assert _heat(store, mid_high) == pytest.approx(1.0)  # 2.0 clamped to 1.0
+    assert _heat(store, mid_low) == pytest.approx(0.5)
+    assert _heat(store, mid_protected) == pytest.approx(0.3)
+    assert _heat(store, mid_stale) == pytest.approx(0.3)
+    assert store.get_homeostatic_factor(_DOMAIN, write_class.AUTO) == 1.0
+
+
+def test_fold_clamps_to_zero_floor(store):
+    mid = _insert(store, 0.4)
+    homeostatic_apply._apply_fold(store, _DOMAIN, 0.0, write_class.AUTO)
+    assert _heat(store, mid) == 0.0

tests_py/invariants/test_I2_canonical_writer.py

@@ -43,15 +43,16 @@
     # PostgreSQL single-row writer.
     ("infrastructure/pg_store_heat.py", 49),
     # PostgreSQL batched writer.
-    ("infrastructure/pg_store_heat.py", 121),
+    ("infrastructure/pg_store_heat.py", 143),
     # SQLite anchor transfer.
     ("infrastructure/sqlite_store.py", 502),
     # SQLite single-row writer.
     ("infrastructure/sqlite_store.py", 529),
     # SQLite batched writer.
-    ("infrastructure/sqlite_store.py", 586),
-    # Homeostatic fold.
-    ("handlers/consolidation/homeostatic_apply.py", 184),
+    ("infrastructure/sqlite_store.py", 606),
+    # Homeostatic fold, one per backend dialect (issue #736).
+    ("infrastructure/pg_store_heat.py", 66),
+    ("infrastructure/sqlite_store.py", 543),
     # Anchor pin.
     ("handlers/anchor.py", 145),
     # Atomic cue boost: the project-scoped file cues moved from the hook in