feat(ingestion): schedule the San Acacia ingest weekly

automated_ingestion/defs/definitions.py

@@ -24,10 +24,16 @@
 from dagster import Definitions
 
 from automated_ingestion.defs.assets import all_assets
+from automated_ingestion.defs.jobs.san_acacia import (
+    san_acacia_job,
+    san_acacia_weekly_schedule,
+)
 from automated_ingestion.defs.resources import OcotilloDatabase
 
 defs = Definitions(
     assets=all_assets(),
+    jobs=[san_acacia_job],
+    schedules=[san_acacia_weekly_schedule],
     resources={"database": OcotilloDatabase()},
 )
 

automated_ingestion/defs/jobs/san_acacia.py

@@ -0,0 +1,80 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+The scheduled run for San Acacia Reach.
+
+One job over the whole `san_acacia` asset group, so the three steps stay in
+order: land the point roster, land the readings, then map and load them. Dagster
+resolves that from the asset dependencies rather than from anything declared
+here, which is why the selection is by group -- a fourth asset added to the
+group joins the schedule without this file changing.
+
+Weekly rather than daily. These are five-minute diver readings and nobody is
+watching them in real time; the vendor's endpoint answers 500 when pushed, and a
+weekly cadence keeps each run's windows comfortably inside what it serves. The
+watermark makes the interval a matter of freshness rather than correctness: a
+run fetches from wherever the last one finished, so a missed week is picked up
+by the next run rather than lost.
+"""
+
+from dagster import (
+    AssetSelection,
+    DefaultScheduleStatus,
+    RetryPolicy,
+    ScheduleDefinition,
+    define_asset_job,
+)
+
+SAN_ACACIA_GROUP = "san_acacia"
+
+san_acacia_job = define_asset_job(
+    name="san_acacia_ingest",
+    selection=AssetSelection.groups(SAN_ACACIA_GROUP),
+    description=(
+        "Land the San Acacia point roster and readings in the raw zone, then "
+        "map and load them into Ocotillo."
+    ),
+    # A retry covers the vendor dropping a request or a token expiring mid-run.
+    # Two attempts, not more: a persistent 500 means the window is wrong or the
+    # endpoint is unwell, and hammering it makes both worse.
+    op_retry_policy=RetryPolicy(max_retries=2, delay=60),
+)
+
+san_acacia_weekly_schedule = ScheduleDefinition(
+    name="san_acacia_weekly",
+    job=san_acacia_job,
+    # Mondays at 05:00 America/Denver -- after midnight so a run covers whole
+    # days, and early enough that a failure is visible at the start of the week
+    # rather than discovered the following Monday.
+    cron_schedule="0 5 * * 1",
+    execution_timezone="America/Denver",
+    # Local time rather than UTC deliberately: the wells, the people who read
+    # the data, and the working day are all in one timezone, so a schedule that
+    # shifts by an hour twice a year would be the surprising choice.
+    #
+    # Stopped by default. Turning it on starts writing to Ocotillo, and the
+    # first run for the 24 wells without history fetches back to the
+    # `INITIAL_START` floor. That should be somebody's decision, taken once,
+    # rather than a consequence of a merge.
+    default_status=DefaultScheduleStatus.STOPPED,
+    description=(
+        "Weekly San Acacia ingest. Each run resumes from each series' "
+        "watermark, so a missed week is caught up rather than lost."
+    ),
+)
+
+
+# ============= EOF =============================================

automated_ingestion/tests/test_schedule.py

@@ -0,0 +1,74 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+The weekly schedule selects what it claims to.
+
+A group name is a string, so a typo yields a schedule that runs successfully and
+ingests nothing -- which looks like everything is fine.
+"""
+
+from dagster import DefaultScheduleStatus
+
+from automated_ingestion.defs.definitions import defs
+
+EXPECTED = {
+    "raw_san_acacia_locations",
+    "raw_san_acacia_readings",
+    "san_acacia_observations",
+}
+
+
+def _schedule():
+    return next(s for s in defs.schedules if s.name == "san_acacia_weekly")
+
+
+def test_the_schedule_is_registered():
+    assert _schedule().job.name == "san_acacia_ingest"
+
+
+def test_it_selects_every_san_acacia_asset_and_nothing_else():
+    selected = {
+        key.to_user_string()
+        for key in _schedule().job.selection.resolve(list(defs.assets))
+    }
+    assert selected == EXPECTED
+
+
+def test_operations_assets_are_excluded():
+    # ingestion_heartbeat and database_connectivity are diagnostics. Running
+    # them weekly would add noise and, for connectivity, a pointless query.
+    selected = {
+        key.to_user_string()
+        for key in _schedule().job.selection.resolve(list(defs.assets))
+    }
+    assert "ingestion_heartbeat" not in selected
+    assert "database_connectivity" not in selected
+
+
+def test_it_runs_weekly_in_local_time():
+    schedule = _schedule()
+    assert schedule.cron_schedule == "0 5 * * 1"
+    assert schedule.execution_timezone == "America/Denver"
+
+
+def test_it_is_stopped_until_somebody_starts_it():
+    # Turning it on begins writing to Ocotillo, and the first run for the wells
+    # without history fetches back to the floor. That is a decision, not a
+    # consequence of a merge.
+    assert _schedule().default_status is DefaultScheduleStatus.STOPPED
+
+
+# ============= EOF =============================================

docs/automated-ingestion-pipeline-plan.md

@@ -419,11 +419,14 @@ Covers raw already in GCS with only the mapping wrong: adapter or unit bug, newl
 
 ### 4.4 — Schedule, observability, alerting
 
-- `san_acacia_schedule` runs the daily pipeline; cron avoids contention with existing Dagster+ jobs in the org, recorded in the source registry.
-- Dagster logs bridge into the repo's existing logging setup, so ingestion failures surface where the team already looks. Confirm which error-tracking destination is current before wiring this — do not assume the repo's existing integrations are live.
-- A failed run notifies someone — not discovered via a stale hydrograph.
-- Every run emits rows ingested, rows upserted, entities processed, entities failed, adapter failures, resulting watermark per series.
-- A zero-new-rows run succeeds and is distinguishable in the logs from a failure.
+Schedule built. `defs/jobs/san_acacia.py` — `san_acacia_ingest` over the whole `san_acacia` asset group, on `san_acacia_weekly`.
+
+- ✅ **Weekly, not daily.** These are five-minute diver readings nobody watches in real time, the vendor's endpoint answers 500 when pushed, and the watermark makes the interval a question of freshness rather than correctness — a missed week is caught up by the next run, not lost.
+- ✅ Mondays 05:00 **America/Denver**, not UTC. The wells, the people reading the data and the working day are in one timezone; a schedule drifting an hour twice a year would be the surprising choice. After midnight so a run covers whole days, early enough that a failure is visible at the start of the week.
+- ✅ Selected **by group**, so an asset added to `san_acacia` joins the schedule without touching the job. A test asserts the selection resolves to exactly the three ingest assets and excludes `ingestion_heartbeat` and `database_connectivity` — a group-name typo would otherwise produce a schedule that runs happily and ingests nothing.
+- ✅ `RetryPolicy(max_retries=2, delay=60)` covers a dropped request or a token expiring mid-run. Two attempts, not more: a persistent 500 means the window is wrong or the endpoint is unwell, and hammering it makes both worse.
+- ✅ **`DefaultScheduleStatus.STOPPED`.** Turning it on starts writing to Ocotillo, and the first run for the 24 wells without history fetches back to `INITIAL_START`. That should be a decision taken once, not a consequence of a merge.
+- ⬜ Observability and alerting — log bridge, failure notification, run metadata.
 
 ### 4.5 — Documentation