fix(data): an integer share_of part never rounds above its whole

share_of promises PART <= WHOLE by construction; integer rounding broke it
in two places (Symbotic SYM_09222008_74E_sch.DC_THROUGHPUT, profiled
read-only 2026-09-28: 256 of 3,333 rows with CASES_PROCESSED > ENERGY_KWH).

Engine: whole * frac stored as fmt "int" rounds a 2-decimal whole at frac
near 1.0 above itself (3852.68 -> 3853). _share_of_value floors an int part
only when rounding lands above a non-negative whole.

Gate: the part_exceeds_whole clamp LEAST(part, whole) into NUMBER(38,0) is
stored rounded half away from zero, back above the whole, so the re-profile
recorded repair_failed for a defect it could fix. The INFORMATION_SCHEMA
query now returns the declared type (bare DATA_TYPE is "NUMBER" for both
NUMBER(38,0) and NUMBER(16,2)), and _repair_statements(column_types=) clamps
an integer part to FLOOR(whole) when the whole has decimals. Decimal parts
keep the exact clamp.

Read-only check on the Symbotic table: of 256 rows over, the old clamp
leaves 256 and the FLOOR clamp leaves 0. tests/test_share_of_integer_rounding.py
(21); both end-to-end tests fail on the pre-fix code.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

demoprep_app/dataset/engine.py

@@ -417,8 +417,8 @@ class BlueprintEngine:
                     frac = min(max(low + (high - low) * position, 0.0), 1.0)
                     values[m.name] = self._store(
                         m,
-                        _format_value(
-                            values.get(m.share_of, 0.0) * frac, m.fmt, pct_max=_pct_bound(m)
+                        _share_of_value(
+                            values.get(m.share_of, 0.0), frac, m.fmt, pct_max=_pct_bound(m)
                         ),
                     )
                     continue
@@ -712,6 +712,21 @@ def _format_value(value: float, fmt: str, clamp: bool = True, pct_max: float = 1
     return round(value, 2)
 
 
+def _share_of_value(whole: float, frac: float, fmt: str, pct_max: float = 100.0) -> Any:
+    """A share_of part for storage: `frac` of its whole, never above it.
+
+    frac <= 1 keeps the raw value under the whole, but rounding an INTEGER
+    part can still overshoot a whole with decimals: 3852.68 x 0.9999 rounds
+    to 3853 (Symbotic SYM_09222008_74E_sch: 256 of 3,333 rows with more
+    cases than kWh). Floor in exactly that case; everywhere else the rounded
+    value is the faithful one.
+    """
+    part = _format_value(whole * frac, fmt, pct_max=pct_max)
+    if fmt == "int" and 0 <= whole < part:
+        part = math.floor(whole * frac)
+    return part
+
+
 def _pct_bound(measure: BlueprintMeasure) -> float:
     """Infer whether a pct measure is on a 0-1 or 0-100 scale from its params.
 

demoprep_app/integrations/snowflake/quality_gate.py

@@ -129,6 +129,25 @@ def is_date_column(data_type: str) -> bool:
     return data_type.upper().split("(")[0].strip() in _DATE_DATA_TYPES
 
 
+_INTEGER_DATA_TYPES = {"INT", "INTEGER", "BIGINT", "SMALLINT", "TINYINT", "BYTEINT"}
+_FIXED_POINT_DATA_TYPES = {"NUMBER", "DECIMAL", "NUMERIC"}
+
+
+def is_integer_type(data_type: str) -> bool:
+    """True when the column stores whole numbers only: NUMBER(p,0) or an INT alias.
+
+    Snowflake's NUMBER with no scale (or NUMBER(p)) is scale 0.
+    """
+    base, _, args = data_type.upper().partition("(")
+    base = base.strip()
+    if base in _INTEGER_DATA_TYPES:
+        return True
+    if base not in _FIXED_POINT_DATA_TYPES:
+        return False
+    precision_scale = args.rstrip(") ").split(",")
+    return len(precision_scale) < 2 or int(precision_scale[1]) == 0
+
+
 # Tolerance when comparing MIN/MAX signatures to shortlist mirrored columns.
 _MIRROR_EPS = 1e-9
 
@@ -563,10 +582,13 @@ def _repair_statements(
     columns: list[str],
     formulas: Mapping[str, str],
     *,
+    column_types: Mapping[str, str],
     allow_clamp: bool = True,
 ) -> list[tuple[str, str]]:
     """(description, SQL) repairs for one defect, most faithful first.
 
+    column_types: {column: declared type} for this table, e.g. "NUMBER(38,0)".
+
     Every statement here targets a table this run created seconds earlier in a
     throwaway demo schema. Nothing outside the generated schema is touched.
     """
@@ -587,9 +609,19 @@ def _repair_statements(
 
     if defect["defect"] == "part_exceeds_whole":
         whole = _quote_ident(defect["whole"])
+        clamp_to, clamp_text = whole, defect["whole"]
+        # An integer column stores LEAST(3853, 3852.68) rounded back up to 3853,
+        # so the clamp never holds and the re-profile records repair_failed
+        # (Symbotic SYM_09222008_74E_sch). Clamp an integer part to the whole's
+        # FLOOR. Only there: a decimal part, such as a 0-1 share, keeps the
+        # whole's exact value.
+        if is_integer_type(column_types[column]) and not is_integer_type(
+            column_types[defect["whole"]]
+        ):
+            clamp_to, clamp_text = f"FLOOR({whole})", f"FLOOR({defect['whole']})"
         repairs = [(
-            f"clamped {table_name}.{column} to {defect['whole']} so the part cannot exceed the whole",
-            f"UPDATE {ref} SET {q} = LEAST({q}, {whole}) WHERE {q} > {whole}",
+            f"clamped {table_name}.{column} to {clamp_text} so the part cannot exceed the whole",
+            f"UPDATE {ref} SET {q} = LEAST({q}, {clamp_to}) WHERE {q} > {whole}",
         )]
         ratio_formula = formulas.get(f"{table_name}.{defect.get('ratio_column', '')}")
         if ratio_formula:
@@ -1072,9 +1104,15 @@ def run_measure_quality_gate(
 
     cursor = connection.cursor()
     try:
+        # DATA_TYPE alone is "NUMBER" for NUMBER(38,0) and NUMBER(16,2) alike.
+        # Spell out the declared type: the part/whole repair must know which
+        # columns hold integers.
         cursor.execute(
             """
-            SELECT TABLE_NAME, COLUMN_NAME, DATA_TYPE
+            SELECT TABLE_NAME, COLUMN_NAME,
+                   IFF(DATA_TYPE = 'NUMBER',
+                       'NUMBER(' || NUMERIC_PRECISION || ',' || NUMERIC_SCALE || ')',
+                       DATA_TYPE)
             FROM INFORMATION_SCHEMA.COLUMNS
             WHERE TABLE_SCHEMA = %s
             ORDER BY TABLE_NAME, ORDINAL_POSITION
@@ -1089,7 +1127,9 @@ def run_measure_quality_gate(
             )
 
         measures_by_table: dict[str, list[str]] = {}
+        types_by_table: dict[str, dict[str, str]] = {}
         for table_name, column_name, data_type in rows:
+            types_by_table.setdefault(table_name, {})[column_name] = data_type or ""
             if is_measure_column(column_name, data_type or ""):
                 measures_by_table.setdefault(table_name, []).append(column_name)
 
@@ -1182,6 +1222,7 @@ def run_measure_quality_gate(
                 statements = _repair_statements(
                     schema_name, table_name, defect,
                     measures_by_table.get(table_name, []), formulas,
+                    column_types=types_by_table[table_name],
                     allow_clamp=allow_clamp,
                 )
                 if not statements and not allow_clamp:

sprint_2026_04.md

@@ -1408,10 +1408,15 @@ composite: a nested ratio has no bound of its own, so only a declared `share_of`
   RX_PLAN_ATTAINMENT, SPEND_VS_PLAN_PCT, OVERCOUNT_INFLATION_RATIO, PRICE_VS_BENCHMARK_RATE.
   Fixing them means changing `measure_names` (`_RATE`/`_RATIO` = 0-1, ATTAINMENT/COVERAGE
   as proportions), which also moves range bounds.
-- [ ] **Integer rounding breaks `share_of`'s guarantee** (separate defect, not changed
-  here). Engine: `_format_value(whole * frac, "int")` can round a part above a
-  non-integer whole. Gate: `LEAST(part, whole)` into an INT column rounds back up, so
-  a genuine int-over-float repair can never be proved.
+- [x] **Integer rounding breaks `share_of`'s guarantee** ✅ Oct 7, branch
+  `fix/share-of-integer-rounding` off test-mcp `cbacde6` (uncommitted, not pushed).
+  Engine `_share_of_value` floors an int part only when rounding lands above its whole.
+  The gate's INFORMATION_SCHEMA query now returns the declared type (`NUMBER(38,0)`;
+  bare `DATA_TYPE` is `NUMBER` for both), and `_repair_statements(column_types=)`
+  clamps an integer part to `FLOOR(whole)` when the whole has decimals. Decimal parts
+  are unchanged. Read-only check on Symbotic DC_THROUGHPUT: 256 rows over; the old clamp
+  leaves 256, the FLOOR clamp leaves 0. `tests/test_share_of_integer_rounding.py` (21);
+  both end-to-end tests fail on pre-fix code (15/2,000 rows over; `repair_failed`).
 - [ ] Symbotic's schema is already rewritten (508 rows clamped; 1-day time travel,
   unrecoverable). A correct board needs a rebuild from the brief.
 - [ ] Not handled (pre-existing): a proportion written `CURRENT / PRIOR - 1`, and a
@@ -1422,5 +1427,5 @@ composite: a nested ratio has no bound of its own, so only a declared `share_of`
 
 ---
 
-*Last updated: September 28, 2026: per-unit rates no longer read as part/whole
-(fix/per-unit-rate-not-part-whole, uncommitted, not deployed)*
+*Last updated: October 7, 2026: an integer share_of part can no longer round above its
+whole (fix/share-of-integer-rounding, uncommitted, not deployed)*

tests/test_share_of_integer_rounding.py

@@ -0,0 +1,185 @@
+"""An integer part never exceeds its whole.
+
+`share_of` promises PART <= WHOLE by construction, and integer rounding broke
+that promise in two places (profiled read-only 2026-09-28 on Symbotic,
+DEMOBUILD_2026_09.SYM_09222008_74E_sch.DC_THROUGHPUT):
+
+  1. Engine: the part was stored as `_format_value(whole * frac, "int")`. A
+     2-decimal whole at frac near 1.0 rounds up past itself: 3852.68 -> 3853.
+     256 of 3,333 rows had CASES_PROCESSED > ENERGY_KWH this way.
+  2. Gate: the part_exceeds_whole repair ran `SET part = LEAST(part, whole)`.
+     Into a NUMBER(38,0) part Snowflake stores 3852.68 rounded half away from
+     zero, back to 3853. The re-profile still violated and the run recorded
+     `repair_failed`, a send-blocking finding for a defect it could fix.
+
+The fix floors only where rounding can overshoot: an integer part of a whole
+with decimals. A decimal part (a 0-1 share of a rate) is never floored.
+
+Run with: pytest tests/test_share_of_integer_rounding.py -v
+"""
+
+import math
+
+import pytest
+
+from demoprep_app.dataset.engine import BlueprintEngine, _share_of_value
+from demoprep_app.integrations.snowflake.quality_gate import (
+    _repair_statements,
+    is_integer_type,
+    run_measure_quality_gate,
+)
+from tests.test_blueprint_measure_hardening import _blueprint
+from tests.test_quality_gate import FakeConnection, FakeCursor, profile_row
+
+
+# --- engine -------------------------------------------------------------------
+
+
+def test_an_integer_part_of_a_decimal_whole_is_floored_only_when_rounding_overshoots():
+    # The Symbotic row: rounding would store 3853 cases against 3852.68 kWh.
+    assert _share_of_value(3852.68, 1.0, "int") == 3852
+    assert _share_of_value(100.5, 1.0, "int") == 100
+    # Rounding UP is kept whenever the result still fits under the whole.
+    assert _share_of_value(100.0, 0.996, "int") == 100
+    assert _share_of_value(3852.68, 0.9, "int") == round(3852.68 * 0.9)
+
+
+def test_a_decimal_part_keeps_its_rounding():
+    assert _share_of_value(3852.68, 0.5, "float") == 1926.34
+    assert _share_of_value(0.8765, 1.0, "pct", pct_max=1.0) == 0.8765
+
+
+_UTILIZATION = [
+    {"name": "AVAILABLE_HOURS", "fmt": "float", "params": {"mean": 400, "sigma": 0.3}},
+    {"name": "UTILIZED_HOURS", "fmt": "int", "share_of": "AVAILABLE_HOURS",
+     "params": {"low": 0.97, "high": 1.0}},
+    {"name": "UTILIZATION_RATE", "kind": "derived", "fmt": "pct",
+     "formula": "UTILIZED_HOURS / AVAILABLE_HOURS"},
+]
+
+
+def test_generated_integer_parts_never_exceed_a_decimal_whole():
+    rows = BlueprintEngine().generate(_blueprint(_UTILIZATION), 2000).table_map()["LOAD_TRANSACTION"].rows
+    over = [r for r in rows if r["UTILIZED_HOURS"] > r["AVAILABLE_HOURS"]]
+    assert not over, f"{len(over)} of {len(rows)} rows have UTILIZED_HOURS > AVAILABLE_HOURS"
+    assert all(isinstance(r["UTILIZED_HOURS"], int) for r in rows)
+    assert max(r["UTILIZATION_RATE"] for r in rows) <= 1.0
+
+
+# --- gate: column types -------------------------------------------------------
+
+
+@pytest.mark.parametrize("data_type, expected", [
+    ("NUMBER(38,0)", True),
+    ("number(38, 0)", True),
+    ("NUMBER", True),          # Snowflake default scale is 0
+    ("NUMBER(10)", True),
+    ("INTEGER", True),
+    ("BIGINT", True),
+    ("NUMBER(16,2)", False),
+    ("DECIMAL(12,4)", False),
+    ("FLOAT", False),
+    ("DOUBLE", False),
+    ("VARCHAR(100)", False),
+    ("DATE", False),
+])
+def test_is_integer_type(data_type, expected):
+    assert is_integer_type(data_type) is expected
+
+
+def _clamp_sql(part_type, whole_type):
+    defect = {"table": "FACT_UTIL", "column": "UTILIZED_HOURS", "defect": "part_exceeds_whole",
+              "whole": "AVAILABLE_HOURS", "detail": "", "ratio_column": "UTILIZATION_RATE"}
+    statements = _repair_statements(
+        "UTIL_sch", "FACT_UTIL", defect, ["UTILIZED_HOURS", "AVAILABLE_HOURS"], {},
+        column_types={"UTILIZED_HOURS": part_type, "AVAILABLE_HOURS": whole_type},
+    )
+    return statements[0][1]
+
+
+def test_an_integer_part_is_clamped_to_the_floor_of_a_decimal_whole():
+    sql = _clamp_sql("NUMBER(38,0)", "NUMBER(16,2)")
+    assert 'SET "UTILIZED_HOURS" = LEAST("UTILIZED_HOURS", FLOOR("AVAILABLE_HOURS"))' in sql
+    # Only rows that actually violate are touched.
+    assert sql.endswith('WHERE "UTILIZED_HOURS" > "AVAILABLE_HOURS"')
+
+
+@pytest.mark.parametrize("part_type, whole_type", [
+    ("NUMBER(16,2)", "NUMBER(16,2)"),   # a decimal share: FLOOR would destroy it
+    ("NUMBER(16,2)", "NUMBER(38,0)"),
+    ("FLOAT", "NUMBER(16,2)"),
+    ("NUMBER(38,0)", "NUMBER(38,0)"),   # integer whole: nothing to round
+])
+def test_the_clamp_is_exact_everywhere_else(part_type, whole_type):
+    sql = _clamp_sql(part_type, whole_type)
+    assert "FLOOR" not in sql
+    assert 'LEAST("UTILIZED_HOURS", "AVAILABLE_HOURS")' in sql
+
+
+# --- gate: end to end ---------------------------------------------------------
+
+
+class _StoringCursor(FakeCursor):
+    """Holds (UTILIZED_HOURS, AVAILABLE_HOURS) values and stores the clamp the way
+    Snowflake does: rounded half away from zero into the NUMBER(38,0) part.
+
+    Snowflake behaviour checked read-only 2026-10-07:
+    LEAST(3853, 3852.68)::NUMBER(38,0) = 3853, LEAST(3853, FLOOR(3852.68)) = 3852.
+    """
+
+    def __init__(self, *args, values, **kwargs):
+        super().__init__(*args, **kwargs)
+        self.values = [list(v) for v in values]
+
+    def execute(self, sql, params=None):
+        if sql.startswith("SELECT SUM(IFF("):
+            self.executed.append(sql)
+            self._last_result = (sum(1 for part, whole in self.values if part > whole),)
+            return
+        if sql.startswith('UPDATE') and 'SET "UTILIZED_HOURS" = LEAST(' in sql:
+            self.executed.append(sql)
+            floored = 'FLOOR("AVAILABLE_HOURS")' in sql
+            for v in self.values:
+                if v[0] > v[1]:
+                    clamped = min(v[0], math.floor(v[1]) if floored else v[1])
+                    v[0] = math.floor(clamped + 0.5)
+            return
+        super().execute(sql, params)
+
+
+def test_the_gate_proves_its_clamp_of_an_integer_part():
+    stats = [
+        ("UTILIZED_HOURS", {"non_zero": 3, "min": 101.0, "max": 3853.0, "avg": 1358.0}),
+        ("AVAILABLE_HOURS", {"non_zero": 3, "min": 100.5, "max": 3852.68, "avg": 1361.14}),
+        ("UTILIZATION_RATE", {"non_zero": 3, "min": 0.92, "max": 1.0, "avg": 0.97}),
+    ]
+    row = profile_row(3, stats)
+    cursor = _StoringCursor(
+        [
+            ("FACT_UTIL", "UTILIZED_HOURS", "NUMBER(38,0)"),
+            ("FACT_UTIL", "AVAILABLE_HOURS", "NUMBER(16,2)"),
+            ("FACT_UTIL", "UTILIZATION_RATE", "NUMBER(16,2)"),
+        ],
+        profiles={"FACT_UTIL": [row, row]},
+        values=[(3853, 3852.68), (101, 100.5), (120, 130.25)],
+    )
+
+    profile = run_measure_quality_gate(
+        FakeConnection(cursor), "UTIL_sch",
+        formulas={"FACT_UTIL.UTILIZATION_RATE": "UTILIZED_HOURS / AVAILABLE_HOURS"},
+        shares={"FACT_UTIL.UTILIZED_HOURS": "AVAILABLE_HOURS"},
+    )
+
+    assert [d["column"] for d in profile["defects"] if d["defect"] == "part_exceeds_whole"] == [
+        "UTILIZED_HOURS"
+    ]
+    assert profile["repair_failures"] == []
+    assert not [f for f in profile["findings"] if f["code"] == "repair_failed"]
+    assert cursor.values == [[3852, 3852.68], [100, 100.5], [120, 130.25]]
+    assert profile["repairs"][0].startswith(
+        "clamped FACT_UTIL.UTILIZED_HOURS to FLOOR(AVAILABLE_HOURS)"
+    )
+    # The gate must read declared types: bare DATA_TYPE says "NUMBER" for both,
+    # which would read the 2-decimal whole as an integer and skip the FLOOR.
+    info_sql = next(s for s in cursor.executed if "INFORMATION_SCHEMA.COLUMNS" in s)
+    assert "NUMERIC_SCALE" in info_sql