fix(tradein/newbuilding): счётчики записи различают вставку и обновление (#2807)

`newbuilding_enrich_backfill` считал свою работу разницей COUNT(*) до и после
сохранения — то есть приростом ЧИСЛА СТРОК, а не числом записанных точек. Вставка в
houses_price_dynamics идёт ON CONFLICT DO UPDATE, поэтому переписывание существующей
точки давало ноль.

Прод 10.08: прогон 3578 отчитался price_dynamics_rows=0, обновив за своё окно 64 строки
по 10 домам — те самые, что вставил прогон 3563 накануне (у него в counters стояло 64).
Ноль читался как «динамика цен снова не пишется».

Соседние два счётчика в том же месте врали по своим причинам: reliability_rows обнулял
_dedup_reliability, схлопывающий строку сразу после вставки, а review_rows игнорировал
число, которое _save_cian_reviews уже возвращал, в пользу разницы COUNT'ов.

Различать вставку и обновление умеет только сам писатель: RETURNING (xmax = 0) —
идиома, уже применяемая в репозитории (services/scrapers/gisogd66.py). Поэтому
save_newbuilding_enrichment возвращает NewbuildingSaveCounts, а вызывающий перестал
восстанавливать число по таблице. Остальные три вызова результат игнорируют.

Ключи counters переименованы (price_dynamics_inserted/_updated, reliability_inserted,
review_upserted): у старых имён в истории прогонов другой смысл, и молча поменять его
под тем же именем — ровно тот дефект, ради которого правка и делается.

Сторож нулевого результата не затронут: mark_backfill_finished судит по
attempted/enriched/gone/blocked, а не по числам записи, — на это добавлен тест.

tradein-mvp/backend/app/api/v1/admin.py

@@ -1961,8 +1961,9 @@ async def scrape_cian_newbuilding(
 
     saved = False
     if house_id is not None:
-        # save_newbuilding_enrichment — sync (def, returns None); await на sync-функции
-        # раньше поднимал TypeError на любом вызове с house_id.
+        # save_newbuilding_enrichment — sync (def, не корутина); await на sync-функции
+        # раньше поднимал TypeError на любом вызове с house_id. Возвращаемый счёт
+        # записанного (#2807) этой ручке не нужен — она отвечает фактом сохранения.
         save_newbuilding_enrichment(db, house_id, enrichment)
         saved = True
 

tradein-mvp/backend/app/tasks/newbuilding_enrich_backfill.py

@@ -109,10 +109,17 @@ class NewbuildingEnrichBackfillResult:
     failed_fetch: int = 0  # fetch returned None / raised
     failed_save: int = 0  # save raised after a good fetch
 
-    # Row-level deltas (how much actually landed).
-    price_dynamics_rows: int = 0
-    reliability_rows: int = 0
-    review_rows: int = 0
+    # Сколько РЕАЛЬНО записано, по словам самих писателей (#2807). Раньше здесь стоял
+    # прирост COUNT(*) по таблице до/после сохранения — то есть «выросла ли таблица», а
+    # не «сколько записали»: при ON CONFLICT DO UPDATE обновление даёт ноль, а у
+    # reliability ноль давал ещё и _dedup_reliability, схлопывающий дубль сразу после
+    # вставки. Ключи переименованы намеренно: у price_dynamics_rows/reliability_rows/
+    # review_rows в истории прогонов старый смысл, и молча поменять его под тем же
+    # именем — ровно тот дефект, ради которого правка и делается.
+    price_dynamics_inserted: int = 0  # новых точек динамики цен
+    price_dynamics_updated: int = 0  # существующих точек переписано свежей ценой
+    reliability_inserted: int = 0  # строк house_reliability_checks вставлено
+    review_upserted: int = 0  # отзывов записано (вставка+обновление, ключ ext_review_id)
 
     duration_sec: float = field(default=0.0)
 
@@ -557,15 +564,15 @@ async def backfill_newbuilding_enrichment(
             continue
 
         # ── Save under a SAVEPOINT so one bad house can't poison the batch ──
-        # begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally,
-        # so we snapshot the row counts BEFORE and recompute the delta AFTER its commit
-        # rather than relying on the nested transaction staying open.
-        pd_before, rc_before, rv_before = _house_enrichment_counts(db, house_id)
+        # begin_nested() = SAVEPOINT; save_newbuilding_enrichment commits internally.
+        # COUNT(*) до сохранения нужен ТОЛЬКО для had_reliability (дедуп ниже): сколько
+        # записано, теперь сообщают сами писатели, а не разница COUNT'ов (#2807).
+        _, rc_before, _ = _house_enrichment_counts(db, house_id)
         try:
             had_reliability = rc_before > 0
 
             # 1) price_dynamics + reliability + houses UPDATE (existing, commits inside).
-            save_newbuilding_enrichment(db, house_id, enrichment)
+            saved = save_newbuilding_enrichment(db, house_id, enrichment)
 
             # 2) reviews — added here (save_newbuilding_enrichment skips them).
             #    SAVEPOINT around the review write so a malformed review can't lose the
@@ -601,16 +608,18 @@ async def backfill_newbuilding_enrichment(
                     sp.rollback()
                     logger.warning("reliability dedup failed house_id=%s: %s", house_id, dexc)
 
-            pd_after, rc_after, rv_after = _house_enrichment_counts(db, house_id)
-            result.price_dynamics_rows += max(0, pd_after - pd_before)
-            result.reliability_rows += max(0, rc_after - rc_before)
-            result.review_rows += max(0, rv_after - rv_before)
+            result.price_dynamics_inserted += saved.price_inserted
+            result.price_dynamics_updated += saved.price_updated
+            result.reliability_inserted += saved.reliability_inserted
+            result.review_upserted += review_written
             result.succeeded += 1
             logger.info(
-                "enriched house_id=%s: +pd=%d +reliability=%d +reviews=%d (parsed reviews=%d)",
+                "enriched house_id=%s: динамика цен +%d новых / %d обновлено, "
+                "reliability +%d, отзывов записано %d (распознано %d)",
                 house_id,
-                max(0, pd_after - pd_before),
-                max(0, rc_after - rc_before),
+                saved.price_inserted,
+                saved.price_updated,
+                saved.reliability_inserted,
                 review_written,
                 len(enrichment.reviews),
             )
@@ -629,8 +638,8 @@ async def backfill_newbuilding_enrichment(
     result.duration_sec = time.time() - t0
     logger.info(
         "newbuilding-enrich backfill done: processed=%d ok=%d skip=%d resolved=%d "
-        "resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
-        "| %.1fs",
+        "resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
+        "%d обновлено, reliability +%d, отзывов %d | %.1fs",
         result.processed,
         result.succeeded,
         result.skipped_already_enriched,
@@ -638,9 +647,10 @@ async def backfill_newbuilding_enrichment(
         result.failed_resolve,
         result.failed_fetch,
         result.failed_save,
-        result.price_dynamics_rows,
-        result.reliability_rows,
-        result.review_rows,
+        result.price_dynamics_inserted,
+        result.price_dynamics_updated,
+        result.reliability_inserted,
+        result.review_upserted,
         result.duration_sec,
     )
     return result
@@ -787,8 +797,8 @@ async def run_newbuilding_enrich(
         )
         logger.info(
             "scheduler: newbuilding_enrich run_id=%d finished — processed=%d ok=%d skip=%d "
-            "resolve_fail=%d fetch_fail=%d save_fail=%d | rows pd=%d reliability=%d reviews=%d "
-            "| pending=%d %.1fs",
+            "resolve_fail=%d fetch_fail=%d save_fail=%d | записано: динамика +%d новых / "
+            "%d обновлено, reliability +%d, отзывов %d | pending=%d %.1fs",
             run_id,
             result.processed,
             result.succeeded,
@@ -796,9 +806,10 @@ async def run_newbuilding_enrich(
             result.failed_resolve,
             result.failed_fetch,
             result.failed_save,
-            result.price_dynamics_rows,
-            result.reliability_rows,
-            result.review_rows,
+            result.price_dynamics_inserted,
+            result.price_dynamics_updated,
+            result.reliability_inserted,
+            result.review_upserted,
             result.cian_houses_pending,
             result.duration_sec,
         )

tradein-mvp/backend/tests/tasks/test_newbuilding_enrich_backfill.py

@@ -19,7 +19,10 @@ _wp_mock = MagicMock()
 sys.modules.setdefault("weasyprint", _wp_mock)
 
 import pytest  # noqa: E402
-from scraper_kit.providers.cian.newbuilding import NewbuildingEnrichment  # noqa: E402
+from scraper_kit.providers.cian.newbuilding import (  # noqa: E402
+    NewbuildingEnrichment,
+    NewbuildingSaveCounts,
+)
 
 from app.tasks.newbuilding_enrich_backfill import (  # noqa: E402
     NewbuildingEnrichBackfillResult,
@@ -168,24 +171,37 @@ def _enrichment_with_everything(seed: int = 0) -> NewbuildingEnrichment:
 
 
 def _fake_save_newbuilding_enrichment(db, house_id, enrichment):
-    """Stand-in for the real saver: lands price_dynamics + reliability into FakeDB."""
+    """Stand-in for the real saver: lands price_dynamics + reliability into FakeDB.
+
+    Возвращает NewbuildingSaveCounts, как настоящий (#2807): вставку от обновления
+    различает сам писатель — снаружи по таблице их не отличить (UPSERT по dim_key).
+    """
+    inserted = updated = 0
     for p in enrichment.realty_valuation_chart:
         if p.get("price_per_sqm") is None:
             continue
-        db.price_dynamics.add(
-            (
-                house_id,
-                p["month_date"],
-                "cian_realty_valuation",
-                p.get("room_count", "all"),
-                p.get("prices_type", "price"),
-                p.get("period", "halfYear"),
-            )
+        key = (
+            house_id,
+            p["month_date"],
+            "cian_realty_valuation",
+            p.get("room_count", "all"),
+            p.get("prices_type", "price"),
+            p.get("period", "halfYear"),
         )
+        if key in db.price_dynamics:
+            updated += 1
+        else:
+            inserted += 1
+        db.price_dynamics.add(key)
+    reliability = 0
     for c in enrichment.reliability_checks:
         if c.get("check_name") or c.get("check_status"):
             db.reliability.append((house_id, "cian_nashdom"))
+            reliability += 1
     db.commit()
+    return NewbuildingSaveCounts(
+        price_inserted=inserted, price_updated=updated, reliability_inserted=reliability
+    )
 
 
 # ---------------------------------------------------------------------------
@@ -279,9 +295,10 @@ async def test_backfill_populates_all_three_tables() -> None:
     assert len(db.price_dynamics) == 2  # 1 chart point × 2 houses
     assert len(db.reliability) == 2
     assert len(db.reviews) == 4  # 2 reviews × 2 houses
-    assert result.price_dynamics_rows == 2
-    assert result.reliability_rows == 2
-    assert result.review_rows == 4
+    assert result.price_dynamics_inserted == 2
+    assert result.price_dynamics_updated == 0
+    assert result.reliability_inserted == 2
+    assert result.review_upserted == 4
 
 
 @pytest.mark.asyncio
@@ -544,7 +561,7 @@ async def test_run_wrapper_marks_done_and_passes_params(monkeypatch: pytest.Monk
     async def _fake_backfill(_db, *, limit, force, request_delay_sec, on_progress=None):
         # on_progress — сигнал живости внутрь цикла (#2725); здесь только принимаем.
         seen.update(limit=limit, force=force, request_delay_sec=request_delay_sec)
-        return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_rows=2)
+        return NewbuildingEnrichBackfillResult(processed=3, succeeded=2, price_dynamics_inserted=2)
 
     monkeypatch.setattr(task_mod, "backfill_newbuilding_enrichment", _fake_backfill)
     monkeypatch.setattr(task_mod.runs_mod, "update_heartbeat", lambda *a, **k: None)

tradein-mvp/backend/tests/test_2767_newbuilding_parse_miss.py

@@ -185,7 +185,7 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
     _stub_backfill(
         monkeypatch,
         NewbuildingEnrichBackfillResult(
-            processed=10, succeeded=3, failed_fetch=6, failed_resolve=1, price_dynamics_rows=7
+            processed=10, succeeded=3, failed_fetch=6, failed_resolve=1, price_dynamics_inserted=7
         ),
     )
     calls = _stub_finalisers(monkeypatch)
@@ -196,5 +196,5 @@ async def test_partial_success_keeps_rich_counters(monkeypatch) -> None:
     assert counters["attempted"] == 10
     assert counters["enriched"] == 3
     assert counters["failed"] == 7
-    assert counters["price_dynamics_rows"] == 7  # исходные счётчики на месте
+    assert counters["price_dynamics_inserted"] == 7  # исходные счётчики на месте
     assert counters["succeeded"] == 3

tradein-mvp/backend/tests/test_2807_write_counters_honesty.py

@@ -0,0 +1,193 @@
+"""#2807: счётчик мерил прирост таблицы, а читался как «сколько записали».
+
+`newbuilding_enrich_backfill` считал свою работу разницей `COUNT(*)` до и после
+сохранения. Вставка в houses_price_dynamics идёт `ON CONFLICT … DO UPDATE`, поэтому
+обновление существующей точки давало ноль. Прод 10.08: прогон 3578 отчитался
+`price_dynamics_rows: 0`, обновив за своё окно **64 строки по 10 домам** — те самые,
+что вставил прогон 3563 накануне (у него в тех же counters стояло 64). Ноль читался
+как «динамика цен снова не пишется».
+
+Соседние счётчики врали в том же месте по своим причинам: `reliability_rows` обнулял
+`_dedup_reliability`, схлопывающий строку сразу после вставки, а `review_rows`
+игнорировал число, которое `_save_cian_reviews` УЖЕ возвращал, в пользу разницы COUNT'ов.
+
+Фальсификация (см. прогон в PR): на коде до правки `test_second_pass_reports_updates`
+даёт `price_dynamics_rows == 0` при 64 переписанных точках — ровно прод-симптом.
+
+Отдельно проверяется, что правка НЕ ослабила сторожа нулевого результата: он смотрит на
+`attempted`/`enriched`/`gone`/`blocked` (#2695), а не на счётчики записи, и прогон,
+который ничего не обогатил, обязан остаться 'failed' при любых числах в `*_written`.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from unittest.mock import MagicMock, patch
+
+os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://test:test@localhost:5432/test")
+sys.modules.setdefault("weasyprint", MagicMock())
+
+from scraper_kit.providers.cian.newbuilding import (  # noqa: E402
+    NewbuildingSaveCounts,
+    save_newbuilding_enrichment,
+)
+
+from app.services import scrape_runs as runs_mod  # noqa: E402
+from app.tasks.newbuilding_enrich_backfill import (  # noqa: E402
+    NewbuildingEnrichBackfillResult,
+)
+
+# Прод-масштаб прогона 3578: 10 домов × 64/10 точек. Держим ровно 64, чтобы число в
+# тесте совпадало с числом в задаче.
+PROD_POINTS = 64
+
+
+class _UpsertDB:
+    """Сессия, у которой houses_price_dynamics уже населена (второй проход).
+
+    `RETURNING (xmax = 0)` возвращает False на конфликте — это и есть «обновили».
+    """
+
+    def __init__(self, *, already_present: bool) -> None:
+        self.already_present = already_present
+        self.price_writes = 0
+        self.reliability_writes = 0
+        self.committed = False
+
+    def execute(self, statement, params=None):
+        sql = str(statement)
+        res = MagicMock()
+        if "INSERT INTO houses_price_dynamics" in sql:
+            self.price_writes += 1
+            assert "RETURNING (xmax = 0)" in sql, "писатель обязан различать вставку и update"
+            res.fetchone.return_value = (not self.already_present,)
+            return res
+        if "INSERT INTO house_reliability_checks" in sql:
+            self.reliability_writes += 1
+        res.fetchone.return_value = None
+        return res
+
+    def commit(self) -> None:
+        self.committed = True
+
+
+def _enrichment(points: int):
+    from scraper_kit.providers.cian.newbuilding import NewbuildingEnrichment
+
+    return NewbuildingEnrichment(
+        cian_internal_house_id=1,
+        cian_zhk_url="https://zhk-x.cian.ru/",
+        name="ЖК Тест",
+        realty_valuation_chart=[
+            {
+                "month_date": f"2026-{(i % 12) + 1:02d}-01",
+                "room_count": "all",
+                "prices_type": "price",
+                "period": "halfYear",
+                "price_per_sqm": 150000.0 + i,
+            }
+            for i in range(points)
+        ],
+        reliability_checks=[{"check_name": "Надёжный застройщик", "check_status": "reliable"}],
+        reviews=[],
+    )
+
+
+# ── 1. Писатель различает вставку и обновление ───────────────────────────────
+
+
+def test_first_pass_reports_inserts() -> None:
+    db = _UpsertDB(already_present=False)
+    counts = save_newbuilding_enrichment(db, 42, _enrichment(PROD_POINTS))
+    assert counts.price_inserted == PROD_POINTS
+    assert counts.price_updated == 0
+    assert counts.reliability_inserted == 1
+    assert db.price_writes == PROD_POINTS
+
+
+def test_second_pass_reports_updates() -> None:
+    """Прод-симптом: те же 64 точки, ничего нового — но записаны все 64.
+
+    До правки этот прогон отчитывался нулём по всем трём счётчикам.
+    """
+    db = _UpsertDB(already_present=True)
+    counts = save_newbuilding_enrichment(db, 42, _enrichment(PROD_POINTS))
+    assert counts.price_inserted == 0
+    assert counts.price_updated == PROD_POINTS
+    assert counts.price_written == PROD_POINTS
+    assert db.price_writes == PROD_POINTS
+
+
+def test_nothing_to_write_stays_zero() -> None:
+    """Встречная проверка: пустой график — ноль и во «вставлено», и в «обновлено»."""
+    db = _UpsertDB(already_present=True)
+    counts = save_newbuilding_enrichment(db, 42, _enrichment(0))
+    assert (counts.price_inserted, counts.price_updated, counts.price_written) == (0, 0, 0)
+
+
+def test_points_without_price_are_not_counted_as_written() -> None:
+    """Точка без price_per_sqm пропускается писателем — и не попадает в счёт."""
+    enrichment = _enrichment(2)
+    enrichment.realty_valuation_chart[0]["price_per_sqm"] = None
+    db = _UpsertDB(already_present=False)
+    counts = save_newbuilding_enrichment(db, 42, enrichment)
+    assert counts.price_written == 1
+    assert db.price_writes == 1
+
+
+# ── 2. Сторож нулевого результата не ослаблен ────────────────────────────────
+
+
+def _finalize(counters: dict[str, int]) -> str:
+    """Прогнать counters через боевой финализатор и вернуть выбранный статус."""
+    chosen: dict[str, str] = {}
+    with (
+        patch.object(runs_mod, "mark_done", lambda *a, **k: chosen.setdefault("s", "done")),
+        patch.object(runs_mod, "mark_failed", lambda *a, **k: chosen.setdefault("s", "failed")),
+        patch.object(runs_mod, "mark_banned", lambda *a, **k: chosen.setdefault("s", "banned")),
+    ):
+        runs_mod.mark_backfill_finished(MagicMock(), 1, counters, source="newbuilding_enrich")
+    return chosen["s"]
+
+
+def test_watchdog_still_fails_a_run_that_enriched_nothing() -> None:
+    """Прогон без обогащений остаётся 'failed', сколько бы записей ни насчитали.
+
+    Числа записи в решение сторожа не входят вовсе — он судит по attempted/enriched.
+    Если бы входили, честный счётчик «обновлено» превратил бы холостой прогон в успех.
+    """
+    result = NewbuildingEnrichBackfillResult(
+        processed=25,
+        succeeded=0,
+        failed_fetch=25,
+        price_dynamics_updated=PROD_POINTS,  # «что-то писали» — но никого не обогатили
+    )
+    assert _finalize(result.to_backfill_counters()) == "failed"
+
+
+def test_watchdog_verdict_ignores_the_new_keys() -> None:
+    """Явно: добавление/убирание новых ключей не двигает вердикт ни в одну сторону."""
+    base = {"attempted": 25, "enriched": 3, "failed": 22}
+    assert _finalize(dict(base)) == "done"
+    assert _finalize({**base, "price_dynamics_inserted": 0, "price_dynamics_updated": 0}) == "done"
+    zero = {"attempted": 25, "enriched": 0, "failed": 25}
+    assert _finalize(dict(zero)) == "failed"
+    assert _finalize({**zero, "price_dynamics_updated": 999}) == "failed"
+
+
+def test_counters_carry_both_numbers_into_the_run() -> None:
+    """В scrape_runs.counters уезжают ОБА числа — ноль одного больше не читается как ноль."""
+    counters = NewbuildingEnrichBackfillResult(
+        processed=25, succeeded=25, price_dynamics_updated=PROD_POINTS
+    ).to_backfill_counters()
+    assert counters["price_dynamics_inserted"] == 0
+    assert counters["price_dynamics_updated"] == PROD_POINTS
+    # Старые имена не должны остаться: у них в истории прогонов другой смысл.
+    assert "price_dynamics_rows" not in counters
+    assert "reliability_rows" not in counters
+    assert "review_rows" not in counters
+
+
+def test_save_counts_written_is_the_sum() -> None:
+    assert NewbuildingSaveCounts(price_inserted=3, price_updated=4).price_written == 7

tradein-mvp/packages/scraper-kit/src/scraper_kit/providers/cian/newbuilding.py

@@ -594,11 +594,38 @@ def _extract_nested_offers(offers_state: dict[str, Any]) -> list[dict[str, Any]]
 # ---- save helpers ----
 
 
+@dataclass(frozen=True)
+class NewbuildingSaveCounts:
+    """Что прогон РЕАЛЬНО записал — вставил и обновил отдельно (#2807).
+
+    Заводится потому, что вызывающий мерил свою работу разницей ``COUNT(*)`` по таблице
+    до и после сохранения. Это прирост ЧИСЛА СТРОК, а не число записанных точек: у
+    houses_price_dynamics вставка идёт ``ON CONFLICT … DO UPDATE``, поэтому обновление
+    уже существующей точки даёт ноль. Прод 10.08: прогон 3578 отчитался
+    ``price_dynamics_rows: 0``, обновив за своё окно 64 строки по 10 домам (их вставил
+    прогон 3563 накануне) — ноль читался как «динамика цен не пишется».
+
+    Единственный, кто знает разницу, — сам писатель: ``RETURNING (xmax = 0)`` отличает
+    вставку от обновления (та же идиома, что в
+    ``backend/app/services/scrapers/gisogd66.py``). Поэтому число возвращается отсюда, а
+    не восстанавливается снаружи по таблице.
+    """
+
+    price_inserted: int = 0
+    price_updated: int = 0
+    reliability_inserted: int = 0
+
+    @property
+    def price_written(self) -> int:
+        """Сколько точек динамики прошло через запись (вставка + обновление)."""
+        return self.price_inserted + self.price_updated
+
+
 def save_newbuilding_enrichment(
     db: Any,
     house_id: int,
     enrichment: NewbuildingEnrichment,
-) -> None:
+) -> NewbuildingSaveCounts:
     """Persist NewbuildingEnrichment to DB.
 
     Steps:
@@ -606,6 +633,9 @@ def save_newbuilding_enrichment(
     2. UPDATE houses with Cian metadata (incl. cian_zhk_url if present)
     3. INSERT INTO houses_price_dynamics (chart points, ON CONFLICT DO UPDATE)
     4. INSERT INTO house_reliability_checks (overall + details)
+
+    Returns NewbuildingSaveCounts — вставлено/обновлено раздельно (#2807). Вызывающие,
+    которым счёт не нужен (SERP-sweep, admin re-enrich), просто игнорируют результат.
     """
     from sqlalchemy import text
 
@@ -681,14 +711,18 @@ def save_newbuilding_enrichment(
     # 3. INSERT houses_price_dynamics
     # UNIQUE constraint: houses_price_dynamics_dim_key
     #   (house_id, source, room_count, prices_type, period, month_date) — per migration 029
-    chart_saved = 0
+    price_inserted = 0
+    price_updated = 0
     for point in enrichment.realty_valuation_chart:
         if point.get("price_per_sqm") is None:
             continue
         room_count = point.get("room_count") or "all"
         prices_type = point.get("prices_type") or "price"
         period = point.get("period") or "halfYear"
-        db.execute(
+        # RETURNING (xmax = 0): у только что вставленной строки xmax равен нулю, у
+        # обновлённой конфликтом — id транзакции. Без этого «вставили» и «обновили»
+        # снаружи неразличимы, и обновление читается как «ничего не записали» (#2807).
+        written = db.execute(
             text("""
                 INSERT INTO houses_price_dynamics (
                     house_id, month_date, source,
@@ -705,6 +739,7 @@ def save_newbuilding_enrichment(
                 ON CONFLICT ON CONSTRAINT houses_price_dynamics_dim_key DO UPDATE SET
                     price_per_sqm = EXCLUDED.price_per_sqm,
                     recorded_at   = NOW()
+                RETURNING (xmax = 0) AS is_insert
             """),
             {
                 "hid": house_id,
@@ -714,12 +749,16 @@ def save_newbuilding_enrichment(
                 "pd": period,
                 "pps": point["price_per_sqm"],
             },
-        )
-        chart_saved += 1
+        ).fetchone()
+        if written is not None and written[0]:
+            price_inserted += 1
+        else:
+            price_updated += 1
 
     # 4. INSERT house_reliability_checks (stores overall check + details array)
     # Schema (025): (house_id, check_status, check_name, details jsonb, source, recorded_at)
     # No UNIQUE constraint — caller should manage duplicates if needed
+    reliability_inserted = 0
     for check in enrichment.reliability_checks:
         if not check.get("check_name") and not check.get("check_status"):
             continue
@@ -743,15 +782,23 @@ def save_newbuilding_enrichment(
                 "det": json.dumps(check.get("details") or [], ensure_ascii=False),
             },
         )
+        reliability_inserted += 1
 
     db.commit()
     logger.info(
-        "Cian newbuilding saved house_id=%s (chart=%d points, reliability=%d checks, mc_id=%s)",
+        "Cian newbuilding saved house_id=%s (chart: +%d new / %d updated, "
+        "reliability=%d checks, mc_id=%s)",
         house_id,
-        chart_saved,
-        len(enrichment.reliability_checks),
+        price_inserted,
+        price_updated,
+        reliability_inserted,
         mc_id,
     )
+    return NewbuildingSaveCounts(
+        price_inserted=price_inserted,
+        price_updated=price_updated,
+        reliability_inserted=reliability_inserted,
+    )
 
 
 async def resolve_cian_zhk_url(