fix(ingestion): report the import environment from the heartbeat asset

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 =============================================

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