refactor: run marin-on-iris test as standalone script for streaming logs

Replace the pytest wrapper with a standalone script that streams logs
in real time. The test takes ~10 minutes and pytest swallows all output
until completion, making failures hard to diagnose.

The script runs as its own workflow step with `stream_logs=True` on the
Iris job handle, so executor and child job output appears immediately.

Author: rjpower

.github/workflows/iris-coreweave-ci.yaml

@@ -151,6 +151,21 @@ jobs:
             -o "addopts=" \
             -x
 
+      - name: Run marin-on-iris pipeline test
+        env:
+          WANDB_MODE: disabled
+          WANDB_API_KEY: ""
+          JAX_TRACEBACK_FILTERING: off
+          MARIN_CI_S3_PREFIX: s3://marin-na/temp/ci
+          AWS_ACCESS_KEY_ID: ${{ secrets.R2_ACCESS_KEY_ID }}
+          AWS_SECRET_ACCESS_KEY: ${{ secrets.R2_SECRET_ACCESS_KEY }}
+          AWS_ENDPOINT_URL: https://74981a43be0de7712369306c7b19133d.r2.cloudflarestorage.com
+          FSSPEC_S3: '{"endpoint_url": "https://74981a43be0de7712369306c7b19133d.r2.cloudflarestorage.com"}'
+        run: |
+          IRIS_CONTROLLER_URL="http://localhost:10000"
+          timeout 600 uv run tests/integration/iris/run_marin_on_iris.py \
+            --controller-url "$IRIS_CONTROLLER_URL"
+
       - name: Stop port-forward
         if: always()
         run: |

.github/workflows/iris-integration.yaml

@@ -77,6 +77,15 @@ jobs:
           WANDB_API_KEY: ""
           JAX_TRACEBACK_FILTERING: off
 
+      - name: Run marin-on-iris pipeline test
+        run: |
+          timeout 600 uv run tests/integration/iris/run_marin_on_iris.py \
+            --controller-url "$IRIS_CONTROLLER_URL"
+        env:
+          WANDB_MODE: disabled
+          WANDB_API_KEY: ""
+          JAX_TRACEBACK_FILTERING: off
+
       - name: Stop cluster
         if: always()
         run: kill $CLUSTER_PID 2>/dev/null || true

…s/integration/iris/test_marin_on_iris.py …ts/integration/iris/run_marin_on_iris.pytests/integration/iris/test_marin_on_iris.py renamed to tests/integration/iris/run_marin_on_iris.py tests/integration/iris/test_marin_on_iris.py renamed to tests/integration/iris/run_marin_on_iris.py

@@ -1,48 +1,46 @@
 # Copyright The Marin Authors
 # SPDX-License-Identifier: Apache-2.0
 
-"""Full marin integration pipeline running on Iris.
+"""Run the full marin pipeline on an Iris cluster.
 
-Ports the pipeline from tests/integration_test.py to dispatch through an Iris
-cluster via FrayIrisClient instead of Ray.
+Standalone script (not pytest) so logs stream in real time.
 
-When MARIN_CI_S3_PREFIX is set (e.g. on CoreWeave CI), all paths use S3 so
-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.
+Usage:
+    uv run tests/integration/iris/run_marin_on_iris.py \
+        --controller-url http://localhost:10000
 
-Otherwise falls back to running in-process with a local tmpdir (works when
-the Iris cluster is local / in-process).
+When MARIN_CI_S3_PREFIX is set, uploads test fixtures to S3 and submits
+the executor as an Iris job so child jobs inherit S3 credentials.
+Otherwise runs in-process against local filesystem.
 """
 
+import argparse
 import logging
 import os
 import shutil
+import sys
 import tempfile
 import uuid
 from pathlib import Path
 
 import fsspec
-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 iris.logging import configure_logging
 from marin.execution.executor import ExecutorMainConfig, executor_main
 from tests.integration_test import create_steps
 
+configure_logging(level=logging.INFO)
 logger = logging.getLogger(__name__)
 
 REPO_ROOT = Path(__file__).resolve().parents[3]
 LOCAL_SYNTH_DATA = REPO_ROOT / "tests" / "quickstart-data"
 
-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."""
     fs, _ = fsspec.core.url_to_fs(s3_dest)
     for path in local_root.rglob("*"):
         if not path.is_file():
@@ -52,7 +50,6 @@ def _upload_tree(local_root: Path, s3_dest: str) -> None:
 
 
 def _rm_s3(s3_prefix: str) -> None:
-    """Remove all objects under an S3 prefix (best-effort)."""
     fs, _ = fsspec.core.url_to_fs(s3_prefix)
     try:
         fs.rm(s3_prefix, recursive=True)
@@ -61,12 +58,10 @@ def _rm_s3(s3_prefix: str) -> None:
 
 
 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",
@@ -75,44 +70,38 @@ def _run_executor(prefix: str, synth_data: str) -> None:
     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."""
+def main():
+    parser = argparse.ArgumentParser(description="Run marin pipeline on Iris")
+    parser.add_argument("--controller-url", required=True)
+    args = parser.parse_args()
+
     s3_base = os.environ.get("MARIN_CI_S3_PREFIX")
 
     if s3_base:
-        # Remote cluster: use S3 so Zephyr coordinator pods can access data.
         run_id = f"marin-itest-{uuid.uuid4().hex[:8]}"
         prefix = f"{s3_base}/{run_id}"
         synth_data = f"{prefix}/quickstart-data"
+        logger.info("Uploading test fixtures to %s", synth_data)
         _upload_tree(LOCAL_SYNTH_DATA, synth_data)
         cleanup = lambda: _rm_s3(prefix)  # noqa: E731
     else:
-        # Local cluster: local filesystem is fine.
         prefix = tempfile.mkdtemp(prefix="iris-marin-itest-")
         synth_data = str(LOCAL_SYNTH_DATA)
         cleanup = lambda: shutil.rmtree(prefix, ignore_errors=True)  # noqa: E731
 
-    try:
-        monkeypatch.setenv("MARIN_PREFIX", prefix)
-        monkeypatch.setenv("WANDB_MODE", "disabled")
-        monkeypatch.setenv("WANDB_API_KEY", "")
-        monkeypatch.setenv("JAX_TRACEBACK_FILTERING", "off")
+    os.environ["MARIN_PREFIX"] = prefix
+    os.environ["WANDB_MODE"] = "disabled"
+    os.environ["WANDB_API_KEY"] = ""
+    os.environ["JAX_TRACEBACK_FILTERING"] = "off"
 
+    try:
         iris_client = FrayIrisClient(
-            controller_address=integration_cluster.url,
+            controller_address=args.controller_url,
             workspace=REPO_ROOT,
         )
 
-        config = ExecutorMainConfig(
-            prefix=prefix,
-            executor_info_base_path=f"{prefix}/experiments",
-        )
-        steps = create_steps("quickstart-tests", synth_data)
-
         if s3_base:
-            # Remote cluster: submit the executor as an Iris job so child jobs
-            # (Zephyr coordinator/workers) inherit S3 env vars automatically.
+            logger.info("Submitting executor as Iris job (S3 mode)")
             env_vars = {
                 "MARIN_PREFIX": prefix,
                 "WANDB_MODE": "disabled",
@@ -133,10 +122,24 @@ def test_marin_pipeline_on_iris(integration_cluster, monkeypatch):
                         environment=create_environment(env_vars=env_vars),
                     )
                 )
-                handle.wait(raise_on_failure=True)
+                handle.wait(raise_on_failure=True, stream_logs=True)
         else:
-            # Local cluster: run in-process (local filesystem is accessible).
+            logger.info("Running executor in-process (local mode)")
+            config = ExecutorMainConfig(
+                prefix=prefix,
+                executor_info_base_path=f"{prefix}/experiments",
+            )
+            steps = create_steps("quickstart-tests", synth_data)
             with set_current_client(iris_client):
                 executor_main(config, steps=steps)
+
+        logger.info("Marin pipeline completed successfully")
+    except Exception:
+        logger.exception("Marin pipeline failed")
+        sys.exit(1)
     finally:
         cleanup()
+
+
+if __name__ == "__main__":
+    main()