fix: restore cleanup invariants

Author: 1012839419a-alt

backend/control/measurements.py

@@ -27,9 +27,11 @@ class SourceMeasurement(BaseModel):
     duplicate_rate: float
 
     freshness_lag_seconds: int | None = None
+    cursor_advanced: bool
 
     odp_stream_lag: int | None = None
     odp_pending: int | None = None
+    dlq_count: int = 0
 
     #: {ErrorKind.value: count} — PR-Control-3. Reuses the SAME taxonomy the
     #: persisted backend.models.source_measurement.SourceMeasurement.error_kinds

tests/integration/test_control_api.py

@@ -3,7 +3,7 @@
 rows, POST /control/outcomes/evaluate, and GET /control/advisory-report.
 """
 
-from datetime import datetime, timedelta
+from datetime import UTC, datetime, timedelta
 
 import pytest
 from sqlalchemy import select
@@ -25,7 +25,7 @@ async def _seed_measurement_row(session, source_id: str, **overrides):
     kwargs = dict(
         source_id=source_id,
         run_id="row-run-1",
-        measured_at=datetime(2026, 7, 2, tzinfo=datetime.UTC),
+        measured_at=datetime(2026, 7, 2, tzinfo=UTC),
         accepted=0,
         duplicates=0,
         rejected=1,
@@ -154,7 +154,7 @@ async def test_evaluate_and_advisory_report_recovery_flow(
             .all()
         )
         assert rows
-        backdated = datetime.now(datetime.UTC) - timedelta(seconds=120)
+        backdated = datetime.now(UTC) - timedelta(seconds=120)
         for row in rows:
             row.created_at = backdated
         await db_session.commit()
@@ -166,7 +166,7 @@ async def test_evaluate_and_advisory_report_recovery_flow(
             db_session,
             source_id,
             run_id="row-run-post",
-            measured_at=datetime.now(datetime.UTC),
+            measured_at=datetime.now(UTC),
             accepted=5,
             duplicates=0,
             rejected=0,
@@ -299,7 +299,7 @@ async def test_kill_switch_engaged_short_circuits_control_cycle(
                         executed=False,
                         measurement_before={},
                         outcome="recovered" if i < 8 else "persisted",
-                        evaluated_at=datetime.now(datetime.UTC),
+                        evaluated_at=datetime.now(UTC),
                     )
                 )
         await db_session.commit()
@@ -309,7 +309,7 @@ async def test_kill_switch_engaged_short_circuits_control_cycle(
 
         from backend.control.cycle import run_control_cycle_once
 
-        result = await run_control_cycle_once(db_session, now=datetime.now(datetime.UTC))
+        result = await run_control_cycle_once(db_session, now=datetime.now(UTC))
         await db_session.commit()
 
         assert result.executions == []

tests/integration/test_sources_api.py

@@ -357,7 +357,7 @@ async def test_control_state_with_run_evidence(client, db_session, sample_source
     monkeypatch.delenv("ODP_DATABASE_URL", raising=False)
     monkeypatch.delenv("ODP_INGEST_URL", raising=False)
 
-    from datetime import datetime
+    from datetime import UTC, datetime
 
     from backend.models.task import CollectionTask, TaskRun, TaskRunEvent
 
@@ -371,7 +371,7 @@ async def test_control_state_with_run_evidence(client, db_session, sample_source
     run = TaskRun(
         task_id=task.id,
         status="completed",
-        finished_at=datetime(2026, 7, 2, tzinfo=datetime.UTC),
+        finished_at=datetime(2026, 7, 2, tzinfo=UTC),
         duration_ms=1000,
         records_collected=9,
     )
@@ -427,7 +427,7 @@ async def test_control_state_degraded_when_errors(
     monkeypatch.delenv("ODP_DATABASE_URL", raising=False)
     monkeypatch.delenv("ODP_INGEST_URL", raising=False)
 
-    from datetime import datetime
+    from datetime import UTC, datetime
 
     from backend.models.task import CollectionTask, TaskRun, TaskRunEvent
 
@@ -441,7 +441,7 @@ async def test_control_state_degraded_when_errors(
     run = TaskRun(
         task_id=task.id,
         status="completed",
-        finished_at=datetime(2026, 7, 2, tzinfo=datetime.UTC),
+        finished_at=datetime(2026, 7, 2, tzinfo=UTC),
         duration_ms=1000,
         records_collected=1,
     )
@@ -485,7 +485,7 @@ async def test_control_state_trend_fallback_for_pre_measurement_source(
     monkeypatch.delenv("ODP_DATABASE_URL", raising=False)
     monkeypatch.delenv("ODP_INGEST_URL", raising=False)
 
-    from datetime import datetime
+    from datetime import UTC, datetime
 
     from backend.models.task import CollectionTask, TaskRun, TaskRunEvent
 
@@ -501,8 +501,8 @@ async def test_control_state_trend_fallback_for_pre_measurement_source(
         run = TaskRun(
             task_id=task.id,
             status="completed",
-            created_at=datetime(2026, 7, day, tzinfo=datetime.UTC),
-            finished_at=datetime(2026, 7, day, tzinfo=datetime.UTC),
+            created_at=datetime(2026, 7, day, tzinfo=UTC),
+            finished_at=datetime(2026, 7, day, tzinfo=UTC),
             duration_ms=1000,
             records_collected=stored,
         )
@@ -546,14 +546,14 @@ async def test_control_state_trend_fallback_for_pre_measurement_source(
 
 
 async def _seed_measurement_row(session, source_id: str, **overrides):
-    from datetime import datetime
+    from datetime import UTC, datetime
 
     from backend.models.source_measurement import SourceMeasurement as SourceMeasurementRow
 
     kwargs = dict(
         source_id=source_id,
         run_id="row-run-1",
-        measured_at=datetime(2026, 7, 2, tzinfo=datetime.UTC),
+        measured_at=datetime(2026, 7, 2, tzinfo=UTC),
         accepted=5,
         duplicates=0,
         rejected=0,
@@ -930,14 +930,14 @@ async def test_control_state_classification_flips_with_objective_override(
     create_resp = await client.post("/api/v1/sources", json=sample_source_data)
     source_id = create_resp.json()["data"]["id"]
 
-    from datetime import datetime
+    from datetime import UTC, datetime
 
     from backend.models.source_measurement import SourceMeasurement as SourceMeasurementRow
 
     row = SourceMeasurementRow(
         source_id=source_id,
         run_id="row-run-1",
-        measured_at=datetime(2026, 7, 2, tzinfo=datetime.UTC),
+        measured_at=datetime(2026, 7, 2, tzinfo=UTC),
         accepted=97,
         duplicates=0,
         rejected=3,