fix(iris): keep tasks in BUILDING until container reaches Running phase (#3106)

## Summary

- Tasks now stay in `BUILDING` until the container runtime reports
`ContainerPhase.RUNNING`, instead of transitioning immediately when
`run()` is called
- For process/docker runtimes this is instant (no behavioral change).
For K8s, tasks wait in BUILDING while pods are in
`Pending`/`ContainerCreating` phase.
- This makes `max_building_tasks_per_worker` actually effective — it now
gates concurrent pod creation instead of just counting tasks that
instantly flip to RUNNING

<details>
<summary>Context</summary>

Stacked on #3105 (ContainerPhase enum). Extracted from the [BUILDING
semantics investigation](#3090).

Without this fix, submitting 20+ tasks to a K8s worker creates all pods
simultaneously. The pod creation storm overwhelmed the controller (4 GiB
memory limit), causing it to crash ~15s after job submission. With
backpressure, only `max_building_tasks_per_worker` (default 4) pods are
created concurrently.

Validated on CW: [workflow run
22498485767](https://github.com/marin-community/marin/actions/runs/22498485767)
— 70+ task transitions observed with 5.2–10.4s BUILDING durations,
controller survived the entire run. See #3090 comment for full findings.

Related: #3102 (controller RBAC/scheduling), #3103 (worker kubectl
saturation)

</details>

## Test plan

- [x] `uv run pytest lib/iris/tests/cluster/worker/
lib/iris/tests/cluster/runtime/test_kubernetes_runtime.py` — 77 passed
- [x] Pre-commit clean
- [ ] CI
- [x] `test_building_backpressure.py` e2e test — passed (20 tasks, peak
building within limit)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: yonromai

lib/iris/src/iris/cluster/worker/task_attempt.py

@@ -433,6 +433,7 @@ def _download_bundle(self) -> None:
         """
         self.transition_to(cluster_pb2.TASK_STATE_BUILDING, message="downloading bundle")
         self.started_at = Timestamp.now()
+        self._building_start_monotonic = time.monotonic()
         self._report_state()  # Report BUILDING state to controller
 
         download_start = time.monotonic()
@@ -562,15 +563,9 @@ def _build_container(self) -> None:
             logger.info("Build phase completed for task %s", self.task_id)
 
     def _run_container(self) -> None:
-        """Start the main command during RUNNING state.
-
-        Non-blocking - returns immediately after starting.
-        """
+        """Start the container. Task stays in BUILDING until _monitor() confirms readiness."""
         assert self._container_handle is not None
 
-        self.transition_to(cluster_pb2.TASK_STATE_RUNNING)
-        self._report_state()
-
         self._container_handle.run()
         logger.info(
             "Container started for task %s (container_id=%s, ports=%s)",
@@ -625,6 +620,13 @@ def _monitor(self) -> None:
 
             # Check container status
             status = handle.status()
+
+            if self.status == cluster_pb2.TASK_STATE_BUILDING and status.phase == ContainerPhase.RUNNING:
+                building_duration = time.monotonic() - self._building_start_monotonic
+                logger.info("Task %s BUILDING→RUNNING after %.1fs", self.task_id, building_duration)
+                self.transition_to(cluster_pb2.TASK_STATE_RUNNING)
+                self._report_state()
+
             if status.phase == ContainerPhase.STOPPED:
                 logger.info(
                     "Container exited for task %s (container_id=%s, exit_code=%s, error=%s)",

lib/iris/tests/e2e/test_building_backpressure.py

@@ -0,0 +1,119 @@
+# Copyright 2025 The Marin Authors
+# SPDX-License-Identifier: Apache-2.0
+
+"""Backpressure test: slow-starting containers must not exceed max_building_tasks_per_worker."""
+
+import time
+from pathlib import Path
+
+import pytest
+from iris.client.client import IrisClient
+from iris.cluster.config import load_config, make_local_config
+from iris.cluster.manager import connect_cluster
+from iris.cluster.runtime.process import ProcessContainerHandle
+from iris.cluster.runtime.types import ContainerPhase, ContainerStatus
+from iris.rpc import cluster_pb2, config_pb2
+from iris.rpc.cluster_connect import ControllerServiceClientSync
+
+from .conftest import TestCluster
+from .helpers import _quick
+
+pytestmark = pytest.mark.e2e
+
+IRIS_ROOT = Path(__file__).resolve().parents[2]
+READY_DELAY = 3.0
+NUM_TASKS = 20
+MAX_BUILDING = 4  # default max_building_tasks_per_worker
+
+
+@pytest.fixture
+def single_worker_cluster():
+    """Single-worker cluster so building backpressure is observable."""
+    config = load_config(IRIS_ROOT / "examples" / "demo.yaml")
+    config.scale_groups.clear()
+    sg = config.scale_groups["local-cpu"]
+    sg.name = "local-cpu"
+    sg.accelerator_type = config_pb2.ACCELERATOR_TYPE_CPU
+    sg.num_vms = 1
+    sg.min_slices = 1
+    sg.max_slices = 1
+    sg.resources.cpu_millicores = 8000
+    sg.resources.memory_bytes = 16 * 1024**3
+    sg.resources.disk_bytes = 50 * 1024**3
+    sg.slice_template.local.SetInParent()
+    config = make_local_config(config)
+    with connect_cluster(config) as url:
+        client = IrisClient.remote(url, workspace=IRIS_ROOT)
+        controller_client = ControllerServiceClientSync(address=url, timeout_ms=30000)
+        tc = TestCluster(url=url, client=client, controller_client=controller_client)
+        tc.wait_for_workers(1, timeout=30)
+        yield tc
+        controller_client.close()
+
+
+def test_building_backpressure_with_slow_starting_containers(single_worker_cluster, monkeypatch):
+    """With slow-starting containers, at most max_building_tasks_per_worker tasks
+    should be in BUILDING/ASSIGNED at any time.
+
+    Monkeypatches ProcessContainerHandle.status() to return ContainerPhase.PENDING
+    for READY_DELAY seconds after run(), simulating K8S pod pending phase.
+    """
+    cluster = single_worker_cluster
+    original_run = ProcessContainerHandle.run
+    original_status = ProcessContainerHandle.status
+    run_times: dict[int, float] = {}
+
+    def patched_run(self):
+        run_times[id(self)] = time.monotonic()
+        return original_run(self)
+
+    def patched_status(self) -> ContainerStatus:
+        result = original_status(self)
+        run_time = run_times.get(id(self))
+        if run_time is not None and result.phase == ContainerPhase.RUNNING:
+            elapsed = time.monotonic() - run_time
+            if elapsed < READY_DELAY:
+                result = ContainerStatus(
+                    phase=ContainerPhase.PENDING,
+                    exit_code=result.exit_code,
+                    error=result.error,
+                    error_kind=result.error_kind,
+                    oom_killed=result.oom_killed,
+                )
+        return result
+
+    monkeypatch.setattr(ProcessContainerHandle, "run", patched_run)
+    monkeypatch.setattr(ProcessContainerHandle, "status", patched_status)
+
+    jobs = [cluster.submit(_quick, f"burst-{i}", cpu=0) for i in range(NUM_TASKS)]
+
+    peak_building = 0
+    all_done = False
+    deadline = time.monotonic() + 120
+
+    while not all_done and time.monotonic() < deadline:
+        building_count = 0
+        all_done = True
+        for job in jobs:
+            status = cluster.status(job)
+            for task in status.tasks:
+                if task.state in (cluster_pb2.TASK_STATE_BUILDING, cluster_pb2.TASK_STATE_ASSIGNED):
+                    building_count += 1
+            if not status.state or status.state in (
+                cluster_pb2.JOB_STATE_PENDING,
+                cluster_pb2.JOB_STATE_RUNNING,
+            ):
+                all_done = False
+
+        peak_building = max(peak_building, building_count)
+        time.sleep(0.3)
+
+    # Allow +1 slack for race between scheduling and heartbeat reporting
+    assert peak_building <= MAX_BUILDING + 1, (
+        f"Peak building count {peak_building} exceeded limit {MAX_BUILDING}. " f"Backpressure is not working."
+    )
+
+    # All jobs should eventually succeed
+    for job in jobs:
+        status = cluster.wait(job, timeout=120)
+        assert status.state == cluster_pb2.JOB_STATE_SUCCEEDED