add timeout enforcement compliance test

Deploys a flow where a step exceeds its @timeout(seconds=5) and verifies the
run fails.  The existing test_timeout only checks that @timeout doesn't break
normal execution — it never tests that a step actually gets killed.

Adds wait_for_deployed_run_allow_failure() helper that returns the Run even
when it fails, so tests can assert on run.successful == False.

Author: Nissan Pow

test/ux/core/flows/basic/timeout_enforce_flow.py

@@ -0,0 +1,27 @@
+import time
+
+from metaflow import FlowSpec, step, timeout, project
+
+
+@project(name="timeout_enforce_flow")
+class TimeoutEnforceFlow(FlowSpec):
+    """Step exceeds its @timeout — the run MUST fail."""
+
+    @step
+    def start(self):
+        self.next(self.slow)
+
+    @timeout(seconds=5)
+    @step
+    def slow(self):
+        # Sleep well beyond the 5-second timeout to guarantee enforcement.
+        time.sleep(120)
+        self.next(self.end)
+
+    @step
+    def end(self):
+        pass
+
+
+if __name__ == "__main__":
+    TimeoutEnforceFlow()

test/ux/core/test_compliance.py

@@ -20,6 +20,7 @@
 from .test_utils import (
     deploy_flow_to_scheduler,
     wait_for_deployed_run,
+    wait_for_deployed_run_allow_failure,
 )
 
 
@@ -311,3 +312,45 @@ def test_nested_foreach_or_skip(
         f"Nested foreach produced wrong results: {all_results!r}. "
         "Expected ['x-1', 'y-1']."
     )
+
+
+# ---------------------------------------------------------------------------
+# test_timeout_enforcement
+#
+# WHY: The existing test_timeout only verifies that @timeout doesn't break
+# normal execution (step sleeps 1s with a 10-minute timeout — always passes).
+# If the timeout= kwarg on subprocess.run() is completely broken, that test
+# still passes.  This test deploys a flow where a step sleeps well beyond its
+# @timeout(seconds=5) and verifies the run actually fails.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.compliance
+@pytest.mark.scheduler_only
+def test_timeout_enforcement(exec_mode, decospecs, compute_env, tag, scheduler_config):
+    """A step that exceeds its @timeout must be killed — the run must fail."""
+    if exec_mode != "deployer":
+        pytest.skip("compliance test requires deployer mode")
+
+    test_unique_tag = f"test_compliance_timeout_enforce_{exec_mode}"
+    combined_tags = tag + [test_unique_tag]
+
+    tl_args = {
+        "env": compute_env,
+        "decospecs": decospecs,
+    }
+
+    deployed_flow = deploy_flow_to_scheduler(
+        flow_name="basic/timeout_enforce_flow.py",
+        tl_args=tl_args,
+        scheduler_args={"cluster": scheduler_config.cluster},
+        deploy_args={"tags": combined_tags, **(scheduler_config.deploy_args or {})},
+        scheduler_type=scheduler_config.scheduler_type,
+    )
+
+    run = wait_for_deployed_run_allow_failure(deployed_flow)
+
+    assert not run.successful, (
+        "Run should have failed because the 'slow' step exceeds its "
+        "@timeout(seconds=5), but it succeeded. Timeout enforcement may be broken."
+    )

test/ux/core/test_utils.py

@@ -199,6 +199,45 @@ def wait_for_deployed_run(
     return triggered_run.run
 
 
+def wait_for_deployed_run_allow_failure(
+    deployed_flow,
+    timeout: int = 3600,
+    run_kwargs: Optional[Dict[str, Any]] = None,
+    polling_interval: int = 3,
+):
+    """Trigger a deployed flow and wait for it to finish, even if it fails.
+
+    Same as wait_for_deployed_run but does NOT raise on failed status.
+    Returns the Run regardless of success/failure so the caller can assert
+    on run.successful.
+    """
+    print(f"Deployed flow {deployed_flow.name}")
+    run_kwargs = run_kwargs or {}
+    triggered_run = deployed_flow.trigger(**run_kwargs)
+
+    start_time = time.time()
+    while triggered_run.run is None:
+        if time.time() - start_time > timeout:
+            raise RuntimeError(f"Run failed to start within {timeout} seconds")
+        print("Waiting for run to start...")
+        time.sleep(polling_interval)
+
+    print(f"Run {triggered_run.run.id} started")
+
+    while not triggered_run.run.finished:
+        if time.time() - start_time > timeout:
+            raise RuntimeError(
+                f"Run {triggered_run.run.id} failed to complete within {timeout} seconds"
+            )
+        print(f"Waiting for run {triggered_run.run.id} to complete...")
+        time.sleep(polling_interval)
+
+    print(
+        f"Run {triggered_run.run.id} completed (successful={triggered_run.run.successful})"
+    )
+    return triggered_run.run
+
+
 def verify_run_provenance(run: Run, decospecs: Any) -> None:
     """Verify the run used the expected datastore and execution environment.