Gate DIE->Dagster log forwarding behind env var (default off)

forward_die_logs is now a no-op unless DIE_FORWARD_LOGS_TO_DAGSTER is
truthy. Keeps the Dagster compute log clean by default; enable per
deployment when DIE-level logs are needed.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: jirhiker

orchestration/logging_bridge.py

@@ -6,9 +6,19 @@
 so we never leak handlers across runs.
 """
 import logging
+import os
 from contextlib import contextmanager
 
 
+def _enabled() -> bool:
+    return os.environ.get("DIE_FORWARD_LOGS_TO_DAGSTER", "false").strip().lower() in (
+        "1",
+        "true",
+        "yes",
+        "on",
+    )
+
+
 class _DagsterForwardHandler(logging.Handler):
     def __init__(self, dagster_logger):
         super().__init__()
@@ -24,7 +34,14 @@ def emit(self, record):
 
 @contextmanager
 def forward_die_logs(context, level=logging.INFO):
-    """Relay root-logger records to `context.log` for the duration of the block."""
+    """Relay root-logger records to `context.log` for the duration of the block.
+
+    No-op unless the DIE_FORWARD_LOGS_TO_DAGSTER env var is truthy (default off).
+    """
+    if not _enabled():
+        yield
+        return
+
     handler = _DagsterForwardHandler(context.log)
     handler.setLevel(level)
     handler.setFormatter(logging.Formatter("%(name)-30s %(message)s"))