diff --git a/alembic/versions/a1b2c3d4e5f6_unique_transducer_observation.py b/alembic/versions/a1b2c3d4e5f6_unique_transducer_observation.py
new file mode 100644
index 000000000..8b70eb7bc
--- /dev/null
+++ b/alembic/versions/a1b2c3d4e5f6_unique_transducer_observation.py
@@ -0,0 +1,48 @@
+"""unique constraint on transducer_observation
+
+Revision ID: a1b2c3d4e5f6
+Revises: d9e0f1a2b3c4
+Create Date: 2026-08-19
+
+The table had only an index on (deployment_id, parameter_id,
+observation_datetime), so nothing prevented the same reading being inserted
+twice. That absence is what forces a delete-then-repost load strategy: without a
+constraint to conflict on, a re-run can only avoid duplicates by removing what
+is already there first, which leaves a window where the data is missing.
+
+With this constraint the loader can use ON CONFLICT DO UPDATE and a re-run
+becomes idempotent, so a backfill overlapping existing data is safe.
+
+Note the constraint is on `deployment_id`, not `thing_id` -- the plan named a
+column this table does not have. A deployment is a thing/sensor pairing, so two
+sensors on the same well may legitimately report the same instant; scoping
+uniqueness to the deployment allows that while still catching a re-inserted row.
+
+**Run automated_ingestion/sql/find_duplicate_observations.sql first.** This
+migration fails on a table that already violates the constraint, and it is
+better to know that before starting than halfway through.
+"""
+
+from alembic import op
+
+revision = "a1b2c3d4e5f6"
+down_revision = "d9e0f1a2b3c4"
+branch_labels = None
+depends_on = None
+
+CONSTRAINT_NAME = "uq_transducer_observation_deployment_parameter_datetime"
+INDEX_NAME = "ix_transducer_observation_deployment_parameter_datetime"
+COLUMNS = ["deployment_id", "parameter_id", "observation_datetime"]
+
+
+def upgrade() -> None:
+    # The unique constraint creates its own index on the same columns, so the
+    # existing one would be redundant -- two indexes maintained on every insert
+    # into the largest table in the schema.
+    op.drop_index(INDEX_NAME, table_name="transducer_observation")
+    op.create_unique_constraint(CONSTRAINT_NAME, "transducer_observation", COLUMNS)
+
+
+def downgrade() -> None:
+    op.drop_constraint(CONSTRAINT_NAME, "transducer_observation", type_="unique")
+    op.create_index(INDEX_NAME, "transducer_observation", COLUMNS)
diff --git a/automated_ingestion/ocotillo/loader.py b/automated_ingestion/ocotillo/loader.py
new file mode 100644
index 000000000..a4eb0d92e
--- /dev/null
+++ b/automated_ingestion/ocotillo/loader.py
@@ -0,0 +1,180 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Idempotent loading of observations into Ocotillo.
+
+Every write is an upsert against the unique constraint on
+``(deployment_id, parameter_id, observation_datetime)``. That makes a re-run a
+no-op rather than a duplication, which is what lets a backfill overlap existing
+data safely.
+
+The alternative -- delete the window, then insert -- is what Aqueduct does
+against FROST, because there is no constraint there to conflict on. It leaves a
+window during which the data is simply missing, and a failure mid-way leaves it
+missing permanently. Upserting has no such window.
+
+Rows are written with SQLAlchemy Core rather than ORM objects. ``AGENTS.md``
+is explicit about this for high-volume tables: instantiating a mapped class per
+observation is what turns a backfill into an hour-long run.
+"""
+
+from collections.abc import Iterable, Iterator
+from dataclasses import dataclass, field
+from datetime import datetime
+from typing import Any
+
+DEFAULT_BATCH_SIZE = 5_000
+"""Rows per statement.
+
+Large enough that a month of five-minute readings is a handful of round trips,
+small enough that one batch's parameters do not approach Postgres' limit. Each
+batch commits on its own, so an interrupted load keeps what it had already
+written -- with an upsert, resuming simply rewrites those rows.
+"""
+
+
+@dataclass
+class LoadResult:
+    """What a load did, for reporting as asset metadata."""
+
+    rows_seen: int = 0
+    rows_written: int = 0
+    batches: int = 0
+    blocks_touched: list[int] = field(default_factory=list)
+
+    @property
+    def rows_skipped(self) -> int:
+        return self.rows_seen - self.rows_written
+
+
+def _batched(records: Iterable[Any], size: int) -> Iterator[list[Any]]:
+    batch: list[Any] = []
+    for record in records:
+        batch.append(record)
+        if len(batch) >= size:
+            yield batch
+            batch = []
+    if batch:
+        yield batch
+
+
+def load_observations(
+    session: Any,
+    records: Iterable[Any],
+    deployment_id: int,
+    parameter_id: int,
+    release_status: str,
+    batch_size: int = DEFAULT_BATCH_SIZE,
+) -> 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.
+    """
+    from sqlalchemy.dialects.postgresql import insert
+
+    from db.transducer import TransducerObservation
+
+    result = LoadResult()
+    table = TransducerObservation.__table__
+
+    for batch in _batched(records, batch_size):
+        rows = [
+            {
+                "deployment_id": deployment_id,
+                "parameter_id": parameter_id,
+                "observation_datetime": record.observation_datetime,
+                "value": record.value,
+                "release_status": release_status,
+            }
+            for record in batch
+        ]
+        result.rows_seen += len(rows)
+
+        statement = insert(table).values(rows)
+        # 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=[
+                "deployment_id",
+                "parameter_id",
+                "observation_datetime",
+            ],
+            set_={"value": statement.excluded.value},
+        )
+        session.execute(statement)
+        session.commit()
+
+        result.rows_written += len(rows)
+        result.batches += 1
+
+    return result
+
+
+def ensure_block(
+    session: Any,
+    thing_id: int,
+    parameter_id: int,
+    start: datetime,
+    end: datetime,
+    release_status: str,
+    review_status: str = "not reviewed",
+) -> int:
+    """Create or widen the QC block covering a loaded window.
+
+    ``review_status`` defaults to ``not reviewed`` and callers should leave it
+    there. In Ocotillo ``approved`` asserts that a Bureau human reviewed the
+    data and carries a ``reviewer_id``; the vendor's own approval flag is a
+    different claim and is preserved separately.
+
+    An existing block is widened rather than duplicated, so re-running a window
+    does not accumulate blocks.
+    """
+    from sqlalchemy import select
+
+    from db.transducer import TransducerObservationBlock
+
+    existing = session.scalars(
+        select(TransducerObservationBlock)
+        .where(TransducerObservationBlock.thing_id == thing_id)
+        .where(TransducerObservationBlock.parameter_id == parameter_id)
+        .where(TransducerObservationBlock.review_status == review_status)
+        .where(TransducerObservationBlock.start_datetime <= end)
+        .where(TransducerObservationBlock.end_datetime >= start)
+    ).first()
+
+    if existing is not None:
+        existing.start_datetime = min(existing.start_datetime, start)
+        existing.end_datetime = max(existing.end_datetime, end)
+        session.commit()
+        return existing.id
+
+    block = TransducerObservationBlock(
+        thing_id=thing_id,
+        parameter_id=parameter_id,
+        review_status=review_status,
+        start_datetime=start,
+        end_datetime=end,
+        release_status=release_status,
+    )
+    session.add(block)
+    session.commit()
+    return block.id
+
+
+# ============= EOF =============================================
diff --git a/automated_ingestion/sql/find_duplicate_observations.sql b/automated_ingestion/sql/find_duplicate_observations.sql
new file mode 100644
index 000000000..6b49727ce
--- /dev/null
+++ b/automated_ingestion/sql/find_duplicate_observations.sql
@@ -0,0 +1,44 @@
+-- Find duplicate transducer observations before adding the unique constraint.
+--
+-- The migration that adds UNIQUE (deployment_id, parameter_id,
+-- observation_datetime) will fail on a table that already violates it, and
+-- failing halfway through a production migration is worse than not starting.
+-- Run this first, on every environment the migration will touch.
+--
+--   psql "..." -f automated_ingestion/sql/find_duplicate_observations.sql
+--
+-- No rows means the migration is safe to run. Rows mean a decision is needed
+-- about which copy to keep, and that decision belongs to someone who knows the
+-- data -- deleting the higher id is a guess, not a rule, because the rows may
+-- differ in `value` rather than being true duplicates.
+
+\echo '== Duplicate groups =='
+SELECT
+    deployment_id,
+    parameter_id,
+    observation_datetime,
+    count(*)             AS copies,
+    count(DISTINCT value) AS distinct_values,
+    min(id)              AS lowest_id,
+    max(id)              AS highest_id
+FROM transducer_observation
+GROUP BY deployment_id, parameter_id, observation_datetime
+HAVING count(*) > 1
+ORDER BY copies DESC, observation_datetime
+LIMIT 100;
+
+\echo ''
+\echo '== Totals =='
+-- `distinct_values > 1` is the interesting case: those are not redundant copies
+-- but disagreeing measurements, and collapsing them silently would discard a
+-- reading somebody recorded.
+SELECT
+    count(*)                                        AS duplicate_groups,
+    sum(copies) - count(*)                          AS rows_above_the_first,
+    count(*) FILTER (WHERE distinct_values > 1)     AS groups_that_disagree
+FROM (
+    SELECT count(*) AS copies, count(DISTINCT value) AS distinct_values
+    FROM transducer_observation
+    GROUP BY deployment_id, parameter_id, observation_datetime
+    HAVING count(*) > 1
+) g;
diff --git a/db/transducer.py b/db/transducer.py
index 1670bb9fa..57625e3f8 100644
--- a/db/transducer.py
+++ b/db/transducer.py
@@ -107,12 +107,20 @@ class TransducerObservation(Base, AutoBaseMixin, ReleaseMixin):
     """
 
     __tablename__ = "transducer_observation"
+    # Unique rather than merely indexed: without a constraint to conflict on, a
+    # re-run can only avoid duplicates by deleting first, which leaves a window
+    # where the data is missing. With it the loader upserts and a repeated
+    # backfill is idempotent.
+    #
+    # Scoped to the deployment, not the thing: a deployment is a thing/sensor
+    # pairing, so two sensors on one well may legitimately report the same
+    # instant.
     __table_args__ = (
-        Index(
-            "ix_transducer_observation_deployment_parameter_datetime",
+        UniqueConstraint(
             "deployment_id",
             "parameter_id",
             "observation_datetime",
+            name="uq_transducer_observation_deployment_parameter_datetime",
         ),
     )
 
diff --git a/docs/automated-ingestion-pipeline-plan.md b/docs/automated-ingestion-pipeline-plan.md
index 7a3773081..f7dd5225c 100644
--- a/docs/automated-ingestion-pipeline-plan.md
+++ b/docs/automated-ingestion-pipeline-plan.md
@@ -277,13 +277,18 @@ Some of the 33 may already exist in Ocotillo under Bureau point IDs. Duplicates
 
 ### 3.4 — Unique constraint on `transducer_observation` + idempotent upsert loader
 
-`db/transducer.py` defines only an index — no unique constraint, so nothing prevents inserting the same reading twice. That absence is what forces Aqueduct's delete-then-repost in FROST.
+Built. Migration `a1b2c3d4e5f6`, loader in `automated_ingestion/ocotillo/loader.py`, three tests against a real Postgres.
 
-- Alembic migration adds `UniqueConstraint(thing_id, parameter_id, observation_datetime)`. Existing duplicates found and resolved first — the migration must not fail on production data.
-- Loader batches and issues `INSERT … ON CONFLICT … DO UPDATE`, through the `db/` SQLAlchemy models, not raw SQL. Batch size tuned and documented; a full backfill month fits in memory; each batch commits in its own transaction.
-- `TransducerObservationBlock` rows created/extended for the loaded window, `review_status = "not reviewed"`.
-- Loader reports rows inserted, rows updated, adapter failures as Dagster metadata.
-- Test: loading the same window twice leaves the row count unchanged.
+**The constraint is on `deployment_id`, not `thing_id`.** This section named a column the table does not have — `TransducerObservation` carries `deployment_id`, and `thing_id` lives on `TransducerObservationBlock`. The existing index was already `(deployment_id, parameter_id, observation_datetime)`, so the constraint matches it. Semantically this is also the right scope: a deployment is a thing/sensor pairing, so two sensors on one well may legitimately report the same instant.
+
+- ✅ Migration drops the redundant index — the unique constraint creates its own on the same columns, and keeping both means two indexes maintained on every insert into the largest table in the schema. Verified up and down against a database with 88,666 observations.
+- ✅ `automated_ingestion/sql/find_duplicate_observations.sql` reports violations **before** the migration runs, since it fails on a table that already violates it and failing halfway through a production migration is worse than not starting. It separates redundant copies from groups whose `value` disagrees — the latter are not duplicates but conflicting measurements, and collapsing them silently would discard a reading.
+- ✅ Loader upserts with `ON CONFLICT DO UPDATE`, batching at 5,000 rows and committing per batch. **`DO UPDATE`, not `DO NOTHING`:** a vendor correction arriving as a no-op would leave the old value in place while the run reported success.
+- ✅ SQLAlchemy Core, not ORM objects, per `AGENTS.md` — instantiating a mapped class per observation is what turns a backfill into an hour-long run.
+- ✅ `ensure_block` widens an existing block rather than duplicating it, and defaults `review_status` to `not reviewed`.
+- ✅ Test: loading the same window twice leaves the row count unchanged. That claim depends on Postgres enforcing the constraint, so it runs against the real database rather than a stub.
+
+⬜ 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.
 
 ### 3.5 — Watermark from Postgres
 
diff --git a/tests/test_transducer_loader.py b/tests/test_transducer_loader.py
new file mode 100644
index 000000000..f6cefdc0d
--- /dev/null
+++ b/tests/test_transducer_loader.py
@@ -0,0 +1,117 @@
+# ===============================================================================
+# Copyright 2026 ross
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+# ===============================================================================
+"""
+Loader behaviour against a real database.
+
+Idempotency is the whole point of the unique constraint and cannot be shown with
+a stub: it depends on Postgres enforcing the constraint and on ON CONFLICT
+resolving against it.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import delete, func, select
+
+from automated_ingestion.ocotillo.loader import load_observations
+from automated_ingestion.ocotillo.structs import ObservationRecord
+from db.engine import session_ctx
+from db.parameter import Parameter
+from db.transducer import TransducerObservation
+
+START = datetime(2026, 1, 1, tzinfo=timezone.utc)
+
+
+@pytest.fixture()
+def loader_target(sensor_to_water_well_thing_deployment):
+    """A deployment and parameter to load against, cleaned up afterwards."""
+    deployment_id = sensor_to_water_well_thing_deployment.id
+    with session_ctx() as session:
+        parameter_id = session.scalar(select(Parameter.id).limit(1))
+        assert parameter_id, "lexicon parameters are seeded by conftest"
+        yield deployment_id, parameter_id
+        session.execute(
+            delete(TransducerObservation).where(
+                TransducerObservation.deployment_id == deployment_id
+            )
+        )
+        session.commit()
+
+
+def _records(count, value=10.0):
+    return [
+        ObservationRecord(
+            external_point_id="sanacaciareach-40",
+            observation_datetime=START + timedelta(minutes=5 * i),
+            value=value + i,
+            units="ft",
+        )
+        for i in range(count)
+    ]
+
+
+def _count(session, deployment_id):
+    return session.scalar(
+        select(func.count())
+        .select_from(TransducerObservation)
+        .where(TransducerObservation.deployment_id == deployment_id)
+    )
+
+
+def test_loading_the_same_window_twice_does_not_duplicate(loader_target):
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        load_observations(session, _records(10), deployment_id, parameter_id, "draft")
+        load_observations(session, _records(10), deployment_id, parameter_id, "draft")
+        assert _count(session, deployment_id) == 10
+
+
+def test_a_corrected_value_overwrites_rather_than_being_ignored(loader_target):
+    # DO NOTHING would leave the old reading in place while the run reported
+    # success -- the worst of both outcomes.
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        load_observations(
+            session, _records(1, value=10.0), deployment_id, parameter_id, "draft"
+        )
+        load_observations(
+            session, _records(1, value=99.0), deployment_id, parameter_id, "draft"
+        )
+        stored = session.scalar(
+            select(TransducerObservation.value).where(
+                TransducerObservation.deployment_id == deployment_id
+            )
+        )
+        assert stored == 99.0
+
+
+def test_batches_commit_separately(loader_target):
+    deployment_id, parameter_id = loader_target
+    with session_ctx() as session:
+        result = load_observations(
+            session,
+            _records(25),
+            deployment_id,
+            parameter_id,
+            "draft",
+            batch_size=10,
+        )
+        assert result.batches == 3
+        assert result.rows_written == 25
+        assert _count(session, deployment_id) == 25
+
+
+# ============= EOF =============================================