fix(iris): use marin_temp_bucket for bundle storage (#3112)

- Add default_bundle_prefix() fallback in controller CLI using
marin_temp_bucket(ttl_days=7, prefix="iris/bundles") when no explicit
bundle_prefix is configured
- Remove hardcoded gs://marin-us-central2/tmp/iris/bundles from GCP
example configs (smoke.yaml, demo.yaml, marin.yaml)
- Bundles now go to gs://marin-tmp-{region}/ttl=7d/iris/bundles
(lifecycle-managed) instead of the main data bucket

Fixes #3068

Author: claude[bot]

lib/iris/examples/demo.yaml

@@ -21,9 +21,6 @@ defaults:
     port: 10001
     controller_address: "${IRIS_CONTROLLER_ADDRESS}"
 
-storage:
-  bundle_prefix: gs://marin-us-central2/tmp/iris/bundles
-
 controller:
   image: ghcr.io/marin-community/iris-controller:latest
   gcp:

lib/iris/examples/marin.yaml

@@ -19,9 +19,6 @@ defaults:
     default_task_image: ghcr.io/marin-community/iris-task:latest
     port: 10001
 
-storage:
-  bundle_prefix: gs://marin-us-central2/tmp/iris/bundles
-
 controller:
   image: ghcr.io/marin-community/iris-controller:latest
   gcp:

lib/iris/examples/smoke.yaml

@@ -19,9 +19,6 @@ defaults:
     default_task_image: ghcr.io/marin-community/iris-task:latest
     port: 10001
 
-storage:
-  bundle_prefix: gs://marin-us-central2/tmp/iris/bundles
-
 controller:
   image: ghcr.io/marin-community/iris-controller:latest
   gcp:

lib/iris/src/iris/cluster/controller/main.py

@@ -14,11 +14,21 @@
 from iris.cluster.controller.controller import Controller, ControllerConfig, RpcWorkerStubFactory
 from iris.cluster.controller.state import HEARTBEAT_FAILURE_THRESHOLD
 from iris.logging import configure_logging
+from iris.marin_fs import marin_temp_bucket
 from iris.time_utils import Duration
 
 logger = logging.getLogger(__name__)
 
 
+def default_bundle_prefix() -> str:
+    """Return a region-local temp bucket path for bundle storage.
+
+    Uses the same marin_temp_bucket API that log_prefix uses, with a 7-day TTL
+    since bundles are ephemeral and regenerated on each job submission.
+    """
+    return marin_temp_bucket(ttl_days=7, prefix="iris/bundles")
+
+
 @click.group()
 def cli():
     """Iris Controller - Cluster control plane."""
@@ -108,6 +118,10 @@ def serve(
         cluster_config.controller.heartbeat_failure_threshold if cluster_config else HEARTBEAT_FAILURE_THRESHOLD
     )
 
+    if bundle_prefix is None:
+        bundle_prefix = default_bundle_prefix()
+        logger.info("Using auto-detected bundle_prefix: %s", bundle_prefix)
+
     logger.info("Configuration: host=%s port=%d bundle_prefix=%s", host, port, bundle_prefix)
     logger.info("Configuration: scheduler_interval=%.2fs", scheduler_interval)