refactor(iris): replace ContainerStatus bool with ContainerPhase enum (#3105)

- Replace `ContainerStatus.running: bool` with `ContainerStatus.phase:
ContainerPhase` where `ContainerPhase` is a `StrEnum` with `PENDING`,
`RUNNING`, `STOPPED`
- The old boolean conflated "not yet started" with "finished" — callers
had to inspect `exit_code` to disambiguate. The enum makes the state
machine explicit and enables the upcoming BUILDING→RUNNING backpressure
fix to distinguish K8s pods in Pending vs Running phase.
- All three runtime implementations (process, docker, kubernetes) and
all tests updated

<details>
<summary>Context</summary>

Extracted from the [BUILDING semantics investigation](#3090). The next
PR in the stack (BUILDING→RUNNING backpressure) depends on
`ContainerPhase.PENDING` vs `RUNNING` to gate task state transitions.

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

Author: yonromai

lib/iris/src/iris/cluster/runtime/__init__.py

@@ -15,6 +15,7 @@
 from iris.cluster.runtime.types import (
     ContainerConfig,
     ContainerHandle,
+    ContainerPhase,
     ContainerResult,
     ContainerRuntime,
     ContainerStats,
@@ -26,6 +27,7 @@
 __all__ = [
     "ContainerConfig",
     "ContainerHandle",
+    "ContainerPhase",
     "ContainerResult",
     "ContainerRuntime",
     "ContainerStats",

lib/iris/src/iris/cluster/runtime/docker.py

@@ -36,6 +36,7 @@
 from iris.cluster.runtime.types import (
     ContainerConfig,
     ContainerErrorKind,
+    ContainerPhase,
     ContainerStats,
     ContainerStatus,
     ImageInfo,
@@ -283,7 +284,7 @@ def build(self) -> list[LogLine]:
                     build_logs.extend(new_logs)
                     last_log_time = Timestamp.from_seconds(new_logs[-1].timestamp.timestamp()).add_ms(1)
 
-                if not status.running:
+                if status.phase == ContainerPhase.STOPPED:
                     break
                 time.sleep(0.5)
 
@@ -368,7 +369,7 @@ def stop(self, force: bool = False) -> None:
     def status(self) -> ContainerStatus:
         """Check container status (running, exit code, error)."""
         if not self._run_container_id:
-            return ContainerStatus(running=False, error="Container not started")
+            return ContainerStatus(phase=ContainerPhase.STOPPED, error="Container not started")
         return self._docker_inspect(self._run_container_id)
 
     def log_reader(self) -> DockerLogReader:
@@ -589,7 +590,7 @@ def _docker_inspect(self, container_id: str) -> ContainerStatus:
 
         if result.returncode != 0:
             return ContainerStatus(
-                running=False,
+                phase=ContainerPhase.STOPPED,
                 error=f"Container not found: id={container_id}",
                 error_kind=ContainerErrorKind.INFRA_NOT_FOUND,
             )
@@ -602,15 +603,15 @@ def _docker_inspect(self, container_id: str) -> ContainerStatus:
             oom_killed = state.get("OOMKilled", False)
 
             return ContainerStatus(
-                running=running,
+                phase=ContainerPhase.RUNNING if running else ContainerPhase.STOPPED,
                 exit_code=exit_code if not running else None,
                 error=error_msg,
                 error_kind=ContainerErrorKind.USER_CODE if error_msg else ContainerErrorKind.NONE,
                 oom_killed=oom_killed,
             )
         except (json.JSONDecodeError, KeyError) as e:
             return ContainerStatus(
-                running=False,
+                phase=ContainerPhase.STOPPED,
                 error=f"Failed to parse inspect output: {e}",
                 error_kind=ContainerErrorKind.RUNTIME_ERROR,
             )
@@ -661,7 +662,7 @@ def _docker_stats(self, container_id: str) -> ContainerStats:
     def _docker_kill(self, container_id: str, force: bool = False) -> None:
         """Kill container."""
         status = self._docker_inspect(container_id)
-        if not status.running:
+        if status.phase == ContainerPhase.STOPPED:
             return
 
         signal = "SIGKILL" if force else "SIGTERM"

lib/iris/src/iris/cluster/runtime/kubernetes.py

@@ -30,7 +30,13 @@
     resolve_cpu_spec,
     resolve_memory_spec,
 )
-from iris.cluster.runtime.types import ContainerConfig, ContainerErrorKind, ContainerStats, ContainerStatus
+from iris.cluster.runtime.types import (
+    ContainerConfig,
+    ContainerErrorKind,
+    ContainerPhase,
+    ContainerStats,
+    ContainerStatus,
+)
 from iris.cluster.worker.worker_types import LogLine
 from iris.rpc import cluster_pb2
 from iris.time_utils import Deadline, Duration
@@ -334,7 +340,7 @@ def stop(self, force: bool = False) -> None:
 
     def status(self) -> ContainerStatus:
         if not self._pod_name:
-            return ContainerStatus(running=False, error="Pod not started")
+            return ContainerStatus(phase=ContainerPhase.STOPPED, error="Pod not started")
 
         pod = self.kubectl.get_json("pod", self._pod_name)
         if pod is None:
@@ -350,11 +356,11 @@ def status(self) -> ContainerStatus:
                     self._pod_not_found_count,
                     _POD_NOT_FOUND_RETRY_COUNT,
                 )
-                return ContainerStatus(running=True)
+                return ContainerStatus(phase=ContainerPhase.PENDING)
 
             attempt = self.config.attempt_id if self.config.attempt_id is not None else -1
             return ContainerStatus(
-                running=False,
+                phase=ContainerPhase.STOPPED,
                 error=(
                     "Task pod not found after retry window: "
                     f"name={self._pod_name}, namespace={self.kubectl.namespace}, "
@@ -369,8 +375,10 @@ def status(self) -> ContainerStatus:
         self._pod_not_found_deadline = None
 
         phase = pod.get("status", {}).get("phase", "")
-        if phase in ("Pending", "Running"):
-            return ContainerStatus(running=True)
+        if phase == "Pending":
+            return ContainerStatus(phase=ContainerPhase.PENDING)
+        if phase == "Running":
+            return ContainerStatus(phase=ContainerPhase.RUNNING)
 
         statuses = pod.get("status", {}).get("containerStatuses", [])
         terminated = {}
@@ -388,7 +396,7 @@ def status(self) -> ContainerStatus:
             error = message or reason or None
         oom_killed = reason == "OOMKilled"
         return ContainerStatus(
-            running=False,
+            phase=ContainerPhase.STOPPED,
             exit_code=exit_code if isinstance(exit_code, int) else 1,
             error=error,
             oom_killed=oom_killed,

lib/iris/src/iris/cluster/runtime/process.py

@@ -42,7 +42,7 @@
     resolve_cpu_spec,
     resolve_memory_spec,
 )
-from iris.cluster.runtime.types import ContainerConfig, ContainerStats, ContainerStatus, RuntimeLogReader
+from iris.cluster.runtime.types import ContainerConfig, ContainerPhase, ContainerStats, ContainerStatus, RuntimeLogReader
 from iris.cluster.worker.worker_types import LogLine
 from iris.managed_thread import ManagedThread, get_thread_container
 from iris.rpc import cluster_pb2
@@ -515,9 +515,9 @@ def stop(self, force: bool = False) -> None:
     def status(self) -> ContainerStatus:
         """Check container status (running, exit code, error)."""
         if not self._container:
-            return ContainerStatus(running=False, error="Container not started")
+            return ContainerStatus(phase=ContainerPhase.STOPPED, error="Container not started")
         return ContainerStatus(
-            running=self._container._running,
+            phase=ContainerPhase.RUNNING if self._container._running else ContainerPhase.STOPPED,
             exit_code=self._container._exit_code,
             error=self._container._error,
         )

lib/iris/src/iris/cluster/runtime/types.py

@@ -35,6 +35,19 @@ class ContainerErrorKind(StrEnum):
     RUNTIME_ERROR = "runtime_error"
 
 
+class ContainerPhase(StrEnum):
+    """Lifecycle phase of a container from the runtime's perspective.
+
+    PENDING: container created but not yet executing (K8s pod scheduling, image pull).
+    RUNNING: container is executing the main command.
+    STOPPED: container has exited (check exit_code/error for details).
+    """
+
+    PENDING = "pending"
+    RUNNING = "running"
+    STOPPED = "stopped"
+
+
 @dataclass
 class ContainerConfig:
     """Configuration for running a container."""
@@ -91,7 +104,7 @@ class ContainerStats:
 class ContainerStatus:
     """Container state from runtime inspection."""
 
-    running: bool
+    phase: ContainerPhase
     exit_code: int | None = None
     error: str | None = None
     error_kind: ContainerErrorKind = ContainerErrorKind.NONE

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

@@ -22,6 +22,7 @@
     ContainerConfig,
     ContainerErrorKind,
     ContainerHandle,
+    ContainerPhase,
     ContainerRuntime,
     RuntimeLogReader,
 )
@@ -624,7 +625,7 @@ def _monitor(self) -> None:
 
             # Check container status
             status = handle.status()
-            if not status.running:
+            if status.phase == ContainerPhase.STOPPED:
                 logger.info(
                     "Container exited for task %s (container_id=%s, exit_code=%s, error=%s)",
                     self.task_id,

lib/iris/tests/cluster/runtime/test_kubernetes_runtime.py

@@ -14,7 +14,7 @@
 
 
 from iris.cluster.runtime.kubernetes import KubernetesRuntime
-from iris.cluster.runtime.types import ContainerConfig, ContainerErrorKind
+from iris.cluster.runtime.types import ContainerConfig, ContainerErrorKind, ContainerPhase
 from iris.rpc import cluster_pb2
 
 
@@ -233,11 +233,11 @@ def fake_get_json(resource: str, name: str):
     second = handle.status()
     third = handle.status()
 
-    assert first.running is True
+    assert first.phase == ContainerPhase.PENDING
     assert first.error is None
-    assert second.running is True
+    assert second.phase == ContainerPhase.PENDING
     assert second.error is None
-    assert third.running is True
+    assert third.phase == ContainerPhase.RUNNING
 
 
 def test_status_returns_structured_error_after_persistent_pod_not_found(monkeypatch):
@@ -259,9 +259,9 @@ def fake_get_json(resource: str, name: str):
     second = handle.status()
     third = handle.status()
 
-    assert first.running is True
-    assert second.running is True
-    assert third.running is False
+    assert first.phase == ContainerPhase.PENDING
+    assert second.phase == ContainerPhase.PENDING
+    assert third.phase == ContainerPhase.STOPPED
     assert third.error_kind == ContainerErrorKind.INFRA_NOT_FOUND
     assert third.error is not None
     assert "Task pod not found after retry window" in third.error

lib/iris/tests/cluster/worker/test_dashboard.py

@@ -15,7 +15,7 @@
 from iris.cluster.worker.bundle_cache import BundleCache
 from iris.cluster.worker.dashboard import WorkerDashboard
 from iris.cluster.runtime.docker import DockerRuntime
-from iris.cluster.runtime.types import ContainerStats, ContainerStatus
+from iris.cluster.runtime.types import ContainerPhase, ContainerStats, ContainerStatus
 from iris.cluster.worker.service import WorkerServiceImpl
 from iris.cluster.worker.worker import Worker, WorkerConfig
 from iris.rpc import cluster_pb2
@@ -48,7 +48,7 @@ def create_mock_container_handle():
     handle.container_id = "container123"
     handle.build = Mock(return_value=[])
     handle.run = Mock()
-    handle.status = Mock(return_value=ContainerStatus(running=False, exit_code=0))
+    handle.status = Mock(return_value=ContainerStatus(phase=ContainerPhase.STOPPED, exit_code=0))
     handle.stop = Mock()
     handle.logs = Mock(return_value=[])
     handle.stats = Mock(return_value=ContainerStats(memory_mb=100, cpu_percent=50, process_count=1, available=True))

lib/iris/tests/cluster/worker/test_worker.py

@@ -15,7 +15,7 @@
 from iris.cluster.types import Entrypoint, JobName
 from iris.cluster.worker.bundle_cache import BundleCache
 from iris.cluster.runtime.docker import DockerRuntime
-from iris.cluster.runtime.types import ContainerErrorKind, ContainerStats, ContainerStatus
+from iris.cluster.runtime.types import ContainerErrorKind, ContainerPhase, ContainerStats, ContainerStatus
 from iris.cluster.worker.port_allocator import PortAllocator
 from iris.cluster.worker.service import WorkerServiceImpl
 from iris.cluster.worker.worker import Worker, WorkerConfig
@@ -111,8 +111,8 @@ def create_mock_container_handle(
 
     if status_sequence is None:
         status_sequence = [
-            ContainerStatus(running=True),
-            ContainerStatus(running=False, exit_code=0),
+            ContainerStatus(phase=ContainerPhase.RUNNING),
+            ContainerStatus(phase=ContainerPhase.STOPPED, exit_code=0),
         ]
 
     call_count = [0]
@@ -260,7 +260,9 @@ def test_task_with_ports(worker):
 def test_task_failure_on_nonzero_exit(worker, mock_runtime):
     """Test task fails when container exits with non-zero code."""
     # Update the mock handle's status to return failure immediately
-    mock_handle = create_mock_container_handle(status_sequence=[ContainerStatus(running=False, exit_code=1)])
+    mock_handle = create_mock_container_handle(
+        status_sequence=[ContainerStatus(phase=ContainerPhase.STOPPED, exit_code=1)]
+    )
     mock_runtime.create_container = Mock(return_value=mock_handle)
 
     request = create_run_task_request()
@@ -280,8 +282,8 @@ def test_task_failure_on_error(worker, mock_runtime):
     # Update the mock handle's status to return error after first poll
     mock_handle = create_mock_container_handle(
         status_sequence=[
-            ContainerStatus(running=True),
-            ContainerStatus(running=False, exit_code=1, error="Container crashed"),
+            ContainerStatus(phase=ContainerPhase.RUNNING),
+            ContainerStatus(phase=ContainerPhase.STOPPED, exit_code=1, error="Container crashed"),
         ]
     )
     mock_runtime.create_container = Mock(return_value=mock_handle)
@@ -302,7 +304,7 @@ def test_task_infra_not_found_error_maps_to_worker_failed(worker, mock_runtime):
     mock_handle = create_mock_container_handle(
         status_sequence=[
             ContainerStatus(
-                running=False,
+                phase=ContainerPhase.STOPPED,
                 exit_code=1,
                 error="Task pod not found after retry window: name=iris-task-abc, namespace=iris",
                 error_kind=ContainerErrorKind.INFRA_NOT_FOUND,
@@ -351,7 +353,9 @@ def test_list_tasks(worker):
 def test_kill_running_task(worker, mock_runtime):
     """Test killing a running task with graceful timeout."""
     # Create a handle that stays running until killed
-    mock_handle = create_mock_container_handle(status_sequence=[ContainerStatus(running=True)] * 100)  # Stay running
+    mock_handle = create_mock_container_handle(
+        status_sequence=[ContainerStatus(phase=ContainerPhase.RUNNING)] * 100
+    )  # Stay running
     mock_runtime.create_container = Mock(return_value=mock_handle)
 
     request = create_run_task_request()
@@ -379,7 +383,9 @@ def test_kill_running_task(worker, mock_runtime):
 def test_new_attempt_supersedes_old(worker, mock_runtime):
     """New attempt for same task_id kills the old attempt and starts a new one."""
     # Create a handle that stays running until killed
-    mock_handle = create_mock_container_handle(status_sequence=[ContainerStatus(running=True)] * 100)  # Stay running
+    mock_handle = create_mock_container_handle(
+        status_sequence=[ContainerStatus(phase=ContainerPhase.RUNNING)] * 100
+    )  # Stay running
     mock_runtime.create_container = Mock(return_value=mock_handle)
 
     request_0 = create_run_task_request(task_id=JobName.root("retry-task").task(0).to_wire(), attempt_id=0)
@@ -419,7 +425,9 @@ def test_new_attempt_supersedes_old(worker, mock_runtime):
 def test_duplicate_attempt_rejected(worker, mock_runtime):
     """Same attempt_id for an existing non-terminal task is rejected."""
     # Create a handle that stays running until killed
-    mock_handle = create_mock_container_handle(status_sequence=[ContainerStatus(running=True)] * 100)  # Stay running
+    mock_handle = create_mock_container_handle(
+        status_sequence=[ContainerStatus(phase=ContainerPhase.RUNNING)] * 100
+    )  # Stay running
     mock_runtime.create_container = Mock(return_value=mock_handle)
 
     request = create_run_task_request(task_id=JobName.root("dup-task").task(0).to_wire(), attempt_id=0)

lib/iris/tests/e2e/test_coreweave_live_kubernetes_runtime.py

@@ -19,7 +19,7 @@
 from iris.cluster.config import load_config
 import fsspec.core
 from iris.cluster.runtime.kubernetes import KubernetesRuntime
-from iris.cluster.runtime.types import ContainerConfig
+from iris.cluster.runtime.types import ContainerConfig, ContainerPhase
 from iris.rpc import cluster_pb2
 
 pytestmark = [pytest.mark.e2e, pytest.mark.slow]
@@ -35,7 +35,7 @@ def _wait_finished(handle, timeout_seconds: float) -> cluster_pb2.TaskState:
     deadline = time.monotonic() + timeout_seconds
     while time.monotonic() < deadline:
         status = handle.status()
-        if not status.running:
+        if status.phase == ContainerPhase.STOPPED:
             return cluster_pb2.TASK_STATE_SUCCEEDED if status.exit_code == 0 else cluster_pb2.TASK_STATE_FAILED
         time.sleep(2.0)
     raise TimeoutError(f"pod {handle.container_id} did not finish in {timeout_seconds}s")
@@ -265,7 +265,7 @@ def test_incremental_log_reader_no_duplicates(coreweave_runtime: KubernetesRunti
         for line in new_lines:
             collected.append(line.data)
         status = handle.status()
-        if not status.running:
+        if status.phase == ContainerPhase.STOPPED:
             # One final read to drain remaining
             for line in reader.read():
                 collected.append(line.data)