feat(airflow): scaffold Phase-1 Airflow 3.x orchestration tree

Add a parse-only scaffold for migrating the U19 orchestration layer to
Apache Airflow 3.x (assessment + plan in #95).

- airflow/dags: nightly_populate (unified Python+MATLAB nightly + alerts),
  ephys_processing, imaging_processing — TaskFlow DAGs with stubbed task
  bodies. nightly_populate encodes the FK ordering plus the implicit edges
  found in the make() audit (TowersSessionPsychTask after TowersBlock; a
  barrier before the subject-cumulative psych tables).
- airflow/plugins/u19: stub hooks/helpers that wrap existing code
  (slurm_creator, clusters_paths_and_transfers, slack_utils, dual-write of
  recording_process status, matlab -batch). No logic reimplemented.
- CI: .github/workflows/airflow-dag-validation.yml runs a DagBag
  import-error check via `uv sync --only-group airflow` + `uv run`.
- pyproject.toml: add locked `airflow` dependency group (apache-airflow
  3.2.2 + ssh/standard providers); uv.lock updated.

All three DAGs parse with zero import errors against Airflow 3.2.2. Task
bodies are TODO stubs; deployment/infra questions tracked in #95.

Assisted-by: ClaudeCode:claude-opus-4.8

Author: tabedzki

.github/workflows/airflow-dag-validation.yml

@@ -0,0 +1,61 @@
+# Airflow DAG validation workflow
+# Runs on every push and pull_request to verify that all DAGs in airflow/dags/
+# are importable by Airflow's DagBag with zero import errors.
+# No Airflow database is required — DagBag loads the DAGs in-process.
+#
+# Dependencies are managed with uv against the locked `airflow` dependency group
+# in pyproject.toml (uv.lock), so installs are reproducible without a pip
+# constraints file.
+
+name: Airflow DAG Validation
+
+on:
+  push:
+  pull_request:
+
+jobs:
+  dag-validation:
+    name: Validate DAGs (Python 3.12 / Airflow 3.2.2)
+    runs-on: ubuntu-latest
+
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@v4
+
+      - name: Install uv
+        uses: astral-sh/setup-uv@v5
+        with:
+          enable-cache: true
+
+      # Resolve only the `airflow` dependency group from uv.lock — this keeps the
+      # CI environment small (no GPU/science stack) and fully reproducible.
+      - name: Sync Airflow dependencies
+        run: uv sync --only-group airflow
+
+      # Run DagBag import-error check via `uv run`.
+      # AIRFLOW__CORE__DAGS_FOLDER points at the DAG directory.
+      # PYTHONPATH includes airflow/plugins so the u19.* plugin imports resolve.
+      # AIRFLOW__CORE__LOAD_EXAMPLES=False avoids loading Airflow's bundled examples.
+      - name: Check DAG import errors
+        env:
+          AIRFLOW__CORE__DAGS_FOLDER: "${{ github.workspace }}/airflow/dags"
+          AIRFLOW__CORE__LOAD_EXAMPLES: "False"
+          PYTHONPATH: "${{ github.workspace }}/airflow/plugins"
+        run: |
+          uv run --only-group airflow python - <<'EOF'
+          from airflow.models import DagBag
+          import os
+
+          dags_folder = os.environ["AIRFLOW__CORE__DAGS_FOLDER"]
+          db = DagBag(dag_folder=dags_folder, include_examples=False)
+
+          if db.import_errors:
+              import json
+              print("DAG import errors detected:")
+              print(json.dumps(db.import_errors, indent=2))
+              raise SystemExit(1)
+
+          print(f"OK — {len(db.dags)} DAG(s) loaded with no import errors:")
+          for dag_id in sorted(db.dags):
+              print(f"  {dag_id}")
+          EOF

airflow/README.md

@@ -0,0 +1,54 @@
+# U19 Pipeline — Airflow Phase-1 Scaffold
+
+This directory contains the **Phase-1 Apache Airflow scaffold** for migrating the U19
+neuroscience pipeline from bare cron jobs to Airflow orchestration. The project targets
+**Airflow 3.x** (currently pinned to `apache-airflow==3.2.2`). Tracked in
+[BrainCOGS/U19-pipeline-python#95](https://github.com/BrainCOGS/U19-pipeline-python/issues/95).
+
+## Layout
+
+```
+airflow/
+  README.md                     # this file
+  requirements-airflow.txt      # Airflow + provider pins
+  dags/
+    nightly_populate.py         # @daily — behavior/optogenetics/pupillometry/alerts
+    ephys_processing.py         # @hourly — ephys recording pipeline
+    imaging_processing.py       # @hourly — imaging recording pipeline
+  plugins/u19/
+    __init__.py
+    datajoint_sensor.py         # DataJoint table-populated sensor
+    slurm.py                    # SLURM job submission/polling
+    transfers.py                # Globus file transfer helpers
+    matlab.py                   # SSH-based MATLAB batch runner
+    status.py                   # Dual-write status + log helper
+    callbacks.py                # Slack failure callback
+```
+
+## Status
+
+**All task bodies are stubs** (`NotImplementedError` / `pass`). The DAGs parse and are
+DagBag-importable, but no real work is performed. Each stub references the existing
+`u19_pipeline` module that should be called once the task is fleshed out.
+
+## DAG Validation in CI
+
+A GitHub Actions workflow (`.github/workflows/airflow-dag-validation.yml`) uses **uv**
+to `uv sync --only-group airflow` (Airflow 3.2.2, locked in `uv.lock`) and runs a
+`DagBag` import-error check on every push and pull request. No Airflow database is
+required — the DAGs are loaded in-process. The `airflow` dependency group is declared in
+the project `pyproject.toml`.
+
+## Open Infrastructure Questions
+
+The following decisions are **not yet made** and block a production deployment:
+
+1. **PNI hosting** — where does the Airflow webserver/scheduler run? (dedicated VM,
+   existing `braincogs` server, or cloud?)
+2. **MATLAB host** — which host runs the MATLAB batch processes, and what SSH
+   `conn_id` ("matlab_host") maps to in the Airflow Connections table?
+3. **AcquiredTiff owner** — `imaging_processing.py` notes the `AcquiredTiff` node is
+   contended. Should it be owned by a Python operator or a MATLAB SSH call? Guards
+   against double-run are needed before this DAG goes live.
+4. **Metadata DB** — Airflow's metadata DB (Postgres recommended for production) needs
+   to be provisioned and the connection string set before scheduler startup.

airflow/dags/ephys_processing.py

@@ -0,0 +1,202 @@
+# NOTE: scaffold — bodies are stubs
+"""Ephys recording processing DAG for U19.
+
+Runs @hourly.  Polls for active processing jobs and drives each through the
+linear pipeline: raw transfer → SLURM preprocessing → processed transfer →
+DataJoint element populate.
+
+All task bodies are TODO stubs.  Dynamic mapping is stubbed with a comment
+placeholder — a linear task graph is wired instead so the DAG parses cleanly.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta
+
+from airflow.sdk import dag, task
+
+log = logging.getLogger(__name__)
+
+# NOTE: u19 plugin imports are inside task bodies (TODO stubs) so the DAG
+# parses even when plugins/ is not on PYTHONPATH at collection time.
+
+default_args = {
+    "owner": "u19",
+    "retries": 1,
+    "retry_delay": timedelta(minutes=10),
+}
+
+
+@dag(
+    dag_id="u19_ephys_processing",
+    schedule="@hourly",
+    start_date=datetime(2024, 1, 1),
+    catchup=False,
+    default_args=default_args,
+    tags=["u19", "ephys", "processing"],
+)
+def u19_ephys_processing() -> None:
+    """Ephys processing DAG.
+
+    Intended flow (one task-group per active job)::
+
+        request_raw_transfer
+            → wait_raw_transfer
+            → submit_slurm
+            → wait_slurm
+            → request_proc_transfer
+            → wait_proc_transfer
+            → populate_element
+
+    Each step calls the matching plugin and calls dual_write_status after
+    advancing the job state.
+
+    TODO: replace the linear stub below with dynamic task-group mapping once
+    ``get_active_jobs()`` is implemented, e.g.::
+
+        @task_group
+        def process_job(job_id: int) -> None:
+            ...
+
+        jobs = get_active_jobs()
+        process_job.expand(job_id=jobs)
+    """
+
+    @task(task_id="get_active_jobs")
+    def get_active_jobs() -> list[int]:
+        """Return list of job_ids that are ready for processing.
+
+        TODO: query recording_process.Processing for jobs in the
+        'transfer_request' status; return their job_ids.
+
+        Returns
+        -------
+        list[int]
+            List of ``recording_process.Processing.job_id`` values.
+        """
+        # TODO: from u19_pipeline import recording_process
+        #       jobs = (recording_process.Processing & {"status_processing_id": STATUS_TRANSFER_REQUEST}).fetch("job_id")
+        #       return list(jobs)
+        log.info("get_active_jobs stub — returning empty list")
+        return []
+
+    @task(task_id="request_raw_transfer")
+    def request_raw_transfer(job_ids: list[int]) -> None:
+        """Initiate Globus raw-data transfer from PNI to processing cluster.
+
+        Wraps u19.transfers.globus_transfer.
+        Calls dual_write_status after initiating each transfer.
+
+        TODO: for job_id in job_ids:
+                  task_id = globus_transfer(pni_ep, tiger_ep, raw_src, raw_dst, job_id=job_id)
+                  dual_write_status({"job_id": job_id}, STATUS_RAW_TRANSFER_STARTED)
+        """
+        # TODO: from u19.transfers import globus_transfer
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="wait_raw_transfer")
+    def wait_raw_transfer(job_ids: list[int]) -> None:
+        """Poll until raw Globus transfer reaches SUCCEEDED state.
+
+        Wraps u19.transfers.check_transfer_status.
+        Calls dual_write_status when transfer completes.
+
+        TODO: poll check_transfer_status(task_id) until SUCCEEDED/FAILED
+              dual_write_status({"job_id": job_id}, STATUS_RAW_TRANSFER_DONE)
+        """
+        # TODO: from u19.transfers import check_transfer_status
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="submit_slurm")
+    def submit_slurm(job_ids: list[int]) -> None:
+        """Generate and submit a SLURM job for each active job.
+
+        Wraps u19.slurm.submit_slurm_job.
+        Calls dual_write_status after submission.
+
+        TODO: for job_id in job_ids:
+                  slurm_job_id = submit_slurm_job(job_id, program_selection_params)
+                  dual_write_status({"job_id": job_id}, STATUS_SLURM_SUBMITTED)
+        """
+        # TODO: from u19.slurm import submit_slurm_job
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="wait_slurm")
+    def wait_slurm(job_ids: list[int]) -> None:
+        """Poll SLURM until the job reaches a terminal state.
+
+        Wraps u19.slurm.poll_slurm_job.
+        Calls dual_write_status when SLURM job completes.
+
+        NOTE: should become a deferrable operator in Phase 2.
+
+        TODO: for job_id in job_ids:
+                  state = poll_slurm_job(slurm_job_id, job_id=job_id)
+                  dual_write_status({"job_id": job_id}, STATUS_SLURM_DONE if state=="COMPLETED" else STATUS_ERROR)
+        """
+        # TODO: from u19.slurm import poll_slurm_job
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="request_proc_transfer")
+    def request_proc_transfer(job_ids: list[int]) -> None:
+        """Initiate Globus processed-data transfer from cluster back to PNI.
+
+        Wraps u19.transfers.globus_transfer.
+        Calls dual_write_status after initiating each transfer.
+
+        TODO: for job_id in job_ids:
+                  task_id = globus_transfer(tiger_ep, pni_ep, proc_src, proc_dst, job_id=job_id)
+                  dual_write_status({"job_id": job_id}, STATUS_PROC_TRANSFER_STARTED)
+        """
+        # TODO: from u19.transfers import globus_transfer
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="wait_proc_transfer")
+    def wait_proc_transfer(job_ids: list[int]) -> None:
+        """Poll until processed-data Globus transfer reaches SUCCEEDED state.
+
+        Wraps u19.transfers.check_transfer_status.
+        Calls dual_write_status when transfer completes.
+
+        TODO: poll check_transfer_status(task_id) until SUCCEEDED/FAILED
+              dual_write_status({"job_id": job_id}, STATUS_PROC_TRANSFER_DONE)
+        """
+        # TODO: from u19.transfers import check_transfer_status
+        #       from u19.status import dual_write_status
+        pass
+
+    @task(task_id="populate_element")
+    def populate_element(job_ids: list[int]) -> None:
+        """Populate DataJoint ephys element tables for completed jobs.
+
+        Wraps u19_pipeline.automatic_job.ephys_element_populate.
+        Calls dual_write_status after populate completes.
+
+        TODO: for job_id in job_ids:
+                  ephys_element_populate.run(job_id)
+                  dual_write_status({"job_id": job_id}, STATUS_COMPLETE)
+        """
+        # TODO: import u19_pipeline.automatic_job.ephys_element_populate as ep
+        #       from u19.status import dual_write_status
+        pass
+
+    # -------------------------------------------------------------------------
+    # Wire linear stub graph
+    # -------------------------------------------------------------------------
+    t_jobs = get_active_jobs()
+    t_req_raw = request_raw_transfer(t_jobs)
+    t_wait_raw = wait_raw_transfer(t_req_raw)
+    t_submit_slurm = submit_slurm(t_wait_raw)
+    t_wait_slurm = wait_slurm(t_submit_slurm)
+    t_req_proc = request_proc_transfer(t_wait_slurm)
+    t_wait_proc = wait_proc_transfer(t_req_proc)
+    populate_element(t_wait_proc)
+
+
+u19_ephys_processing()