fix(ingestion): do not overwrite approved observations by default

automated_ingestion/ocotillo/loader.py

@@ -89,12 +89,27 @@ def load_observations(
     release_status: str,
     batch_size: int = DEFAULT_BATCH_SIZE,
     data_maturity: str = DEFAULT_DATA_MATURITY,
+    overwrite_approved: bool = False,
 ) -> LoadResult:
     """Upsert observations, committing per batch.
 
     ``records`` are ``ObservationRecord`` values from an adapter; resolving a
     source's point identifier to a deployment belongs to reference-data
     bootstrapping, not here, so the caller supplies the ids.
+
+    ``overwrite_approved`` guards data somebody has already reviewed. By default
+    a row whose ``data_maturity`` is ``approved`` is left alone: the upsert exists
+    so a vendor correction can revise *our* provisional readings, not so a
+    re-fetch can quietly replace Bureau-approved history with a vendor's numbers
+    and downgrade it to provisional.
+
+    This is not hypothetical. Fourteen of the thirty-eight San Acacia wells
+    already hold 542,161 approved observations from the AMPAPI transfer, running
+    to August 2022. A Mode A backfill over that window would have overwritten
+    every one of them.
+
+    Setting it to True is a deliberate act: it says the incoming data is better
+    than what was reviewed, which is a judgement a person should make.
     """
     from sqlalchemy.dialects.postgresql import insert
 
@@ -121,17 +136,26 @@ def load_observations(
         # DO UPDATE rather than DO NOTHING: a vendor may correct a reading, and
         # a correction arriving as a no-op would leave the old value in place
         # while the run reported success.
-        statement = statement.on_conflict_do_update(
-            index_elements=[
+        conflict_kwargs: dict[str, Any] = {
+            "index_elements": [
                 "deployment_id",
                 "parameter_id",
                 "observation_datetime",
             ],
-            set_={
+            "set_": {
                 "value": statement.excluded.value,
                 "data_maturity": statement.excluded.data_maturity,
             },
-        )
+        }
+        if not overwrite_approved:
+            # IS DISTINCT FROM rather than != so NULL maturity still updates:
+            # a row with no recorded status has not been reviewed, and treating
+            # unknown as approved would freeze 394,086 legacy rows against every
+            # future correction.
+            conflict_kwargs["where"] = table.c.data_maturity.is_distinct_from(
+                "approved"
+            )
+        statement = statement.on_conflict_do_update(**conflict_kwargs)
         session.execute(statement)
         session.commit()
 

docs/automated-ingestion-pipeline-plan.md

@@ -325,6 +325,20 @@ Built. Migration `a1b2c3d4e5f6`, loader in `automated_ingestion/ocotillo/loader.
 
 ⬜ Run the duplicate report against production and staging before applying the migration. The local development database was clean — 0 duplicate groups in 88,666 rows — which is encouraging and not evidence about production.
 
+### Existing San Acacia data — measured 2026-08-19
+
+Ocotillo already holds transducer observations for **14 of the 38 wells**: 542,161 rows from the AMPAPI transfer, running 2016-07-08 to **2022-08-03**. They carry a real QC status, so `data_maturity` backfilled them as `approved`.
+
+Consequences, all of which the earlier plan assumed away:
+
+- **The watermark starts at 2022-08-03 for those 14**, not the 2015 floor, so a normal run fetches a four-year gap rather than a decade. The other 24 wells do start at the floor.
+- **A backfill would have overwritten them.** The upsert's `DO UPDATE` was written for vendor corrections to our own provisional readings; against approved AMPAPI history it would have replaced 542,161 reviewed values with vendor numbers *and* downgraded them to provisional. `load_observations` now refuses to touch an `approved` row unless `overwrite_approved=True` is passed deliberately.
+- **A datum comparison is still owed.** Those rows came from AMPAPI under whatever convention that pipeline used; ours are Diver-HUB ground-surface centimetres converted to feet. Before any window overlapping 2016–2022 is loaded, a few coinciding timestamps should be compared. Same failure shape as the `WaterLevelReference` question: plausible numbers, wrong meaning.
+
+Rows with `NULL` maturity still update. Unknown is not approved, and treating it as such would freeze the 394,086 legacy rows that have no QC record against every future correction.
+
+**The wider table**, for context: 2,180,989 approved, 7,351 provisional, 394,086 NULL. The NULL cohort is 176 deployments on a single parameter spanning 2016 to February 2025 with no AMPAPI provenance at all — a separate network, and **none of the 38 San Acacia wells are in it**. Worth identifying independently of this work.
+
 ### 3.5 — Watermark from Postgres
 
 Built. `automated_ingestion/shared/watermark.py`, seven tests.

tests/test_transducer_loader.py

@@ -190,4 +190,82 @@ def test_maturity_must_be_a_lexicon_term(loader_target):
         session.rollback()
 
 
+def test_approved_rows_are_not_overwritten(loader_target):
+    # 14 of the 38 San Acacia wells already hold 542,161 approved observations
+    # from the AMPAPI transfer. A backfill over that window must not replace
+    # reviewed values with a vendor's numbers.
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        load_observations(
+            session,
+            _records(1, value=10.0),
+            deployment_id,
+            parameter_id,
+            "draft",
+            data_maturity="approved",
+        )
+        load_observations(
+            session, _records(1, value=99.0), deployment_id, parameter_id, "draft"
+        )
+        row = session.execute(
+            select(
+                TransducerObservation.value, TransducerObservation.data_maturity
+            ).where(TransducerObservation.deployment_id == deployment_id)
+        ).one()
+        assert (row.value, row.data_maturity) == (10.0, "approved")
+
+
+def test_approved_rows_can_be_overwritten_deliberately(loader_target):
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        load_observations(
+            session,
+            _records(1, value=10.0),
+            deployment_id,
+            parameter_id,
+            "draft",
+            data_maturity="approved",
+        )
+        load_observations(
+            session,
+            _records(1, value=99.0),
+            deployment_id,
+            parameter_id,
+            "draft",
+            overwrite_approved=True,
+        )
+        value = session.scalar(
+            select(TransducerObservation.value).where(
+                TransducerObservation.deployment_id == deployment_id
+            )
+        )
+        assert value == 99.0
+
+
+def test_rows_with_no_recorded_maturity_still_update(loader_target):
+    # 394,086 legacy rows have NULL maturity. Unknown is not approved, and
+    # treating it as such would freeze them against every future correction.
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        load_observations(
+            session, _records(1, value=10.0), deployment_id, parameter_id, "draft"
+        )
+        session.execute(
+            TransducerObservation.__table__.update()
+            .where(TransducerObservation.deployment_id == deployment_id)
+            .values(data_maturity=None)
+        )
+        session.commit()
+
+        load_observations(
+            session, _records(1, value=99.0), deployment_id, parameter_id, "draft"
+        )
+        value = session.scalar(
+            select(TransducerObservation.value).where(
+                TransducerObservation.deployment_id == deployment_id
+            )
+        )
+        assert value == 99.0
+
+
 # ============= EOF =============================================