test(tradein): cover avito/cian + mark_failed total_seen/new_count persistence (#1926)

#1928 fixed scrape_runs.total_seen/new_count for sweeps via _column_counts
but only regression-tested the yandex mark_done path. Add equivalents for
the CitySweepCounters (avito/cian) sweep and the mark_failed path — both
explicitly named in the #1926 audit — so the all-sweeps + failure-path
guarantee is locked in.

Refs #1926

tradein-mvp/backend/tests/test_city_sweep.py

@@ -223,6 +223,55 @@ def test_mark_done_yandex_counters_populate_columns() -> None:
     assert params["new_count"] == 42
 
 
+def test_mark_done_avito_cian_counters_populate_columns() -> None:
+    """End-to-end shape: CitySweepCounters.to_dict() (avito/cian) → mark_done → columns set.
+
+    CitySweepCounters обслуживает avito + cian city-sweep'ы и проходит тот же путь
+    counters.to_dict() → mark_done, что и Yandex — фикс #1926 должен заполнять
+    total_seen/new_count и для них (issue: «то же касается CitySweepCounters»).
+    """
+    from app.services.scrape_pipeline import CitySweepCounters
+    from app.services.scrape_runs import mark_done
+
+    counters = CitySweepCounters(
+        anchors_total=4, anchors_done=4, lots_fetched=1234, lots_inserted=56
+    )
+    mock_db = MagicMock()
+    mock_db.execute.return_value.first.return_value = MagicMock(id=1)
+
+    mark_done(mock_db, 1, counters.to_dict())
+
+    params = _captured_params(mock_db)
+    assert params["total_seen"] == 1234
+    assert params["new_count"] == 56
+
+
+def test_mark_failed_persists_total_seen_new_count_columns() -> None:
+    """Даже при mark_failed колонки заполняются из частично собранных counters (#1926).
+
+    Sweep может упасть на середине, успев собрать тысячи лотов — observability должна
+    видеть total_seen>0, а не 0 для failed-прогона.
+    """
+    from app.services.scrape_runs import mark_failed
+
+    mock_db = MagicMock()
+    mock_db.execute.return_value.first.return_value = MagicMock(id=1)
+
+    mark_failed(mock_db, 1, "boom", {"lots_fetched": 300, "lots_inserted": 9})
+
+    # mark_failed после UPDATE дёргает _alert_on_run_id (SELECT source ...), поэтому
+    # последний execute — это alert-SELECT, а не наш UPDATE. Ищем UPDATE-вызов явно.
+    update_call = next(
+        c for c in mock_db.execute.call_args_list if "UPDATE scrape_runs" in str(c.args[0])
+    )
+    params = update_call.args[1]
+    assert params["total_seen"] == 300
+    assert params["new_count"] == 9
+    sql = str(update_call.args[0])
+    assert "total_seen" in sql
+    assert "new_count" in sql
+
+
 # ── FastAPI endpoints (offline) ─────────────────────────────────────────────