feat(ingestion): add automated ingestion pipeline foundations

.github/workflows/CD_dagster_branch.yml

@@ -0,0 +1,90 @@
+# Creates a Dagster+ branch deployment for a pull request, so ingestion changes
+# can be materialized against an isolated deployment before they reach prod.
+#
+# Path-filtered: most PRs in this repository touch only the API and should not
+# create a Dagster+ deployment at all.
+name: CD (Dagster+ branch deployment)
+
+on:
+  pull_request:
+    types: [opened, synchronize, reopened, closed]
+    paths:
+      - "automated_ingestion/**"
+      # The code location imports db/ models and domain/ rules in-process,
+      # so a change to either alters what this image runs even when no
+      # ingestion file moves. Without these, a domain fix merged to
+      # production would leave the pipeline running the old rule against
+      # the live database. The cost is that ordinary API changes to these
+      # directories also trigger a build; a stale code location is worse.
+      - "db/**"
+      - "domain/**"
+      - "dagster_cloud.yaml"
+      - "pyproject.toml"
+      - "uv.lock"
+      - ".github/workflows/CD_dagster_branch.yml"
+
+permissions:
+  contents: read
+  pull-requests: write
+
+# One deployment per PR; a force-push supersedes the run it interrupts.
+concurrency:
+  group: dagster-branch-deploy-${{ github.event.pull_request.number }}
+  cancel-in-progress: true
+
+jobs:
+  dagster-branch-deploy:
+    runs-on: ubuntu-latest
+    # Forks cannot read the Dagster+ secrets, and a branch deployment from an
+    # untrusted fork would run our code against our infrastructure regardless.
+    if: github.event.pull_request.head.repo.full_name == github.repository
+
+    # The action's notify steps post build status as a PR comment and read the
+    # token from the workflow environment -- `env.GITHUB_TOKEN`, not the
+    # `secrets` context. Without this the run dies on an empty-token assertion
+    # before it ever reaches Dagster+, which reads as an auth failure but is
+    # not one.
+    env:
+      GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}
+
+    steps:
+      - name: Check out source repository
+        uses: actions/checkout@v7.0.1
+
+      # parse_workspace performs its own `actions/checkout`, which cleans the
+      # working tree. It has to run *before* requirements.txt is generated, or
+      # the generated file is deleted before the deploy step can use it.
+      - name: Parse dagster_cloud.yaml
+        id: parse
+        uses: dagster-io/dagster-cloud-action/actions/utils/parse_workspace@v1.13.18
+        with:
+          dagster_cloud_file: dagster_cloud.yaml
+
+      - name: Install uv in container
+        uses: astral-sh/setup-uv@v10.0.1
+        with:
+          version: "latest"
+
+      - name: Generate requirements.txt
+        run: |
+          uv export \
+            --format requirements-txt \
+            --no-emit-project \
+            --no-dev \
+            --group ingestion \
+            --output-file requirements.txt
+
+      # Runs on `closed` too: the action tears the branch deployment down when
+      # the PR is merged or abandoned, so stale deployments do not accumulate.
+      - name: Deploy to Dagster+ branch deployment
+        uses: dagster-io/dagster-cloud-action/actions/serverless_branch_deploy@v1.13.18
+        with:
+          organization_id: ${{ vars.DAGSTER_CLOUD_ORGANIZATION_ID }}
+          dagster_cloud_api_token: ${{ secrets.DAGSTER_CLOUD_API_TOKEN }}
+          location: ${{ toJson(fromJson(steps.parse.outputs.build_info)[0]) }}
+          checkout_repo: false
+          # The action defaults to python:3.8-slim, which cannot install a
+          # lockfile resolved for requires-python >= 3.13 -- pip reports the
+          # pins as having no matching distribution rather than as a version
+          # conflict, which reads like a broken requirements file.
+          base_image: python:3.13-slim

.github/workflows/CD_dagster_prod.yml

@@ -0,0 +1,89 @@
+# Deploys the `ocotillo-automated-ingestion` code location to the Dagster+ prod
+# deployment.
+#
+# Triggered on `production`, not `main`: `main` was abandoned in July 2025 and
+# the release flow runs feature -> staging -> production (docs/release-flow.md).
+# The plan document's reference to `main` predates that being checked.
+#
+# Path-filtered so an ordinary API change does not spend a Dagster+ build. The
+# filter includes pyproject.toml and uv.lock because the location's dependency
+# set is exported from them, so a lockfile bump changes the built image even
+# when no ingestion source file does.
+name: CD (Dagster+ prod)
+
+on:
+  push:
+    branches: [production]
+    paths:
+      - "automated_ingestion/**"
+      # The code location imports db/ models and domain/ rules in-process,
+      # so a change to either alters what this image runs even when no
+      # ingestion file moves. Without these, a domain fix merged to
+      # production would leave the pipeline running the old rule against
+      # the live database. The cost is that ordinary API changes to these
+      # directories also trigger a build; a stale code location is worse.
+      - "db/**"
+      - "domain/**"
+      - "dagster_cloud.yaml"
+      - "pyproject.toml"
+      - "uv.lock"
+      - ".github/workflows/CD_dagster_prod.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: dagster-prod-deploy
+  cancel-in-progress: false
+
+jobs:
+  dagster-prod-deploy:
+    runs-on: ubuntu-latest
+    environment: production
+
+    steps:
+      - name: Check out source repository
+        uses: actions/checkout@v7.0.1
+
+      # parse_workspace performs its own `actions/checkout`, which cleans the
+      # working tree. It has to run *before* requirements.txt is generated, or
+      # the generated file is deleted before the deploy step can use it.
+      - name: Parse dagster_cloud.yaml
+        id: parse
+        uses: dagster-io/dagster-cloud-action/actions/utils/parse_workspace@v1.13.18
+        with:
+          dagster_cloud_file: dagster_cloud.yaml
+
+      - name: Install uv in container
+        uses: astral-sh/setup-uv@v10.0.1
+        with:
+          version: "latest"
+
+      # Dagster+ builds from a requirements.txt, which the repo does not keep
+      # under version control. `--group ingestion` adds dagster and dlt on top
+      # of the runtime dependencies; the runtime ones are needed too, because
+      # the loader imports `db/` and `domain/`.
+      - name: Generate requirements.txt
+        run: |
+          uv export \
+            --format requirements-txt \
+            --no-emit-project \
+            --no-dev \
+            --group ingestion \
+            --output-file requirements.txt
+
+      # checkout_repo is false because requirements.txt is generated above and
+      # a second checkout would discard it.
+      - name: Deploy to Dagster+ prod
+        uses: dagster-io/dagster-cloud-action/actions/serverless_prod_deploy@v1.13.18
+        with:
+          organization_id: ${{ vars.DAGSTER_CLOUD_ORGANIZATION_ID }}
+          dagster_cloud_api_token: ${{ secrets.DAGSTER_CLOUD_API_TOKEN }}
+          location: ${{ toJson(fromJson(steps.parse.outputs.build_info)[0]) }}
+          checkout_repo: false
+          # The action defaults to python:3.8-slim, which cannot install a
+          # lockfile resolved for requires-python >= 3.13 -- pip reports the
+          # pins as having no matching distribution rather than as a version
+          # conflict, which reads like a broken requirements file.
+          base_image: python:3.13-slim

.github/workflows/tests.yml

@@ -86,7 +86,7 @@ jobs:
           key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }}
 
       - name: Install the project
-        run: uv sync --locked --all-extras --dev --group cli
+        run: uv sync --locked --all-extras --dev --group cli --group ingestion
 
       - name: Show Alembic heads
         run: uv run alembic heads
@@ -214,7 +214,7 @@ jobs:
           key: venv-${{ runner.os }}-${{ steps.setup-python.outputs.python-version }}-${{ hashFiles('uv.lock') }}
 
       - name: Install the project
-        run: uv sync --locked --all-extras --dev --group cli
+        run: uv sync --locked --all-extras --dev --group cli --group ingestion
 
       - name: Show Alembic heads
         run: uv run alembic heads

automated_ingestion/__init__.py

@@ -0,0 +1,33 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+Automated ingestion: scheduled pipelines that land external monitoring data in
+Ocotillo without anyone hand-carrying a file.
+
+This package is deployed as its own Dagster+ code location, separate from the
+API process, but it lives in this repository so the loader can import ``db/``
+models and ``domain/`` rules directly instead of maintaining a second copy of
+the Ocotillo schema elsewhere.
+
+Shape of a source: a dlt pipeline extracts the vendor API into a GCS raw zone,
+an adapter maps raw records onto Ocotillo structures, and a loader writes them
+to Postgres over a direct connection. San Acacia Reach (Van Essen divers) is
+the first source; ``shared/`` holds what later sources reuse.
+
+See ``docs/automated-ingestion-pipeline-plan.md``.
+"""
+
+# ============= EOF =============================================

automated_ingestion/defs/__init__.py

@@ -0,0 +1,18 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""Dagster definitions: the code location's assets, jobs, and schedules."""
+
+# ============= EOF =============================================

automated_ingestion/defs/assets/__init__.py

@@ -0,0 +1,43 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+Asset collection for the code location.
+
+Per-source assets are declared in their own modules and gathered here so
+``definitions.py`` never has to know which sources exist.
+"""
+
+from dagster import AssetsDefinition
+
+from automated_ingestion.defs.assets.connectivity import database_connectivity
+from automated_ingestion.defs.assets.heartbeat import ingestion_heartbeat
+from automated_ingestion.sources.san_acacia.ingest import (
+    raw_san_acacia_locations,
+    raw_san_acacia_readings,
+)
+
+
+def all_assets() -> list[AssetsDefinition]:
+    """Every asset the code location exposes."""
+    return [
+        ingestion_heartbeat,
+        database_connectivity,
+        raw_san_acacia_locations,
+        raw_san_acacia_readings,
+    ]
+
+
+# ============= EOF =============================================

automated_ingestion/defs/assets/connectivity.py

@@ -0,0 +1,62 @@
+# ===============================================================================
+# 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.
+# ===============================================================================
+"""
+Proves the Dagster+ runtime can reach Ocotillo Postgres.
+
+Dagster+ Serverless runs outside the VPC, so Cloud SQL's private IP is
+unreachable from it -- the connection has to go through the Cloud SQL connector
+instead. That is the single riskiest assumption in the foundations task, and it
+fails at run time rather than at deploy time. This asset makes it fail loudly,
+on its own, in an asset whose only job is to fail there.
+
+It reads and never writes: connectivity and permission are separable problems,
+and a write here would leave test rows in a real table.
+"""
+
+from dagster import AssetExecutionContext, MetadataValue, Output, asset
+
+from automated_ingestion.defs.resources import OcotilloDatabase
+
+
+@asset(
+    group_name="operations",
+    description="Reads from Ocotillo Postgres to prove the runtime can connect.",
+)
+def database_connectivity(
+    context: AssetExecutionContext, database: OcotilloDatabase
+) -> Output[int]:
+    """Count transducer observations, returning the count as metadata."""
+    from sqlalchemy import func, select
+
+    from db.transducer import TransducerObservation
+
+    with database.session() as session:
+        count = session.scalar(select(func.count()).select_from(TransducerObservation))
+
+    count = int(count or 0)
+    context.log.info("connected to Ocotillo; transducer_observation rows: %s", count)
+    return Output(
+        count,
+        metadata={
+            "transducer_observation_rows": MetadataValue.int(count),
+            "note": MetadataValue.text(
+                "Read-only. A failure here is connectivity or grants, not data."
+            ),
+        },
+    )
+
+
+# ============= EOF =============================================