fix: submit executor as Iris job so child jobs inherit S3 env vars

The marin-on-iris integration test was running executor_main in-process
on the CI runner. Child Zephyr jobs on remote pods had no S3 credentials
because Iris env propagation only works parent→child, and there was no
parent Iris job. Submit the whole executor as an Iris job with S3 env
vars so Zephyr coordinator/workers inherit them automatically.

Also enable live log output (-s --log-cli-level=INFO) for the local
iris integration test workflow.

Author: rjpower

.github/workflows/iris-integration.yaml

@@ -69,7 +69,7 @@ jobs:
         run: |
           uv run pytest tests/integration/iris/ \
             --controller-url "$IRIS_CONTROLLER_URL" \
-            -v --tb=short --timeout=600 \
+            -v -s --log-cli-level=INFO --tb=short --timeout=600 \
             -o "addopts=" \
             -x
         env:

tests/integration/iris/test_marin_on_iris.py

@@ -7,8 +7,12 @@
 cluster via FrayIrisClient instead of Ray.
 
 When MARIN_CI_S3_PREFIX is set (e.g. on CoreWeave CI), all paths use S3 so
-remote Zephyr pods can access them. Otherwise falls back to a local tmpdir
-(works when the Iris cluster is local / in-process).
+remote Zephyr pods can access them. The executor is submitted as an Iris job
+so that child jobs (Zephyr coordinator/workers) inherit env vars — including
+S3 credentials — via Iris auto-propagation.
+
+Otherwise falls back to running in-process with a local tmpdir (works when
+the Iris cluster is local / in-process).
 """
 
 import logging
@@ -22,6 +26,7 @@
 import pytest
 from fray import set_current_client
 from fray.v2.iris_backend import FrayIrisClient
+from fray.v2.types import Entrypoint, JobRequest, ResourceConfig, create_environment
 from marin.execution.executor import ExecutorMainConfig, executor_main
 from tests.integration_test import create_steps
 
@@ -32,6 +37,9 @@
 
 pytestmark = [pytest.mark.integration, pytest.mark.slow]
 
+# S3/R2 env vars that must be forwarded to Iris jobs for object storage access.
+_S3_ENV_KEYS = ["AWS_ACCESS_KEY_ID", "AWS_SECRET_ACCESS_KEY", "AWS_ENDPOINT_URL", "FSSPEC_S3"]
+
 
 def _upload_tree(local_root: Path, s3_dest: str) -> None:
     """Upload a local directory tree to S3."""
@@ -52,6 +60,21 @@ def _rm_s3(s3_prefix: str) -> None:
         pass
 
 
+def _s3_env_vars() -> dict[str, str]:
+    """Collect S3/R2 env vars from the current process."""
+    return {k: os.environ[k] for k in _S3_ENV_KEYS if k in os.environ}
+
+
+def _run_executor(prefix: str, synth_data: str) -> None:
+    """Entry point for the Iris job that runs the full executor pipeline."""
+    config = ExecutorMainConfig(
+        prefix=prefix,
+        executor_info_base_path=f"{prefix}/experiments",
+    )
+    steps = create_steps("quickstart-tests", synth_data)
+    executor_main(config, steps=steps)
+
+
 @pytest.mark.timeout(600)
 def test_marin_pipeline_on_iris(integration_cluster, monkeypatch):
     """Run the full marin data pipeline dispatched through Iris."""
@@ -81,15 +104,26 @@ def test_marin_pipeline_on_iris(integration_cluster, monkeypatch):
             workspace=REPO_ROOT,
         )
 
-        config = ExecutorMainConfig(
-            prefix=prefix,
-            executor_info_base_path=f"{prefix}/experiments",
-        )
-
-        experiment_prefix = "quickstart-tests"
-        steps = create_steps(experiment_prefix, synth_data)
+        env_vars = {
+            "MARIN_PREFIX": prefix,
+            "WANDB_MODE": "disabled",
+            "WANDB_API_KEY": "",
+            "JAX_TRACEBACK_FILTERING": "off",
+            **_s3_env_vars(),
+        }
 
         with set_current_client(iris_client):
-            executor_main(config, steps=steps)
+            handle = iris_client.submit(
+                JobRequest(
+                    name=f"marin-itest-{uuid.uuid4().hex[:8]}",
+                    entrypoint=Entrypoint.from_callable(
+                        _run_executor,
+                        args=(prefix, synth_data),
+                    ),
+                    resources=ResourceConfig.with_cpu(),
+                    environment=create_environment(env_vars=env_vars),
+                )
+            )
+            handle.wait(raise_on_failure=True)
     finally:
         cleanup()