feat(transducer): publish and range-delete for corrected hydrographs

The hydrograph corrector in OcotilloUI could only download its corrected
series as CSV -- there was no POST for transducer observations at all. This
adds the two write endpoints its upload contract specifies.

POST /observation/transducer-groundwater-level/block publishes one corrected
logger file as one block plus all of its readings, in one transaction. The
block's span is derived from the measurements rather than sent: nothing links
the observation table to the block table, so the reader pairs them by time and
a client-supplied span wider than the data would make the block claim readings
it does not contain. `deployment_id` is optional and resolved from the
deployments covering that span; zero or more than one match is a 422 rather
than a guess, because guessing attributes readings to hardware that did not
record them.

An existing block sharing any instant with the new one is a 409 listing the
collisions. `?replace_overlapping=true` deletes those blocks and their
readings. The readings have to go with the block -- keeping them would leave
rows the reader cannot show that still occupy the deployment/parameter/instant
the new series is about to claim, so a "replace" that kept them would fail on
the very insert it was asked to make room for. Readings orphaned by a
hand-deleted block are caught separately and reported with the earliest
colliding timestamp, rather than letting the insert abort on a constraint name.

Overlap is inclusive on both bounds, unlike TransducerObservationBlock.overlaps,
which is half-open. The reader matches with `start <= t <= end`, so two blocks
sharing an endpoint both claim a reading at that instant -- exactly the
ambiguity the check exists to prevent.

DELETE /observation/transducer-groundwater-level removes every reading for a
well inside a closed range and reconciles the blocks that covered them: one
left empty is deleted, one left partial has its span narrowed to the survivors.
All three parameters are required; there is deliberately no unbounded form.
Scope matches the GET on the same path, so the set a client previews is the set
this removes.

Schema changes: provenance on the block (source_file, source_kind, and an
ordered corrections list), and a per-reading `note` set only where a correction
moved the value, so NULL reads as "as measured" rather than "unknown". A
corrected block is derived data, and a reviewer who cannot see that a series
was snapped to a manual measurement cannot review it. The block time-order
check is relaxed to `end >= start`: a block covering a single instant is
legitimate, either published that way or narrowed to it by a delete.

Both write routes are gated on AMP.Staging, a standalone group -- AMPAdmin does
not satisfy it and it satisfies nothing else -- so they ship dark while the
workbench is validated against real logger files.

Two bugs fixed in passing:

- The read route called get_transducer_observations positionally, and the
  helper's fourth positional parameter is `sensor_id`. `start_time` landed in
  `sensor_id` (unused, dropped), `end_time` landed in `start_time`, and
  `end_time` was never set, so a requested upper bound was ignored and the
  lower bound came from the wrong argument.
- `sort`/`order` on that route were accepted and ignored. They now work over a
  whitelist; an unrecognised field is a 422 rather than a silently differently
  ordered page.

Wellntel support from the contract is deferred: both the readings proxy and the
sensor_type filter on /thing are blocked on where the API key and the
wellname/PointID mapping should live.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: jirhiker

CLAUDE.md

@@ -164,6 +164,12 @@ Authentik groups granted.
 **Role families are orthogonal**: general `Admin` confers nothing in the AMP or
 Lexicon families. Only tiers *within* a family nest.
 
+**`AMP.Staging`** is a standalone group, not a fourth AMP tier — `AMPAdmin`
+does not satisfy it. It gates the hydrograph corrector's publish and range-delete
+routes while the workbench is being validated against real logger files, so they
+ship dark. Read **`docs/hydrograph-correction-publish.md`** before changing
+them.
+
 **Authorization is opt-in per endpoint** — a `user: <role>_dependency` parameter
 in the signature, not a router-level `dependencies=[...]`. Omitting it produces a
 fully public endpoint with no error. `tests/test_authorization.py` holds the

alembic/versions/c3d4e5f6a7b8_hydrograph_correction_publish.py

@@ -0,0 +1,116 @@
+"""publish provenance for corrected transducer blocks
+
+Revision ID: c3d4e5f6a7b8
+Revises: b2c3d4e5f6a7
+Create Date: 2026-08-19
+
+The hydrograph corrector publishes a *derived* series: water head converted to
+depth below ground surface against manual anchors, then shifted, snapped, and
+drift-corrected. None of those numbers are what the instrument recorded, so the
+database has to carry enough to tell a reviewer what happened to them.
+
+Three columns on the block cover the batch: the file it came from, whether that
+file held water head or depth to water, and the ordered list of corrections
+applied. `comment` already exists and takes the publisher's free-text note.
+
+One column on the observation covers the row: `note`, set only on readings a
+correction actually moved. NULL therefore means "as measured", which is the
+distinction review needs. The legacy `nma_waterlevelscontinuous_*_notes`
+columns cannot serve -- each is scoped to one legacy source table.
+
+The block time-order check is relaxed from `>` to `>=`. A block spanning a
+single instant is legitimate: a published file with one reading, or a block
+narrowed by a range delete until one observation survives. The block reader
+matches observations inclusively on both bounds, so a zero-width block still
+covers its reading. Loosening a check constraint cannot invalidate existing
+rows.
+"""
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision = "c3d4e5f6a7b8"
+down_revision = "b2c3d4e5f6a7"
+branch_labels = None
+depends_on = None
+
+# Spelled as it exists in the database, typo included -- renaming it here would
+# leave deployed environments with a constraint this migration cannot find.
+TIME_ORDER_CONSTRAINT = "check_transuder_block_time_order"
+
+
+def upgrade() -> None:
+    op.add_column(
+        "transducer_observation_block",
+        sa.Column(
+            "source_file",
+            sa.String(length=255),
+            nullable=True,
+            comment="Name of the logger file the corrected series was derived from",
+        ),
+    )
+    op.add_column(
+        "transducer_observation_block",
+        sa.Column(
+            "source_kind",
+            sa.String(length=50),
+            nullable=True,
+            comment="What the source file measured: water_head or depth_to_water",
+        ),
+    )
+    op.add_column(
+        "transducer_observation_block",
+        sa.Column(
+            "corrections",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=True,
+            comment="Corrections applied to the source series, in applied order",
+        ),
+    )
+    op.add_column(
+        "transducer_observation",
+        sa.Column(
+            "note",
+            sa.Text(),
+            nullable=True,
+            comment=(
+                "Per-reading correction annotation; NULL means the value is as "
+                "measured"
+            ),
+        ),
+    )
+
+    op.drop_constraint(
+        TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check"
+    )
+    op.create_check_constraint(
+        TIME_ORDER_CONSTRAINT,
+        "transducer_observation_block",
+        "end_datetime >= start_datetime",
+    )
+
+
+def downgrade() -> None:
+    # Zero-width blocks may have been created while the loosened constraint was
+    # in force, so widen them by a second rather than let the stricter
+    # constraint fail to validate. A one-second span on a block that covered an
+    # instant is a smaller lie than a failed downgrade.
+    op.execute(
+        "UPDATE transducer_observation_block "
+        "SET end_datetime = start_datetime + interval '1 second' "
+        "WHERE end_datetime = start_datetime"
+    )
+    op.drop_constraint(
+        TIME_ORDER_CONSTRAINT, "transducer_observation_block", type_="check"
+    )
+    op.create_check_constraint(
+        TIME_ORDER_CONSTRAINT,
+        "transducer_observation_block",
+        "end_datetime > start_datetime",
+    )
+
+    op.drop_column("transducer_observation", "note")
+    op.drop_column("transducer_observation_block", "corrections")
+    op.drop_column("transducer_observation_block", "source_kind")
+    op.drop_column("transducer_observation_block", "source_file")

api/observation.py

@@ -28,6 +28,7 @@
     session_dependency,
     amp_admin_dependency,
     amp_editor_dependency,
+    amp_staging_dependency,
     amp_viewer_dependency,
 )
 from db import Observation, Parameter
@@ -40,7 +41,12 @@
     UpdateGroundwaterLevelObservation,
     UpdateWaterChemistryObservation,
 )
-from schemas.transducer import TransducerObservationWithBlockResponse
+from schemas.transducer import (
+    DeletedTransducerObservationsResponse,
+    PublishedTransducerBlockResponse,
+    PublishTransducerBlock,
+    TransducerObservationWithBlockResponse,
+)
 from schemas.water_level_csv import WaterLevelBulkUploadResponse
 from services.crud_helper import model_deleter, model_adder
 from services.observation_helper import (
@@ -50,10 +56,30 @@
     get_transducer_observations,
 )
 from services.query_helper import simple_get_by_id
+from services.transducer_helper import (
+    delete_transducer_observations,
+    publish_transducer_block,
+)
 from services.water_level_csv import bulk_upload_water_levels
 
 router = APIRouter(prefix="/observation", tags=["observation"])
 
+
+def _groundwater_level_parameter_id(session) -> int:
+    """
+    The lexicon id the transducer routes work in.
+
+    Looked up rather than configured so the publish, read, and delete routes
+    cannot drift onto different parameters.
+    """
+    return (
+        session.query(Parameter)
+        .filter(Parameter.parameter_name == "groundwater level")
+        .one()
+        .id
+    )
+
+
 """
 TODO
 
@@ -88,6 +114,33 @@ def add_water_chemistry_observation(
     return model_adder(session, Observation, obs_data, user=user)
 
 
+@router.post(
+    "/transducer-groundwater-level/block",
+    status_code=HTTP_201_CREATED,
+    summary="Publish a corrected transducer series as one block",
+)
+def publish_transducer_groundwater_level_block(
+    payload: PublishTransducerBlock,
+    session: session_dependency,
+    user: amp_staging_dependency,
+    replace_overlapping: bool = False,
+) -> PublishedTransducerBlockResponse:
+    """
+    Publish one corrected logger file as a single observation block.
+
+    The block's time span is derived from the measurements; the client does not
+    send it. Overlapping an existing block is a 409 listing the collisions --
+    pass `replace_overlapping=true` to supersede them, which deletes those
+    blocks and their readings in the same transaction.
+
+    Written by the hydrograph corrector in OcotilloUI. See
+    `docs/hydrograph-correction-publish.md`.
+    """
+    return publish_transducer_block(
+        session, payload, user=user, replace_overlapping=replace_overlapping
+    )
+
+
 @router.post(
     "/groundwater-level/bulk-upload",
     response_model=WaterLevelBulkUploadResponse,
@@ -155,17 +208,30 @@ def get_transducer_groundwater_level_observations(
     thing_id: int | None = None,
     start_time: datetime | None = None,
     end_time: datetime | None = None,
+    sort: str | None = None,
+    order: str | None = None,
 ) -> CustomPage[TransducerObservationWithBlockResponse]:
+    """
+    Retrieve transducer groundwater level observations paired with the block
+    that covers them.
 
-    groundwater_parameter_id = (
-        session.query(Parameter)
-        .filter(Parameter.parameter_name == "groundwater level")
-        .one()
-        .id
-    )
-
+    `sort` accepts `observation_datetime`, `value`, or `id`; `order` accepts
+    `asc` or `desc`. The default is newest first, so a client that wants the
+    latest stored reading for a well can ask for size 1.
+    """
+    # Keyword arguments deliberately: the helper's fourth positional parameter
+    # is `sensor_id`, so the previous positional call passed `start_time` as a
+    # sensor id (unused, silently dropped), `end_time` as `start_time`, and
+    # nothing as `end_time` -- an upper bound the caller asked for was ignored
+    # and the lower bound came from the wrong argument.
     return get_transducer_observations(
-        session, thing_id, groundwater_parameter_id, start_time, end_time
+        session,
+        thing_id=thing_id,
+        parameter_id=_groundwater_level_parameter_id(session),
+        start_time=start_time,
+        end_time=end_time,
+        sort=sort,
+        order=order,
     )
 
 
@@ -302,6 +368,40 @@ def get_observation_by_id(
 # DELETE =======================================================================
 
 
+@router.delete(
+    "/transducer-groundwater-level",
+    status_code=HTTP_200_OK,
+    summary="Delete transducer groundwater level observations in a time range",
+)
+def delete_transducer_groundwater_level_observations(
+    session: session_dependency,
+    user: amp_staging_dependency,
+    thing_id: int,
+    start_time: datetime,
+    end_time: datetime,
+) -> DeletedTransducerObservationsResponse:
+    """
+    Delete every transducer groundwater level reading for a well inside a
+    closed time range, and reconcile the blocks that covered them: a block left
+    with no readings is deleted, one left with some has its span narrowed to
+    the survivors.
+
+    All three parameters are required -- there is deliberately no form of this
+    request that deletes everything for a well. Scoped exactly like the `GET`
+    on this path, so the set previewed there is the set removed here.
+
+    Irreversible, and it leaves the `transducer_daily_data` materialized view
+    stale until its next refresh.
+    """
+    return delete_transducer_observations(
+        session,
+        thing_id=thing_id,
+        parameter_id=_groundwater_level_parameter_id(session),
+        start_time=start_time,
+        end_time=end_time,
+    )
+
+
 @router.delete(
     "/{observation_id}",
     summary="Delete an observation",

core/dependencies.py

@@ -59,6 +59,21 @@
 amp_viewer_function = authenticated(any_of=["AMPAdmin", "AMPEditor", "AMPViewer"])
 
 
+# Hydrograph-Corrector Staging Permissions -------------------------------------
+# The hydrograph corrector's publish and range-delete routes write and destroy
+# transducer records, and the workbench driving them is still being validated
+# against real logger files. `AMP.Staging` is its own group with no tier below
+# it and no AMP tier above it -- an AMPAdmin does not satisfy it. Nobody holds
+# it until it is granted in Authentik, so the routes ship dark and reachable
+# only by whoever is testing them.
+#
+# This is deliberately not a fourth rung on the AMP ladder. When the workbench
+# is trusted, these routes move to `amp_admin_dependency` and the group goes
+# away; leaving it as a tier would make that a schema change instead of a
+# one-line edit.
+amp_staging_function = authenticated(any_of=["AMP.Staging"])
+
+
 # Lexicon-Specific Authentication/Permissions ----------------------------------
 
 lexicon_admin_function = authenticated(any_of=["LexiconAdmin"])
@@ -89,5 +104,7 @@
 amp_editor_dependency: TypeAlias = Annotated[dict, Depends(amp_editor_function)]
 amp_viewer_dependency: TypeAlias = Annotated[dict, Depends(amp_viewer_function)]
 
+amp_staging_dependency: TypeAlias = Annotated[dict, Depends(amp_staging_function)]
+
 no_permission_dependency: TypeAlias = Annotated[dict, Depends(no_permission_function)]
 # ============= EOF =============================================