From 1df7d8c63a85ce7a48146fd9ba31159a2c32e06c Mon Sep 17 00:00:00 2001
From: jakeross <jirhiker@gmail.com>
Date: Tue, 18 Aug 2026 16:16:37 -0700
Subject: [PATCH] fix(ingestion): set PYTHONPATH and report the import
 environment

Inserting the repository root from automated_ingestion/__init__.py did not fix
the ModuleNotFoundError for db in a step process, so the assumption behind that
fix was wrong somewhere I cannot see from here.

PYTHONPATH=/opt/dagster/app makes the app root importable regardless of how a
process was launched, rather than depending on the package having been imported
first or on the working directory being on the path.

The heartbeat asset now reports cwd, the resolved app root and its contents,
whether db and domain are findable, and sys.path. It needs no credentials, so
it reports even when everything else fails -- which is what a diagnostic asset
is for. If PYTHONPATH does not resolve this, its metadata says why.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 automated_ingestion/defs/assets/heartbeat.py  | 38 +++++++++++++++++--
 .../scripts/set_code_location_env.sh          |  9 +++++
 2 files changed, 43 insertions(+), 4 deletions(-)

diff --git a/automated_ingestion/defs/assets/heartbeat.py b/automated_ingestion/defs/assets/heartbeat.py
index 2fe21ee28..8e78157d8 100644
--- a/automated_ingestion/defs/assets/heartbeat.py
+++ b/automated_ingestion/defs/assets/heartbeat.py
@@ -24,18 +24,48 @@
 
 from datetime import datetime, timezone
 
-from dagster import AssetExecutionContext, asset
+from dagster import AssetExecutionContext, MetadataValue, Output, asset
 
 
 @asset(
     group_name="operations",
     description="Static heartbeat proving the code location loaded and can run.",
 )
-def ingestion_heartbeat(context: AssetExecutionContext) -> str:
-    """Return the materialization timestamp."""
+def ingestion_heartbeat(context: AssetExecutionContext) -> Output[str]:
+    """Return the materialization timestamp, with the import environment.
+
+    The environment metadata is here because a step process is not the process
+    that loaded the code location, and the two do not necessarily agree about
+    sys.path. When an import that works at load time fails at execution, this is
+    the asset that says why -- it runs without credentials, so it reports even
+    when everything else is broken.
+    """
+    import os
+    import sys
+    from importlib.util import find_spec
+
     stamp = datetime.now(timezone.utc).isoformat()
     context.log.info("automated_ingestion code location alive at %s", stamp)
-    return stamp
+
+    app_root = os.path.dirname(
+        os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+    )
+    try:
+        siblings = sorted(os.listdir(app_root))
+    except OSError as exc:
+        siblings = [f"<unreadable: {exc}>"]
+
+    return Output(
+        stamp,
+        metadata={
+            "cwd": MetadataValue.text(os.getcwd()),
+            "app_root": MetadataValue.text(app_root),
+            "app_root_contents": MetadataValue.text(", ".join(siblings)),
+            "db_on_path": MetadataValue.bool(find_spec("db") is not None),
+            "domain_on_path": MetadataValue.bool(find_spec("domain") is not None),
+            "sys_path": MetadataValue.json(sys.path),
+        },
+    )
 
 
 # ============= EOF =============================================
diff --git a/automated_ingestion/scripts/set_code_location_env.sh b/automated_ingestion/scripts/set_code_location_env.sh
index 35a263b59..7d11f647e 100755
--- a/automated_ingestion/scripts/set_code_location_env.sh
+++ b/automated_ingestion/scripts/set_code_location_env.sh
@@ -40,6 +40,15 @@ set_var() { echo "  $1"; $DG plus create env "$@" -y >/dev/null; }
 
 case "$PHASE" in
 storage)
+  # The image copies the repository to /opt/dagster/app but never installs it,
+  # so db/ and domain/ are importable only if that directory is on the path.
+  # The process that loads the code location has it; the process that executes a
+  # step does not reliably, which shows up as ModuleNotFoundError for db at
+  # execution while the location itself loads fine. Setting PYTHONPATH removes
+  # the guesswork instead of depending on how each process was launched.
+  echo "Import path:"
+  set_var PYTHONPATH /opt/dagster/app
+
   echo "Raw-zone buckets (different value per scope):"
   set_var INGESTION_GCS_BUCKET ocotillo-ingestion-production --scope full
   set_var INGESTION_GCS_BUCKET ocotillo-ingestion-staging --scope branch