From 28c0868c40ef008f242c64127271a97589ea3bec Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Mon, 20 Jul 2026 04:30:36 +0000
Subject: [PATCH 01/10] [TRTLLM-13409][fix] hard-kill all ranks when one rank's
 executor loop crashes

When a rank's executor loop dies on an exception, the rank stops
participating in collectives but nothing tells its peers: every peer
blocks in its next collective until its own HangDetector fires 300s
later, and the whole multi-GPU test session burns that long for an
error that was already known (the A4 AutoDeploy catches are this
signature: peers crash, the survivor wedges in ADP until the 300s
backstop).

Escalate at error time instead: after the loop's local cleanup has
woken rank-local waiters, hard-kill the world via the ST-1 propagation
path. A grace period (TLLM_RANK_CRASH_HARD_KILL_GRACE, default 10s,
negative disables) lets the cleaner error paths win the race first --
the stashed error reaches rank-local response waiters, the init-phase
ready handshake returns the real exception to the proxy, and the worker
future completes with the original error -- so the client reports the
actual failure rather than a bare worker death.

Single-rank worlds are exempt (no peers to unblock), and the kill
helper never raises (it runs in a finally where an exception would mask
the loop's original error).

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/pyexecutor/hang_detector.py        |  67 ++++++++++
 tensorrt_llm/_torch/pyexecutor/py_executor.py |  13 +-
 .../executor/test_hang_detector_kill.py       | 126 +++++++++++++++++-
 3 files changed, 204 insertions(+), 2 deletions(-)

diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
index f6dde7c58959..d3cdd211740a 100644
--- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py
+++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
@@ -17,6 +17,7 @@
 import signal
 import sys
 import threading
+import time
 from contextlib import contextmanager
 from typing import Callable, Optional
 
@@ -26,6 +27,11 @@
 # 137 == 128 + SIGKILL(9): the exit code a shell reports for a SIGKILL'd process.
 _HARD_KILL_EXIT_CODE = 137
 
+# Grace (seconds) between a rank's executor-loop crash and the hard kill of the
+# whole world. Negative disables the kill entirely (escape hatch).
+RANK_CRASH_KILL_GRACE_ENV = "TLLM_RANK_CRASH_HARD_KILL_GRACE"
+_RANK_CRASH_KILL_GRACE_DEFAULT = 10.0
+
 
 def _best_effort_flush_streams() -> None:
     """Flush stdout/stderr without ever raising; diagnostics must not block hard kill."""
@@ -80,6 +86,67 @@ def propagate_hard_kill(exit_code: int = _HARD_KILL_EXIT_CODE) -> None:
     os.kill(os.getpid(), signal.SIGKILL)
 
 
+def _rank_crash_kill_grace() -> Optional[float]:
+    """Resolve the crash-kill grace period; ``None`` means the kill is disabled."""
+    raw = os.environ.get(RANK_CRASH_KILL_GRACE_ENV)
+    if raw is None:
+        return _RANK_CRASH_KILL_GRACE_DEFAULT
+    try:
+        grace = float(raw)
+    except ValueError:
+        _best_effort_log_error(
+            f"Invalid {RANK_CRASH_KILL_GRACE_ENV}={raw!r}; "
+            f"using default {_RANK_CRASH_KILL_GRACE_DEFAULT}s"
+        )
+        return _RANK_CRASH_KILL_GRACE_DEFAULT
+    return None if grace < 0 else grace
+
+
+def hard_kill_on_rank_crash(world_size: int) -> bool:
+    """Hard-kill the whole world after this rank's executor loop crashed.
+
+    A rank whose executor loop died on an exception can never rejoin its
+    peers' collectives: without an explicit kill, every peer blocks in its
+    next collective until its own HangDetector fires (300 s), and the whole
+    test session burns that long for an error that was already known.
+
+    The grace sleep before the kill is load-bearing: it gives the crashed
+    rank's cleaner error paths time to win the race, so the client reports
+    the ORIGINAL exception instead of a bare worker death —
+    - rank-local response waiters woken by the executor-loop cleanup read
+      the stashed error and surface it through the response path;
+    - during init, the worker's ready handshake returns the real error to
+      the proxy before the abort tears the world down;
+    - the worker main thread returning lets its mpi4py future complete with
+      the original exception.
+
+    Never raises (it runs in a ``finally`` where an exception would mask the
+    original loop error). Returns True when the kill path was taken — only
+    observable in tests, where ``propagate_hard_kill`` is stubbed; in
+    production that call does not return.
+    """
+    try:
+        if world_size <= 1:
+            # No peers to unblock; the worker's own death already completes
+            # its future/handshake with the original exception.
+            return False
+        grace = _rank_crash_kill_grace()
+        if grace is None:
+            return False
+        _best_effort_log_error(
+            f"Executor loop crashed on this rank; hard-killing all "
+            f"{world_size} ranks in {grace}s (peers cannot make progress "
+            f"without this rank). Set {RANK_CRASH_KILL_GRACE_ENV}=-1 to disable."
+        )
+        if grace > 0:
+            time.sleep(grace)
+        propagate_hard_kill()
+        return True
+    except Exception as e:  # noqa: BLE001 - must not mask the loop's original error
+        _best_effort_log_error(f"hard_kill_on_rank_crash failed (ignored): {e!r}")
+        return False
+
+
 class HangDetector:
     """Watchdog that fires when the executor loop stops checkpointing.
 
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index 6e3252b48688..175e122e8c61 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -66,7 +66,8 @@
 from .guided_decoder import GuidedDecoder
 from .handle_additional_outputs import HandleAdditionalOutputs
 from .handle_logits import HandleLogits
-from .hang_detector import HangDetector, propagate_hard_kill
+from .hang_detector import (HangDetector, hard_kill_on_rank_crash,
+                            propagate_hard_kill)
 from .kv_cache_manager_v2 import KVCacheManagerV2
 from .kv_cache_stats import append_kv_cache_iteration_stats
 from .kv_cache_transceiver import (KvCacheTransceiver,
@@ -1185,6 +1186,7 @@ def _flush_iter_stats_synced(self):
     # Performance metrics methods are in PerfMetricsManager (self.perf_manager)
 
     def _event_loop_wrapper(self):
+        crashed = False
         try:
             # Skip line profiler during warmup/memory estimation phase to avoid
             # saving incomplete results that would be overwritten anyway
@@ -1194,6 +1196,7 @@ def _event_loop_wrapper(self):
                  customized_gc_thresholds(self.garbage_collection_gen0_threshold):
                 self.event_loop()
         except Exception as e:
+            crashed = True
             logger.error(f"Error in event loop: {e}")
             logger.error(traceback.format_exc())
             # Stash the original error so local consumers
@@ -1207,6 +1210,14 @@ def _event_loop_wrapper(self):
             raise e
         finally:
             self._executor_loop_cleanup()
+            if crashed:
+                # Peers cannot make progress without this rank's loop: they
+                # would block in their next collective until their own
+                # HangDetectors fire 300s later. Kill the world now instead;
+                # the grace inside lets the stashed error reach rank-local
+                # waiters and the ready handshake first, so the client sees
+                # the original exception rather than a bare worker death.
+                hard_kill_on_rank_crash(self.dist.world_size)
 
     @property
     def is_warmup(self) -> bool:
diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index 0962df441cc6..800c14a9c65c 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -15,14 +15,22 @@
 """HangDetector timer behavior and the hard-kill propagation mechanism (no GPU)."""
 
 import asyncio
+import contextlib
 import os
 import signal
 import subprocess
 import sys
 import time
+import types
+
+import pytest
 
 from tensorrt_llm._torch.pyexecutor import hang_detector as hang_detector_module
-from tensorrt_llm._torch.pyexecutor.hang_detector import HangDetector
+from tensorrt_llm._torch.pyexecutor.hang_detector import (
+    RANK_CRASH_KILL_GRACE_ENV,
+    HangDetector,
+    hard_kill_on_rank_crash,
+)
 
 
 def test_detector_fires_after_timeout():
@@ -117,3 +125,119 @@ def test_propagate_hard_kill_self_sigkills_without_mpi():
         f"expected self-SIGKILL (-9), got {proc.returncode}; "
         f"stderr={proc.stderr.decode(errors='replace')[-500:]}"
     )
+
+
+# --------------------------------------------------------------------------
+# hard_kill_on_rank_crash: a rank whose executor loop crashed must kill the
+# world (after a grace) instead of leaving peers to burn 300s in collectives.
+# --------------------------------------------------------------------------
+
+
+def test_rank_crash_kill_single_rank_is_noop(monkeypatch):
+    """No peers to unblock: the worker's own death already carries the error."""
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    assert hard_kill_on_rank_crash(world_size=1) is False
+    assert kills == []
+
+
+def test_rank_crash_kill_fires_for_multi_rank(monkeypatch):
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
+    assert hard_kill_on_rank_crash(world_size=4) is True
+    assert kills == [1]
+
+
+def test_rank_crash_kill_sleeps_grace_before_kill(monkeypatch):
+    """The grace must elapse BEFORE the kill so cleaner error paths win the race."""
+    order = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "2.5")
+    assert hard_kill_on_rank_crash(world_size=2) is True
+    assert order == [("sleep", 2.5), "kill"]
+
+
+def test_rank_crash_kill_disabled_by_negative_grace(monkeypatch):
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "-1")
+    assert hard_kill_on_rank_crash(world_size=8) is False
+    assert kills == []
+
+
+def test_rank_crash_kill_invalid_grace_uses_default(monkeypatch):
+    """A malformed env value must not disable the kill (fail-safe default)."""
+    order = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "bogus")
+    assert hard_kill_on_rank_crash(world_size=2) is True
+    assert order == [("sleep", 10.0), "kill"]
+
+
+def test_rank_crash_kill_never_raises(monkeypatch):
+    """It runs in a `finally`: raising would mask the loop's original error."""
+
+    def boom():
+        raise RuntimeError("abort machinery broken")
+
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", boom)
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
+    assert hard_kill_on_rank_crash(world_size=2) is False
+
+
+# --------------------------------------------------------------------------
+# Wiring: PyExecutor._event_loop_wrapper must invoke the kill on the crash
+# path only, and only after local cleanup has woken rank-local waiters.
+# --------------------------------------------------------------------------
+
+
+def _bare_executor(pe, monkeypatch, world_size):
+    # Neutralize the profiling/GC context managers: they are irrelevant to the
+    # crash path and must not depend on env/GC state in a unit test.
+    monkeypatch.setattr(pe, "host_profiler_context", lambda enable: contextlib.nullcontext())
+    monkeypatch.setattr(pe, "customized_gc_thresholds", lambda threshold: contextlib.nullcontext())
+    ex = pe.PyExecutor.__new__(pe.PyExecutor)
+    ex.dist = types.SimpleNamespace(world_size=world_size)
+    ex.garbage_collection_gen0_threshold = None
+    return ex
+
+
+def test_event_loop_wrapper_kills_world_on_crash(monkeypatch):
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    monkeypatch.setattr(
+        pe, "hard_kill_on_rank_crash", lambda world_size: events.append(("kill", world_size))
+    )
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+
+    def crash():
+        raise ValueError("boom")
+
+    ex.event_loop = crash
+
+    with pytest.raises(ValueError, match="boom"):
+        ex._event_loop_wrapper()
+
+    # Cleanup wakes rank-local waiters (who read the stashed error) BEFORE
+    # the world is torn down.
+    assert events == ["cleanup", ("kill", 4)]
+    assert isinstance(ex._event_loop_error, ValueError)
+
+
+def test_event_loop_wrapper_no_kill_on_clean_exit(monkeypatch):
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    monkeypatch.setattr(pe, "hard_kill_on_rank_crash", lambda world_size: events.append("kill"))
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+    ex.event_loop = lambda: None
+
+    ex._event_loop_wrapper()
+
+    assert events == ["cleanup"]

From 3a2e7a754355e2e405a407ba92269e30a9d78fc7 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Wed, 22 Jul 2026 03:51:40 +0000
Subject: [PATCH 02/10] [TRTLLM-13409][fix] keep rank-crash hard kill reachable
 when cleanup blocks or raises

The kill sat after _executor_loop_cleanup() in the finally block, so it
was skippable in exactly the situations it exists for: cleanup blocking
without bound (unbounded wait() on a PP send handle wedged by the crash)
or cleanup raising (aborting the finally before the kill). Either way
peers fall back to burning 300s in their own HangDetectors.

Arm a daemon watchdog thread BEFORE cleanup that fires the kill at
crash + grace regardless of cleanup progress, and nest the post-cleanup
kill in its own finally so a raising cleanup cannot skip it.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/pyexecutor/hang_detector.py        | 33 +++++++
 tensorrt_llm/_torch/pyexecutor/py_executor.py | 28 ++++--
 .../executor/test_hang_detector_kill.py       | 98 +++++++++++++++++--
 3 files changed, 143 insertions(+), 16 deletions(-)

diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
index d3cdd211740a..52d11dbf7163 100644
--- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py
+++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
@@ -147,6 +147,39 @@ def hard_kill_on_rank_crash(world_size: int) -> bool:
         return False
 
 
+def start_rank_crash_kill_watchdog(world_size: int) -> Optional[threading.Thread]:
+    """Arm a daemon thread that hard-kills the world once the grace elapses.
+
+    Must be armed BEFORE executor-loop cleanup: cleanup can block without
+    bound (e.g. ``wait()`` on a pending PP send handle wedged by the crash),
+    and a kill placed after it would never be reached — leaving peers to
+    burn in their own 300 s HangDetectors, the exact failure this kill
+    exists to avoid. The thread reuses ``hard_kill_on_rank_crash``, so the
+    kill fires at crash + grace whether cleanup finishes, blocks, or raises.
+
+    Never raises. Returns the armed thread, or ``None`` when the kill is
+    not applicable (single rank, disabled by env) or the thread could not
+    be started — in that case the caller's post-cleanup kill remains the
+    only mechanism.
+    """
+    try:
+        if world_size <= 1:
+            return None
+        if _rank_crash_kill_grace() is None:
+            return None
+        watchdog = threading.Thread(
+            target=hard_kill_on_rank_crash,
+            args=(world_size,),
+            name="rank_crash_kill_watchdog",
+            daemon=True,
+        )
+        watchdog.start()
+        return watchdog
+    except Exception as e:  # noqa: BLE001 - must not mask the loop's original error
+        _best_effort_log_error(f"failed to arm rank-crash kill watchdog (ignored): {e!r}")
+        return None
+
+
 class HangDetector:
     """Watchdog that fires when the executor loop stops checkpointing.
 
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index 175e122e8c61..a9e3b70d6e4d 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -67,7 +67,7 @@
 from .handle_additional_outputs import HandleAdditionalOutputs
 from .handle_logits import HandleLogits
 from .hang_detector import (HangDetector, hard_kill_on_rank_crash,
-                            propagate_hard_kill)
+                            propagate_hard_kill, start_rank_crash_kill_watchdog)
 from .kv_cache_manager_v2 import KVCacheManagerV2
 from .kv_cache_stats import append_kv_cache_iteration_stats
 from .kv_cache_transceiver import (KvCacheTransceiver,
@@ -1209,15 +1209,25 @@ def _event_loop_wrapper(self):
             self._event_loop_error = e
             raise e
         finally:
-            self._executor_loop_cleanup()
             if crashed:
-                # Peers cannot make progress without this rank's loop: they
-                # would block in their next collective until their own
-                # HangDetectors fire 300s later. Kill the world now instead;
-                # the grace inside lets the stashed error reach rank-local
-                # waiters and the ready handshake first, so the client sees
-                # the original exception rather than a bare worker death.
-                hard_kill_on_rank_crash(self.dist.world_size)
+                # Armed BEFORE cleanup: cleanup can block without bound on a
+                # send handle wedged by the crash, and a kill placed only
+                # after it would never fire.
+                start_rank_crash_kill_watchdog(self.dist.world_size)
+            try:
+                self._executor_loop_cleanup()
+            finally:
+                if crashed:
+                    # Peers cannot make progress without this rank's loop:
+                    # they would block in their next collective until their
+                    # own HangDetectors fire 300s later. Kill the world now
+                    # instead; the grace inside lets the stashed error reach
+                    # rank-local waiters and the ready handshake first, so
+                    # the client sees the original exception rather than a
+                    # bare worker death. Nested finally: the kill must fire
+                    # even when cleanup itself raises; the watchdog above
+                    # covers cleanup blocking.
+                    hard_kill_on_rank_crash(self.dist.world_size)
 
     @property
     def is_warmup(self) -> bool:
diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index 800c14a9c65c..b497440ffd02 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -20,6 +20,7 @@
 import signal
 import subprocess
 import sys
+import threading
 import time
 import types
 
@@ -30,6 +31,7 @@
     RANK_CRASH_KILL_GRACE_ENV,
     HangDetector,
     hard_kill_on_rank_crash,
+    start_rank_crash_kill_watchdog,
 )
 
 
@@ -188,6 +190,45 @@ def boom():
     assert hard_kill_on_rank_crash(world_size=2) is False
 
 
+# --------------------------------------------------------------------------
+# start_rank_crash_kill_watchdog: the kill must fire even when executor-loop
+# cleanup never returns (e.g. blocked on a PP send handle wedged by the
+# crash), so it is armed in a daemon thread BEFORE cleanup starts.
+# --------------------------------------------------------------------------
+
+
+def test_watchdog_kills_while_caller_blocks(monkeypatch):
+    """The kill fires from the watchdog thread with no help from the caller."""
+    killed = threading.Event()
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", killed.set)
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
+
+    watchdog = start_rank_crash_kill_watchdog(world_size=2)
+
+    assert watchdog is not None
+    assert watchdog.daemon  # must never block interpreter exit
+    # The caller does nothing further (it would be blocked in cleanup);
+    # the kill must fire regardless.
+    assert killed.wait(timeout=30.0)
+    watchdog.join(timeout=30.0)
+
+
+def test_watchdog_not_armed_for_single_rank(monkeypatch):
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
+    assert start_rank_crash_kill_watchdog(world_size=1) is None
+    assert kills == []
+
+
+def test_watchdog_not_armed_when_disabled(monkeypatch):
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "-1")
+    assert start_rank_crash_kill_watchdog(world_size=8) is None
+    assert kills == []
+
+
 # --------------------------------------------------------------------------
 # Wiring: PyExecutor._event_loop_wrapper must invoke the kill on the crash
 # path only, and only after local cleanup has woken rank-local waiters.
@@ -205,13 +246,22 @@ def _bare_executor(pe, monkeypatch, world_size):
     return ex
 
 
+def _stub_kill_paths(pe, monkeypatch, events):
+    monkeypatch.setattr(
+        pe, "hard_kill_on_rank_crash", lambda world_size: events.append(("kill", world_size))
+    )
+    monkeypatch.setattr(
+        pe,
+        "start_rank_crash_kill_watchdog",
+        lambda world_size: events.append(("watchdog", world_size)),
+    )
+
+
 def test_event_loop_wrapper_kills_world_on_crash(monkeypatch):
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
     events = []
-    monkeypatch.setattr(
-        pe, "hard_kill_on_rank_crash", lambda world_size: events.append(("kill", world_size))
-    )
+    _stub_kill_paths(pe, monkeypatch, events)
     ex = _bare_executor(pe, monkeypatch, world_size=4)
     ex._executor_loop_cleanup = lambda: events.append("cleanup")
 
@@ -223,9 +273,43 @@ def crash():
     with pytest.raises(ValueError, match="boom"):
         ex._event_loop_wrapper()
 
-    # Cleanup wakes rank-local waiters (who read the stashed error) BEFORE
-    # the world is torn down.
-    assert events == ["cleanup", ("kill", 4)]
+    # The watchdog is armed BEFORE cleanup (cleanup can block forever);
+    # cleanup wakes rank-local waiters (who read the stashed error) BEFORE
+    # the direct kill tears the world down.
+    assert events == [("watchdog", 4), "cleanup", ("kill", 4)]
+    assert isinstance(ex._event_loop_error, ValueError)
+
+
+def test_event_loop_wrapper_kills_world_when_cleanup_raises(monkeypatch):
+    """The kill must not be skippable by a cleanup failure.
+
+    Cleanup runs precisely when the process is already unhealthy; if its
+    exception aborted the finally block before the kill, peers would burn
+    300s in their HangDetectors — the worst case is exactly when the kill
+    matters most.
+    """
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+
+    def broken_cleanup():
+        events.append("cleanup")
+        raise RuntimeError("cleanup exploded")
+
+    ex._executor_loop_cleanup = broken_cleanup
+
+    def crash():
+        raise ValueError("boom")
+
+    ex.event_loop = crash
+
+    with pytest.raises(RuntimeError, match="cleanup exploded"):
+        ex._event_loop_wrapper()
+
+    assert events == [("watchdog", 4), "cleanup", ("kill", 4)]
+    # The original loop error stays reachable for rank-local consumers.
     assert isinstance(ex._event_loop_error, ValueError)
 
 
@@ -233,7 +317,7 @@ def test_event_loop_wrapper_no_kill_on_clean_exit(monkeypatch):
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
     events = []
-    monkeypatch.setattr(pe, "hard_kill_on_rank_crash", lambda world_size: events.append("kill"))
+    _stub_kill_paths(pe, monkeypatch, events)
     ex = _bare_executor(pe, monkeypatch, world_size=4)
     ex._executor_loop_cleanup = lambda: events.append("cleanup")
     ex.event_loop = lambda: None

From 8b5a3587cec078735a10fea88f7172f27cf49ab4 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Wed, 22 Jul 2026 03:51:57 +0000
Subject: [PATCH 03/10] [https://nvbugs/6480574][fix] set the attributes
 shutdown() reads in pool-session shutdown test

The test builds MpiPoolSession via __new__ (a real spawn is neither
needed nor wanted), but MpiPoolSession.shutdown() now reads n_workers
(added by #16456 while the test was in flight) and _wait_shutdown, both
set only in __init__. Provide them explicitly and drop the waive.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 tests/unittest/executor/test_proxy_fast_death.py | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/tests/unittest/executor/test_proxy_fast_death.py b/tests/unittest/executor/test_proxy_fast_death.py
index c72ed8c21b1a..f428a7862d90 100644
--- a/tests/unittest/executor/test_proxy_fast_death.py
+++ b/tests/unittest/executor/test_proxy_fast_death.py
@@ -522,6 +522,10 @@ def test_pool_session_shutdown_never_blocks_after_release():
     from tensorrt_llm.llmapi.mpi_session import MpiPoolSession
 
     session = MpiPoolSession.__new__(MpiPoolSession)
+    # __new__ bypasses __init__ (which would spawn real MPI workers), so the
+    # attributes shutdown() reads must be provided here.
+    session.n_workers = 2
+    session._wait_shutdown = False
     pool = _Mock()
     pool._pool.thread = None  # no real manager thread to deregister
     session.mpi_pool = pool

From 81fd1b64771af98a221309fb3b58f4d58f772430 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 15:42:53 +0000
Subject: [PATCH 04/10] [TRTLLM-13409][fix] scope the rank-crash hard kill to
 crashes that strand peers

The kill was armed for ANY exception escaping the executor loop wrapper,
including one raised on the way out of an already-shutting-down loop.
A hardware A/B on 2x GB200 (mpirun -n 2 trtllm-llmapi-launch trtllm-bench
--tp 2) showed it turning a benign late exception -- raised after
event_loop() returned, i.e. after the shutdown request was processed and
all work was done -- into a whole-job SIGKILL: mpirun exit 137 at
crash+10s, where the same injection on the merge-base logged the error,
let the worker thread die, and completed the benchmark with exit 0. It
fired during the memory-profiling dry run, before the benchmark even
started: PyExecutor is constructed and shut down twice per process and
that first shutdown is a normal lifecycle event.

Three changes:

- crashed = not self.is_shutdown. Once is_shutdown is set every rank has
  processed the shutdown broadcast, so no peer is waiting on this one and
  there is nothing to escalate.
- Scope the flag to event_loop() itself via an inner try. Teardown of the
  enclosing host-profiler / GC context managers is not a stranded-peer
  condition either.
- Make the watchdog cancellable and give it an explicit deadline. It was
  a bare daemon thread that slept the grace and killed with no cancel
  path, so once armed it fired at crash+10s even if the process went on
  to shut down cleanly and would have exited 0. It also double-armed: the
  watchdog and the post-cleanup hard_kill_on_rank_crash each ran their
  own timer. The watchdog now covers only the window where cleanup may
  block forever; once cleanup returns the caller cancels it and carries
  the kill inline on the watchdog's ORIGINAL deadline, so exactly one
  timer is live at any moment and the handover cannot push the kill out
  by a second grace.

A genuine mid-loop crash still arms the watchdog and still hard-kills the
world -- the behavior the kill exists for is unchanged.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/pyexecutor/hang_detector.py        |  88 +++++++--
 tensorrt_llm/_torch/pyexecutor/py_executor.py |  44 ++++-
 .../executor/test_hang_detector_kill.py       | 172 ++++++++++++++++--
 3 files changed, 268 insertions(+), 36 deletions(-)

diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
index 52d11dbf7163..b7ef25a5aa95 100644
--- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py
+++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
@@ -102,7 +102,32 @@ def _rank_crash_kill_grace() -> Optional[float]:
     return None if grace < 0 else grace
 
 
-def hard_kill_on_rank_crash(world_size: int) -> bool:
+def _wait_out_kill_grace(
+    grace: float,
+    deadline: Optional[float],
+    cancelled: Optional[threading.Event],
+) -> bool:
+    """Sleep out the crash-kill grace; return False if the kill was cancelled.
+
+    ``deadline`` (a ``time.monotonic()`` stamp) lets a kill that was already
+    armed elsewhere keep its ORIGINAL fire time when it is handed over to
+    another waiter, so a handover cannot push the kill out by a second grace.
+    ``cancelled`` makes the wait interruptible: a rank that turns out not to
+    need the kill can disarm it instead of being killed while exiting cleanly.
+    """
+    remaining = grace if deadline is None else max(0.0, deadline - time.monotonic())
+    if cancelled is None:
+        if remaining > 0:
+            time.sleep(remaining)
+        return True
+    return not cancelled.wait(remaining)
+
+
+def hard_kill_on_rank_crash(
+    world_size: int,
+    deadline: Optional[float] = None,
+    cancelled: Optional[threading.Event] = None,
+) -> bool:
     """Hard-kill the whole world after this rank's executor loop crashed.
 
     A rank whose executor loop died on an exception can never rejoin its
@@ -123,7 +148,9 @@ def hard_kill_on_rank_crash(world_size: int) -> bool:
     Never raises (it runs in a ``finally`` where an exception would mask the
     original loop error). Returns True when the kill path was taken — only
     observable in tests, where ``propagate_hard_kill`` is stubbed; in
-    production that call does not return.
+    production that call does not return. Returns False when the kill does
+    not apply (single rank, disabled by env) or was cancelled during the
+    grace.
     """
     try:
         if world_size <= 1:
@@ -138,8 +165,9 @@ def hard_kill_on_rank_crash(world_size: int) -> bool:
             f"{world_size} ranks in {grace}s (peers cannot make progress "
             f"without this rank). Set {RANK_CRASH_KILL_GRACE_ENV}=-1 to disable."
         )
-        if grace > 0:
-            time.sleep(grace)
+        if not _wait_out_kill_grace(grace, deadline, cancelled):
+            _best_effort_log_error("Rank-crash hard kill cancelled before the grace elapsed.")
+            return False
         propagate_hard_kill()
         return True
     except Exception as e:  # noqa: BLE001 - must not mask the loop's original error
@@ -147,7 +175,40 @@ def hard_kill_on_rank_crash(world_size: int) -> bool:
         return False
 
 
-def start_rank_crash_kill_watchdog(world_size: int) -> Optional[threading.Thread]:
+class RankCrashKillWatchdog(threading.Thread):
+    """Daemon thread that hard-kills the world once the crash grace elapses.
+
+    A plain ``Thread`` for backwards compatibility (callers may still join it
+    or inspect ``daemon``), plus the two things the caller needs to own the
+    kill deadline:
+
+    - ``cancel()`` disarms the timer. Without it an armed watchdog fires at
+      crash + grace unconditionally, so a rank that ends up exiting cleanly
+      is SIGKILLed anyway and a would-be exit 0 becomes exit 137.
+    - ``deadline`` exposes the original fire time so whoever takes the kill
+      over after cancelling still fires at crash + grace rather than
+      restarting the clock.
+    """
+
+    def __init__(self, world_size: int, grace: float):
+        super().__init__(name="rank_crash_kill_watchdog", daemon=True)
+        self._world_size = world_size
+        self.deadline = time.monotonic() + max(0.0, grace)
+        self._cancelled = threading.Event()
+
+    def cancel(self) -> None:
+        """Disarm the kill. Never raises; safe to call more than once."""
+        self._cancelled.set()
+
+    @property
+    def cancelled(self) -> bool:
+        return self._cancelled.is_set()
+
+    def run(self) -> None:
+        hard_kill_on_rank_crash(self._world_size, deadline=self.deadline, cancelled=self._cancelled)
+
+
+def start_rank_crash_kill_watchdog(world_size: int) -> Optional[RankCrashKillWatchdog]:
     """Arm a daemon thread that hard-kills the world once the grace elapses.
 
     Must be armed BEFORE executor-loop cleanup: cleanup can block without
@@ -157,7 +218,12 @@ def start_rank_crash_kill_watchdog(world_size: int) -> Optional[threading.Thread
     exists to avoid. The thread reuses ``hard_kill_on_rank_crash``, so the
     kill fires at crash + grace whether cleanup finishes, blocks, or raises.
 
-    Never raises. Returns the armed thread, or ``None`` when the kill is
+    The returned watchdog is the ONLY timer while cleanup runs; the caller
+    is expected to ``cancel()`` it once cleanup returns and carry the kill
+    (with the same ``deadline``) itself, so the two paths never race with
+    two independent clocks.
+
+    Never raises. Returns the armed watchdog, or ``None`` when the kill is
     not applicable (single rank, disabled by env) or the thread could not
     be started — in that case the caller's post-cleanup kill remains the
     only mechanism.
@@ -165,14 +231,10 @@ def start_rank_crash_kill_watchdog(world_size: int) -> Optional[threading.Thread
     try:
         if world_size <= 1:
             return None
-        if _rank_crash_kill_grace() is None:
+        grace = _rank_crash_kill_grace()
+        if grace is None:
             return None
-        watchdog = threading.Thread(
-            target=hard_kill_on_rank_crash,
-            args=(world_size,),
-            name="rank_crash_kill_watchdog",
-            daemon=True,
-        )
+        watchdog = RankCrashKillWatchdog(world_size, grace)
         watchdog.start()
         return watchdog
     except Exception as e:  # noqa: BLE001 - must not mask the loop's original error
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index a9e3b70d6e4d..c1ac0f7c7314 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -1194,9 +1194,23 @@ def _event_loop_wrapper(self):
                 "TLLM_LINE_PROFILER_PATH")) and not self.is_warmup
             with host_profiler_context(enable=enable_profiler), \
                  customized_gc_thresholds(self.garbage_collection_gen0_threshold):
-                self.event_loop()
+                try:
+                    self.event_loop()
+                except Exception:
+                    # Only a loop that dies BEFORE it processed its shutdown
+                    # request strands its peers. Once is_shutdown is set,
+                    # every rank has already seen the shutdown broadcast and
+                    # no peer is waiting on this one, so a raise on the way
+                    # out of an already-shutting-down loop (e.g. from the
+                    # profiler's or hang detector's __exit__) is a teardown
+                    # error to log, never a reason to SIGKILL the job.
+                    #
+                    # The flag is scoped to event_loop() itself for the same
+                    # reason: teardown of the enclosing host-profiler / GC
+                    # context managers is not a stranded-peer condition.
+                    crashed = not self.is_shutdown
+                    raise
         except Exception as e:
-            crashed = True
             logger.error(f"Error in event loop: {e}")
             logger.error(traceback.format_exc())
             # Stash the original error so local consumers
@@ -1209,11 +1223,11 @@ def _event_loop_wrapper(self):
             self._event_loop_error = e
             raise e
         finally:
-            if crashed:
-                # Armed BEFORE cleanup: cleanup can block without bound on a
-                # send handle wedged by the crash, and a kill placed only
-                # after it would never fire.
-                start_rank_crash_kill_watchdog(self.dist.world_size)
+            # Armed BEFORE cleanup: cleanup can block without bound on a
+            # send handle wedged by the crash, and a kill placed only after
+            # it would never fire.
+            watchdog = start_rank_crash_kill_watchdog(
+                self.dist.world_size) if crashed else None
             try:
                 self._executor_loop_cleanup()
             finally:
@@ -1225,9 +1239,19 @@ def _event_loop_wrapper(self):
                     # rank-local waiters and the ready handshake first, so
                     # the client sees the original exception rather than a
                     # bare worker death. Nested finally: the kill must fire
-                    # even when cleanup itself raises; the watchdog above
-                    # covers cleanup blocking.
-                    hard_kill_on_rank_crash(self.dist.world_size)
+                    # even when cleanup itself raises.
+                    #
+                    # Cleanup returned, so the watchdog's only job (covering
+                    # a cleanup that never returns) is done: disarm it and
+                    # carry the kill here, on its ORIGINAL deadline. Exactly
+                    # one timer is live at any moment, and the handover
+                    # cannot push the kill out by a second grace.
+                    deadline = None
+                    if watchdog is not None:
+                        watchdog.cancel()
+                        deadline = watchdog.deadline
+                    hard_kill_on_rank_crash(self.dist.world_size,
+                                            deadline=deadline)
 
     @property
     def is_warmup(self) -> bool:
diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index b497440ffd02..4ee53488c3e5 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -229,13 +229,52 @@ def test_watchdog_not_armed_when_disabled(monkeypatch):
     assert kills == []
 
 
+def test_watchdog_cancel_prevents_the_kill(monkeypatch):
+    """A cancelled watchdog must not SIGKILL a rank that goes on to exit cleanly.
+
+    Without a cancel path an armed watchdog fires at crash + grace no matter
+    what happens afterwards, turning a would-be exit 0 into exit 137.
+    """
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
+
+    watchdog = start_rank_crash_kill_watchdog(world_size=2)
+    assert watchdog is not None
+    watchdog.cancel()
+    # Cancel must break the grace wait immediately, not merely be observed
+    # after it elapses -- otherwise the process still dies 30s later.
+    watchdog.join(timeout=10.0)
+    assert not watchdog.is_alive()
+    assert kills == []
+    assert watchdog.cancelled is True
+
+
+def test_kill_keeps_original_deadline_on_handover(monkeypatch):
+    """Handing the kill over must not restart the grace clock.
+
+    The caller cancels the watchdog once cleanup returns and carries the kill
+    itself; passing the watchdog's deadline keeps the kill at crash + grace
+    instead of crash + 2*grace.
+    """
+    slept = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: slept.append("kill"))
+    monkeypatch.setattr(hd_module.time, "sleep", lambda s: slept.append(round(s, 1)))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "10")
+
+    # 6s of the 10s grace has already been burned by the watchdog.
+    deadline = hd_module.time.monotonic() + 4.0
+    assert hard_kill_on_rank_crash(world_size=2, deadline=deadline) is True
+    assert slept == [4.0, "kill"]
+
+
 # --------------------------------------------------------------------------
 # Wiring: PyExecutor._event_loop_wrapper must invoke the kill on the crash
 # path only, and only after local cleanup has woken rank-local waiters.
 # --------------------------------------------------------------------------
 
 
-def _bare_executor(pe, monkeypatch, world_size):
+def _bare_executor(pe, monkeypatch, world_size, is_shutdown=False):
     # Neutralize the profiling/GC context managers: they are irrelevant to the
     # crash path and must not depend on env/GC state in a unit test.
     monkeypatch.setattr(pe, "host_profiler_context", lambda enable: contextlib.nullcontext())
@@ -243,26 +282,51 @@ def _bare_executor(pe, monkeypatch, world_size):
     ex = pe.PyExecutor.__new__(pe.PyExecutor)
     ex.dist = types.SimpleNamespace(world_size=world_size)
     ex.garbage_collection_gen0_threshold = None
+    ex.is_shutdown = is_shutdown
     return ex
 
 
-def _stub_kill_paths(pe, monkeypatch, events):
-    monkeypatch.setattr(
-        pe, "hard_kill_on_rank_crash", lambda world_size: events.append(("kill", world_size))
-    )
+class _FakeWatchdog:
+    """Stand-in for RankCrashKillWatchdog that records cancellation."""
+
+    def __init__(self, events, world_size):
+        self._events = events
+        self.deadline = 1234.5
+        self.cancelled = False
+        events.append(("watchdog", world_size))
+
+    def cancel(self):
+        self.cancelled = True
+        self._events.append("cancel")
+
+
+def _stub_kill_paths(pe, monkeypatch, events, arm_watchdog=True):
     monkeypatch.setattr(
         pe,
-        "start_rank_crash_kill_watchdog",
-        lambda world_size: events.append(("watchdog", world_size)),
+        "hard_kill_on_rank_crash",
+        lambda world_size, deadline=None: events.append(("kill", world_size, deadline)),
     )
+    watchdogs = []
+
+    def _start(world_size):
+        if not arm_watchdog:
+            events.append(("watchdog", world_size))
+            return None
+        wd = _FakeWatchdog(events, world_size)
+        watchdogs.append(wd)
+        return wd
+
+    monkeypatch.setattr(pe, "start_rank_crash_kill_watchdog", _start)
+    return watchdogs
 
 
 def test_event_loop_wrapper_kills_world_on_crash(monkeypatch):
+    """A genuine mid-loop crash (is_shutdown still False) must kill the world."""
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
     events = []
-    _stub_kill_paths(pe, monkeypatch, events)
-    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    watchdogs = _stub_kill_paths(pe, monkeypatch, events)
+    ex = _bare_executor(pe, monkeypatch, world_size=4, is_shutdown=False)
     ex._executor_loop_cleanup = lambda: events.append("cleanup")
 
     def crash():
@@ -275,8 +339,11 @@ def crash():
 
     # The watchdog is armed BEFORE cleanup (cleanup can block forever);
     # cleanup wakes rank-local waiters (who read the stashed error) BEFORE
-    # the direct kill tears the world down.
-    assert events == [("watchdog", 4), "cleanup", ("kill", 4)]
+    # the direct kill tears the world down. Once cleanup returns, the
+    # watchdog is disarmed and the kill is carried inline on the watchdog's
+    # ORIGINAL deadline, so only one timer is ever live.
+    assert events == [("watchdog", 4), "cleanup", "cancel", ("kill", 4, 1234.5)]
+    assert watchdogs[0].cancelled is True
     assert isinstance(ex._event_loop_error, ValueError)
 
 
@@ -292,7 +359,7 @@ def test_event_loop_wrapper_kills_world_when_cleanup_raises(monkeypatch):
 
     events = []
     _stub_kill_paths(pe, monkeypatch, events)
-    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex = _bare_executor(pe, monkeypatch, world_size=4, is_shutdown=False)
 
     def broken_cleanup():
         events.append("cleanup")
@@ -308,11 +375,31 @@ def crash():
     with pytest.raises(RuntimeError, match="cleanup exploded"):
         ex._event_loop_wrapper()
 
-    assert events == [("watchdog", 4), "cleanup", ("kill", 4)]
+    assert events == [("watchdog", 4), "cleanup", "cancel", ("kill", 4, 1234.5)]
     # The original loop error stays reachable for rank-local consumers.
     assert isinstance(ex._event_loop_error, ValueError)
 
 
+def test_event_loop_wrapper_kills_world_when_watchdog_cannot_arm(monkeypatch):
+    """A watchdog that fails to start must not silently drop the escalation."""
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events, arm_watchdog=False)
+    ex = _bare_executor(pe, monkeypatch, world_size=4, is_shutdown=False)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+
+    def crash():
+        raise ValueError("boom")
+
+    ex.event_loop = crash
+
+    with pytest.raises(ValueError, match="boom"):
+        ex._event_loop_wrapper()
+
+    assert events == [("watchdog", 4), "cleanup", ("kill", 4, None)]
+
+
 def test_event_loop_wrapper_no_kill_on_clean_exit(monkeypatch):
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
@@ -325,3 +412,62 @@ def test_event_loop_wrapper_no_kill_on_clean_exit(monkeypatch):
     ex._event_loop_wrapper()
 
     assert events == ["cleanup"]
+
+
+# --------------------------------------------------------------------------
+# The kill must stay scoped to crashes that actually strand peers. A raise
+# on the way out of an already-shut-down loop happens after every rank has
+# processed the shutdown broadcast and all work is done: escalating it turns
+# a benign teardown error into a whole-job SIGKILL (exit 137 instead of 0).
+# --------------------------------------------------------------------------
+
+
+def test_event_loop_wrapper_no_kill_when_loop_raises_after_shutdown(monkeypatch):
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+
+    def late_raise():
+        # The loop processed its shutdown request and drained all work, then
+        # something raised on the way out (e.g. a context manager's __exit__).
+        ex.is_shutdown = True
+        raise RuntimeError("teardown hiccup")
+
+    ex.event_loop = late_raise
+
+    with pytest.raises(RuntimeError, match="teardown hiccup"):
+        ex._event_loop_wrapper()
+
+    # Logged and re-raised, but no watchdog and no kill: peers are not stranded.
+    assert events == ["cleanup"]
+    assert isinstance(ex._event_loop_error, RuntimeError)
+
+
+def test_event_loop_wrapper_no_kill_when_enclosing_context_manager_raises(monkeypatch):
+    """Teardown of the host-profiler / GC context managers is not a crash.
+
+    They wrap event_loop() but are not part of it; a failure while unwinding
+    them leaves no peer waiting on this rank.
+    """
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+
+    @contextlib.contextmanager
+    def exploding_ctx(**_kwargs):
+        yield
+        raise RuntimeError("profiler teardown failed")
+
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    monkeypatch.setattr(pe, "host_profiler_context", lambda enable: exploding_ctx())
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+    ex.event_loop = lambda: None
+
+    with pytest.raises(RuntimeError, match="profiler teardown failed"):
+        ex._event_loop_wrapper()
+
+    assert events == ["cleanup"]

From 6aa332ab7706f37cebd6fc22fef43394fb4daee6 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 16:06:05 +0000
Subject: [PATCH 05/10] [TRTLLM-13409][test] never leave an armed kill watchdog
 past the stubbed kill

If cancel() ever regresses, the watchdog thread outlives monkeypatch
teardown and SIGKILLs the pytest process instead of failing the test.
Join it inside the test, while propagate_hard_kill is still stubbed.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../executor/test_hang_detector_kill.py       | 21 ++++++++++++-------
 1 file changed, 14 insertions(+), 7 deletions(-)

diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index 4ee53488c3e5..19d925751caf 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -241,13 +241,20 @@ def test_watchdog_cancel_prevents_the_kill(monkeypatch):
 
     watchdog = start_rank_crash_kill_watchdog(world_size=2)
     assert watchdog is not None
-    watchdog.cancel()
-    # Cancel must break the grace wait immediately, not merely be observed
-    # after it elapses -- otherwise the process still dies 30s later.
-    watchdog.join(timeout=10.0)
-    assert not watchdog.is_alive()
-    assert kills == []
-    assert watchdog.cancelled is True
+    try:
+        watchdog.cancel()
+        # Cancel must break the grace wait immediately, not merely be observed
+        # after it elapses -- otherwise the process still dies 30s later.
+        watchdog.join(timeout=10.0)
+        assert not watchdog.is_alive()
+        assert kills == []
+        assert watchdog.cancelled is True
+    finally:
+        # Never let an armed killer thread outlive the stubbed
+        # propagate_hard_kill: if cancel() ever regresses, the real one would
+        # SIGKILL the pytest process once monkeypatch restores it.
+        watchdog.cancel()
+        watchdog.join(timeout=60.0)
 
 
 def test_kill_keeps_original_deadline_on_handover(monkeypatch):

From f9d49434f6bfae0372d3976502ef0acfc2fd4814 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 16:51:51 +0000
Subject: [PATCH 06/10] [TRTLLM-13409][fix] gate the rank-crash kill on loop
 completion, not is_shutdown

The previous commit keyed the decision off `not self.is_shutdown`, on the
premise that is_shutdown means "every rank has processed the shutdown
broadcast". That premise is false, and the mistake silently disabled the
kill for this feature's most common trigger.

`_handle_errors` sets `self.is_shutdown = True` rank-locally whenever an
error is fatal, and `classify_error` returns `immediate_fatal` for a CUDA
illegal address / device-side assert / launch failure, bypassing the error
budget entirely. The only "broadcast" that follows is
`enqueue_shutdown_request()`, which pushes into this process's own queue --
peers are told nothing. `should_stop_processing` additionally requires empty
active/waiting queues, so the loop keeps entering collectives after the flag
is set.

So: rank 3 of 4 takes cudaErrorIllegalAddress in the forward pass,
_handle_errors sets is_shutdown and returns None, and the next unguarded
statement (guided_decoder.execute(batch_outputs['logits'])) raises TypeError
on None. The wrapper computed crashed=False, armed nothing, and ranks 0-2
blocked in their next NCCL collective for the full 300s -- exactly the
failure this PR exists to remove.

Gate on an explicit completion sentinel instead. `_event_loop_completed` is
set at the three executor loops' normal-exit `break` sites and nowhere else,
so it answers the actual question: did event_loop() reach its own
termination? A raise after that point (profiler/hang-detector __exit__, or
the enclosing context managers) is a teardown error; anything else -- now
including a failure before the loop ever started, which the previous inner
try narrowed away -- strands peers and escalates.

Also corrects two overstated claims from the previous commit. The
watchdog/post-cleanup "double-arm" never produced an extra or later kill:
earliest-wins gave crash+grace before, and max(cleanup_end, crash+grace) =
crash+grace after. cancel() cannot spare a rank that exits cleanly either --
it is followed one line later by the same kill on the same deadline. The
handover machinery is kept because one timer is clearer than two, but all
the protection against a spurious SIGKILL rests on the predicate above.

Log fixes: the disarm message drops from ERROR to DEBUG (it printed
"cancelled before the grace elapsed" moments before the world was SIGKILLed,
which reads during triage as "the kill was called off"), and the countdown
now logs the remaining time rather than the full grace on the handover path.

Tests: a rank-local-fatal regression test that fails against the previous
predicate; a pre-loop-failure test; coverage for the already-elapsed
deadline (the max(0.0, ...) clamp is the only thing keeping a negative sleep
from raising into the blanket except and dropping the kill); and an AST
check that every `break` in the three loops is preceded by the sentinel, so
a new normal-exit path cannot make clean shutdowns look like crashes.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/pyexecutor/hang_detector.py        |  70 ++++--
 tensorrt_llm/_torch/pyexecutor/py_executor.py |  53 ++--
 .../executor/test_hang_detector_kill.py       | 228 +++++++++++++++---
 3 files changed, 281 insertions(+), 70 deletions(-)

diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
index b7ef25a5aa95..1254b21361ec 100644
--- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py
+++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
@@ -50,6 +50,14 @@ def _best_effort_log_error(message: str) -> None:
         pass
 
 
+def _best_effort_log_debug(message: str) -> None:
+    """Log at debug level without ever raising; diagnostics must not block hard kill."""
+    try:
+        logger.debug(message)
+    except Exception:  # noqa: BLE001 - diagnostics must not block hard kill
+        pass
+
+
 def propagate_hard_kill(exit_code: int = _HARD_KILL_EXIT_CODE) -> None:
     """Hard-kill this rank and propagate the kill to peer ranks.
 
@@ -102,20 +110,28 @@ def _rank_crash_kill_grace() -> Optional[float]:
     return None if grace < 0 else grace
 
 
-def _wait_out_kill_grace(
-    grace: float,
-    deadline: Optional[float],
-    cancelled: Optional[threading.Event],
-) -> bool:
-    """Sleep out the crash-kill grace; return False if the kill was cancelled.
+def _remaining_kill_grace(grace: float, deadline: Optional[float]) -> float:
+    """Time left before the kill must fire.
 
     ``deadline`` (a ``time.monotonic()`` stamp) lets a kill that was already
     armed elsewhere keep its ORIGINAL fire time when it is handed over to
-    another waiter, so a handover cannot push the kill out by a second grace.
-    ``cancelled`` makes the wait interruptible: a rank that turns out not to
-    need the kill can disarm it instead of being killed while exiting cleanly.
+    another waiter, so the handover cannot push the kill out by a second
+    grace. Clamped at 0: a deadline already in the past means fire now, and
+    a negative sleep would raise into the caller's blanket except and drop
+    the kill entirely -- exactly in the case (cleanup outlasted the grace)
+    the watchdog exists for.
+    """
+    if deadline is None:
+        return grace
+    return max(0.0, deadline - time.monotonic())
+
+
+def _wait_out_kill_grace(remaining: float, cancelled: Optional[threading.Event]) -> bool:
+    """Sleep out the crash-kill grace; return False if the kill was cancelled.
+
+    ``cancelled`` makes the wait interruptible so the timer can be handed
+    over to another waiter instead of two clocks running at once.
     """
-    remaining = grace if deadline is None else max(0.0, deadline - time.monotonic())
     if cancelled is None:
         if remaining > 0:
             time.sleep(remaining)
@@ -160,13 +176,22 @@ def hard_kill_on_rank_crash(
         grace = _rank_crash_kill_grace()
         if grace is None:
             return False
+        remaining = _remaining_kill_grace(grace, deadline)
         _best_effort_log_error(
             f"Executor loop crashed on this rank; hard-killing all "
-            f"{world_size} ranks in {grace}s (peers cannot make progress "
+            f"{world_size} ranks in {remaining:g}s (peers cannot make progress "
             f"without this rank). Set {RANK_CRASH_KILL_GRACE_ENV}=-1 to disable."
         )
-        if not _wait_out_kill_grace(grace, deadline, cancelled):
-            _best_effort_log_error("Rank-crash hard kill cancelled before the grace elapsed.")
+        if not _wait_out_kill_grace(remaining, cancelled):
+            # Debug, not error: the only caller that cancels does so to take
+            # the same kill over on the same deadline. Logging "cancelled" at
+            # ERROR right before the world is SIGKILLed reads during triage
+            # as "the kill was called off", which is the opposite of what
+            # happens.
+            _best_effort_log_debug(
+                "Rank-crash hard kill timer disarmed (handed over or no "
+                "longer needed); this timer will not fire."
+            )
             return False
         propagate_hard_kill()
         return True
@@ -179,15 +204,16 @@ class RankCrashKillWatchdog(threading.Thread):
     """Daemon thread that hard-kills the world once the crash grace elapses.
 
     A plain ``Thread`` for backwards compatibility (callers may still join it
-    or inspect ``daemon``), plus the two things the caller needs to own the
-    kill deadline:
-
-    - ``cancel()`` disarms the timer. Without it an armed watchdog fires at
-      crash + grace unconditionally, so a rank that ends up exiting cleanly
-      is SIGKILLed anyway and a would-be exit 0 becomes exit 137.
-    - ``deadline`` exposes the original fire time so whoever takes the kill
-      over after cancelling still fires at crash + grace rather than
-      restarting the clock.
+    or inspect ``daemon``), plus the two things needed to hand the timer over
+    to another waiter instead of running two clocks:
+
+    - ``cancel()`` disarms THIS timer. It is a bookkeeping aid, not a safety
+      net: the only caller cancels in order to take the same kill over on the
+      same deadline one line later, so cancelling does not spare a rank. What
+      decides whether a rank is killed at all is the ``crashed`` predicate in
+      ``PyExecutor._event_loop_wrapper``.
+    - ``deadline`` exposes the original fire time so the caller that takes
+      over still fires at crash + grace rather than restarting the clock.
     """
 
     def __init__(self, world_size: int, grace: float):
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index c1ac0f7c7314..704252a54914 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -823,6 +823,15 @@ def __init__(
             self.kv_cache_manager.snapshot_warmup_baseline()
 
         self.is_shutdown = False
+        # Set at the executor loops' normal-exit `break` sites, and ONLY
+        # there. It answers exactly one question for _event_loop_wrapper:
+        # "did event_loop() reach its own termination?" -- which decides
+        # whether an escaping exception stranded this rank's peers.
+        # is_shutdown cannot answer it: _handle_errors sets is_shutdown
+        # rank-locally on a fatal error (e.g. a CUDA illegal address on this
+        # rank alone) while peers are told nothing and the loop keeps running
+        # collectives, so a crash after that point still strands them.
+        self._event_loop_completed = False
         self._fatal_error: Optional[BaseException] = None
         self._error_budget = ErrorBudget()
         self._disagg_timed_out_ctx_cancelled_ids: set[int] = set()
@@ -1187,6 +1196,7 @@ def _flush_iter_stats_synced(self):
 
     def _event_loop_wrapper(self):
         crashed = False
+        self._event_loop_completed = False
         try:
             # Skip line profiler during warmup/memory estimation phase to avoid
             # saving incomplete results that would be overwritten anyway
@@ -1194,23 +1204,22 @@ def _event_loop_wrapper(self):
                 "TLLM_LINE_PROFILER_PATH")) and not self.is_warmup
             with host_profiler_context(enable=enable_profiler), \
                  customized_gc_thresholds(self.garbage_collection_gen0_threshold):
-                try:
-                    self.event_loop()
-                except Exception:
-                    # Only a loop that dies BEFORE it processed its shutdown
-                    # request strands its peers. Once is_shutdown is set,
-                    # every rank has already seen the shutdown broadcast and
-                    # no peer is waiting on this one, so a raise on the way
-                    # out of an already-shutting-down loop (e.g. from the
-                    # profiler's or hang detector's __exit__) is a teardown
-                    # error to log, never a reason to SIGKILL the job.
-                    #
-                    # The flag is scoped to event_loop() itself for the same
-                    # reason: teardown of the enclosing host-profiler / GC
-                    # context managers is not a stranded-peer condition.
-                    crashed = not self.is_shutdown
-                    raise
+                self.event_loop()
         except Exception as e:
+            # A raise AFTER the loop reached its own normal-exit `break` (from
+            # the profiler's or hang detector's __exit__, or from the enclosing
+            # context managers) is a teardown error: this rank finished its
+            # work and no peer is waiting on it, so log it but never escalate
+            # to SIGKILLing the job. Anything else -- including a raise before
+            # the loop ever started -- leaves peers blocked in their next
+            # collective, which is what the kill exists to cut short.
+            #
+            # Deliberately NOT is_shutdown: _handle_errors flips that flag
+            # rank-locally on a fatal error (a CUDA illegal address on one rank
+            # is classified immediate_fatal and bypasses the error budget) and
+            # tells peers nothing, so a crash after that point -- the single
+            # most common trigger for this kill -- still strands them.
+            crashed = not self._event_loop_completed
             logger.error(f"Error in event loop: {e}")
             logger.error(traceback.format_exc())
             # Stash the original error so local consumers
@@ -1242,10 +1251,11 @@ def _event_loop_wrapper(self):
                     # even when cleanup itself raises.
                     #
                     # Cleanup returned, so the watchdog's only job (covering
-                    # a cleanup that never returns) is done: disarm it and
-                    # carry the kill here, on its ORIGINAL deadline. Exactly
-                    # one timer is live at any moment, and the handover
-                    # cannot push the kill out by a second grace.
+                    # a cleanup that never returns) is done: hand the timer
+                    # over rather than leave two running. This is bookkeeping,
+                    # not protection -- the kill still fires, on the SAME
+                    # deadline, one line below. Whether a rank is killed at
+                    # all is decided solely by `crashed` above.
                     deadline = None
                     if watchdog is not None:
                         watchdog.cancel()
@@ -2591,6 +2601,7 @@ def _executor_loop_pp(self):
                 # Fetch new requests from request queue
                 new_requests = self._fetch_and_activate_new_requests()
                 if self.should_stop_processing:
+                    self._event_loop_completed = True
                     break
 
                 self._handle_control_request()
@@ -4057,6 +4068,7 @@ def _executor_loop(self):
                 scheduled_batch, iter_stats = self._prepare_and_schedule_batch()
 
                 if scheduled_batch is None:
+                    self._event_loop_completed = True
                     break
 
                 can_forward, should_retry = self._check_benchmark_disagg_gate(
@@ -4533,6 +4545,7 @@ def _executor_loop_overlap(self):
                 scheduled_batch, iter_stats = self._prepare_and_schedule_batch()
 
                 if scheduled_batch is None:
+                    self._event_loop_completed = True
                     break
 
                 can_forward, should_retry = self._check_benchmark_disagg_gate(
diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index 19d925751caf..0b6c2cd9671c 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -229,11 +229,13 @@ def test_watchdog_not_armed_when_disabled(monkeypatch):
     assert kills == []
 
 
-def test_watchdog_cancel_prevents_the_kill(monkeypatch):
-    """A cancelled watchdog must not SIGKILL a rank that goes on to exit cleanly.
+def test_watchdog_cancel_disarms_this_timer(monkeypatch):
+    """cancel() must break the grace wait immediately, not after it elapses.
 
-    Without a cancel path an armed watchdog fires at crash + grace no matter
-    what happens afterwards, turning a would-be exit 0 into exit 137.
+    This is the handover primitive, NOT protection against a spurious kill:
+    the only production caller cancels in order to take the same kill over on
+    the same deadline. What decides whether a rank is killed at all is the
+    `crashed` predicate in _event_loop_wrapper.
     """
     kills = []
     monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
@@ -243,8 +245,6 @@ def test_watchdog_cancel_prevents_the_kill(monkeypatch):
     assert watchdog is not None
     try:
         watchdog.cancel()
-        # Cancel must break the grace wait immediately, not merely be observed
-        # after it elapses -- otherwise the process still dies 30s later.
         watchdog.join(timeout=10.0)
         assert not watchdog.is_alive()
         assert kills == []
@@ -257,22 +257,49 @@ def test_watchdog_cancel_prevents_the_kill(monkeypatch):
         watchdog.join(timeout=60.0)
 
 
+def test_watchdog_deadline_is_grace_from_arming(monkeypatch):
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "5")
+    watchdog = hd_module.RankCrashKillWatchdog(world_size=2, grace=5.0)
+    assert watchdog.deadline == pytest.approx(time.monotonic() + 5.0, abs=0.5)
+
+
 def test_kill_keeps_original_deadline_on_handover(monkeypatch):
     """Handing the kill over must not restart the grace clock.
 
     The caller cancels the watchdog once cleanup returns and carries the kill
     itself; passing the watchdog's deadline keeps the kill at crash + grace
-    instead of crash + 2*grace.
+    instead of crash + 2*grace. Uses a real (short) sleep rather than patching
+    time.sleep process-wide, so no background thread can busy-spin into the
+    assertion.
+    """
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
+
+    # Nearly all of the 30s grace has already been burned by the watchdog.
+    t0 = time.monotonic()
+    assert hard_kill_on_rank_crash(world_size=2, deadline=t0 + 0.3) is True
+    elapsed = time.monotonic() - t0
+    assert kills == [1]
+    # Slept out the REMAINING 0.3s, not a fresh 30s grace.
+    assert elapsed == pytest.approx(0.3, abs=0.25)
+
+
+def test_kill_fires_immediately_when_deadline_already_passed(monkeypatch):
+    """A deadline in the past must fire now, not raise into the blanket except.
+
+    Without the max(0.0, ...) clamp this sleeps a negative duration, raises,
+    and hard_kill_on_rank_crash returns False -- silently skipping the kill in
+    exactly the case the watchdog exists for (cleanup outlasted the grace).
     """
-    slept = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: slept.append("kill"))
-    monkeypatch.setattr(hd_module.time, "sleep", lambda s: slept.append(round(s, 1)))
-    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "10")
+    kills = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
-    # 6s of the 10s grace has already been burned by the watchdog.
-    deadline = hd_module.time.monotonic() + 4.0
-    assert hard_kill_on_rank_crash(world_size=2, deadline=deadline) is True
-    assert slept == [4.0, "kill"]
+    t0 = time.monotonic()
+    assert hard_kill_on_rank_crash(world_size=2, deadline=t0 - 100.0) is True
+    assert kills == [1]
+    assert time.monotonic() - t0 < 1.0
 
 
 # --------------------------------------------------------------------------
@@ -289,21 +316,33 @@ def _bare_executor(pe, monkeypatch, world_size, is_shutdown=False):
     ex = pe.PyExecutor.__new__(pe.PyExecutor)
     ex.dist = types.SimpleNamespace(world_size=world_size)
     ex.garbage_collection_gen0_threshold = None
+    # is_shutdown must NOT influence the crash decision -- _handle_errors sets
+    # it rank-locally on a fatal error while peers are told nothing.
     ex.is_shutdown = is_shutdown
+    ex._event_loop_completed = False
     return ex
 
 
 class _FakeWatchdog:
-    """Stand-in for RankCrashKillWatchdog that records cancellation."""
+    """Stand-in for RankCrashKillWatchdog that records cancellation.
+
+    ``cancelled`` is a read-only property, matching the real class: a wiring
+    change that assigned to it would pass against a plain attribute here and
+    raise AttributeError in production.
+    """
 
     def __init__(self, events, world_size):
         self._events = events
+        self._cancelled = False
         self.deadline = 1234.5
-        self.cancelled = False
         events.append(("watchdog", world_size))
 
+    @property
+    def cancelled(self):
+        return self._cancelled
+
     def cancel(self):
-        self.cancelled = True
+        self._cancelled = True
         self._events.append("cancel")
 
 
@@ -422,14 +461,15 @@ def test_event_loop_wrapper_no_kill_on_clean_exit(monkeypatch):
 
 
 # --------------------------------------------------------------------------
-# The kill must stay scoped to crashes that actually strand peers. A raise
-# on the way out of an already-shut-down loop happens after every rank has
-# processed the shutdown broadcast and all work is done: escalating it turns
-# a benign teardown error into a whole-job SIGKILL (exit 137 instead of 0).
+# The kill must stay scoped to crashes that actually strand peers, and the
+# only signal that says so is _event_loop_completed -- set at the loops'
+# normal-exit `break` sites and nowhere else. is_shutdown does NOT mean
+# "peers were told": _handle_errors flips it rank-locally on a fatal error.
 # --------------------------------------------------------------------------
 
 
-def test_event_loop_wrapper_no_kill_when_loop_raises_after_shutdown(monkeypatch):
+def test_event_loop_wrapper_no_kill_when_loop_raises_after_completing(monkeypatch):
+    """A raise after the loop's normal-exit break is a teardown error, not a crash."""
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
     events = []
@@ -438,9 +478,9 @@ def test_event_loop_wrapper_no_kill_when_loop_raises_after_shutdown(monkeypatch)
     ex._executor_loop_cleanup = lambda: events.append("cleanup")
 
     def late_raise():
-        # The loop processed its shutdown request and drained all work, then
+        # The loop hit its normal-exit `break` and drained all work, then
         # something raised on the way out (e.g. a context manager's __exit__).
-        ex.is_shutdown = True
+        ex._event_loop_completed = True
         raise RuntimeError("teardown hiccup")
 
     ex.event_loop = late_raise
@@ -453,11 +493,65 @@ def late_raise():
     assert isinstance(ex._event_loop_error, RuntimeError)
 
 
+def test_event_loop_wrapper_kills_world_on_rank_local_fatal(monkeypatch):
+    """REGRESSION: a rank-local CUDA fatal sets is_shutdown but strands peers.
+
+    _handle_errors classifies a device-side fault as immediate_fatal, sets
+    is_shutdown=True on THIS rank and enqueues a shutdown into THIS process's
+    own queue -- peers are told nothing and keep waiting in their collective.
+    An exception raised after that point (e.g. the unguarded
+    guided_decoder.execute(batch_outputs['logits']) on a None batch_outputs)
+    must still hard-kill the world. Keying the decision off is_shutdown
+    silently disabled the kill for this, the feature's most common trigger.
+    """
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+
+    def cuda_fatal_then_crash():
+        ex.is_shutdown = True  # what _handle_errors does, rank-locally
+        assert ex._event_loop_completed is False  # the loop never terminated
+        raise TypeError("'NoneType' object is not subscriptable")
+
+    ex.event_loop = cuda_fatal_then_crash
+
+    with pytest.raises(TypeError):
+        ex._event_loop_wrapper()
+
+    assert events == [("watchdog", 4), "cleanup", "cancel", ("kill", 4, 1234.5)]
+
+
+def test_event_loop_wrapper_kills_world_when_loop_never_started(monkeypatch):
+    """A failure before the loop runs strands peers just as surely as one inside it."""
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+
+    @contextlib.contextmanager
+    def failing_enter(**_kwargs):
+        raise RuntimeError("profiler setup failed")
+        yield  # pragma: no cover
+
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    monkeypatch.setattr(pe, "host_profiler_context", lambda enable: failing_enter())
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+    ex.event_loop = lambda: pytest.fail("event_loop must not be reached")
+
+    with pytest.raises(RuntimeError, match="profiler setup failed"):
+        ex._event_loop_wrapper()
+
+    assert events == [("watchdog", 4), "cleanup", "cancel", ("kill", 4, 1234.5)]
+
+
 def test_event_loop_wrapper_no_kill_when_enclosing_context_manager_raises(monkeypatch):
-    """Teardown of the host-profiler / GC context managers is not a crash.
+    """Teardown of the host-profiler / GC context managers after a completed loop.
 
     They wrap event_loop() but are not part of it; a failure while unwinding
-    them leaves no peer waiting on this rank.
+    them once the loop has completed leaves no peer waiting on this rank.
     """
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
@@ -472,9 +566,87 @@ def exploding_ctx(**_kwargs):
     ex = _bare_executor(pe, monkeypatch, world_size=4)
     monkeypatch.setattr(pe, "host_profiler_context", lambda enable: exploding_ctx())
     ex._executor_loop_cleanup = lambda: events.append("cleanup")
-    ex.event_loop = lambda: None
+
+    def completed_loop():
+        ex._event_loop_completed = True
+
+    ex.event_loop = completed_loop
 
     with pytest.raises(RuntimeError, match="profiler teardown failed"):
         ex._event_loop_wrapper()
 
     assert events == ["cleanup"]
+
+
+# --------------------------------------------------------------------------
+# The sentinel is only trustworthy if the real loops actually set it. Assert
+# against the shipped source so a new normal-exit path (or a moved break)
+# cannot silently make every clean shutdown look like a peer-stranding crash.
+# --------------------------------------------------------------------------
+
+
+def _executor_loop_ast_nodes():
+    import ast
+    import inspect
+
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    tree = ast.parse(inspect.getsource(pe))
+    wanted = {"_executor_loop", "_executor_loop_pp", "_executor_loop_overlap"}
+    return {
+        node.name: node
+        for node in ast.walk(tree)
+        if isinstance(node, ast.FunctionDef) and node.name in wanted
+    }
+
+
+def test_every_executor_loop_break_sets_the_completion_sentinel():
+    import ast
+
+    loops = _executor_loop_ast_nodes()
+    assert set(loops) == {"_executor_loop", "_executor_loop_pp", "_executor_loop_overlap"}, (
+        f"executor loops renamed or removed: {sorted(loops)}"
+    )
+
+    for name, fn in loops.items():
+        # Only breaks belonging to THIS function (not to a nested def) end
+        # the event loop.
+        nested = {
+            id(n)
+            for d in ast.walk(fn)
+            if isinstance(d, (ast.FunctionDef, ast.AsyncFunctionDef)) and d is not fn
+            for n in ast.walk(d)
+        }
+        own_breaks = [n for n in ast.walk(fn) if isinstance(n, ast.Break) and id(n) not in nested]
+        assert own_breaks, f"{name}: no break found -- did the loop exit change?"
+
+        checked = 0
+        for parent in ast.walk(fn):
+            for field in ("body", "orelse", "finalbody"):
+                block = getattr(parent, field, None)
+                if not isinstance(block, list):
+                    continue
+                for i, stmt in enumerate(block):
+                    if not isinstance(stmt, ast.Break) or id(stmt) in nested:
+                        continue
+                    checked += 1
+                    prev = block[i - 1] if i else None
+                    sets_sentinel = isinstance(prev, ast.Assign) and any(
+                        isinstance(t, ast.Attribute) and t.attr == "_event_loop_completed"
+                        for t in prev.targets
+                    )
+                    assert sets_sentinel, (
+                        f"{name}: the `break` at line {stmt.lineno} is not preceded "
+                        "by `self._event_loop_completed = True`. Every normal exit "
+                        "must set it, or _event_loop_wrapper treats a clean "
+                        "shutdown as a peer-stranding crash and SIGKILLs the job."
+                    )
+        assert checked == len(own_breaks)
+
+
+def test_completion_sentinel_is_initialized_false():
+    import inspect
+
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    assert "self._event_loop_completed = False" in inspect.getsource(pe.PyExecutor.__init__)

From de9e90c227c074c54ab709d641d50f12cb86e84c Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 17:00:20 +0000
Subject: [PATCH 07/10] [TRTLLM-13409][test] make the grace-ordering assertions
 robust to the process-wide sleep patch

monkeypatching time.sleep is process-wide, so any background thread that
sleeps during these two tests appends its own entry into the asserted
list. Assert the ordering these tests are about instead of exact list
equality.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/executor/test_hang_detector_kill.py   | 16 ++++++++++++++--
 1 file changed, 14 insertions(+), 2 deletions(-)

diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index 0b6c2cd9671c..f3ba84dd5a3f 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -151,6 +151,18 @@ def test_rank_crash_kill_fires_for_multi_rank(monkeypatch):
     assert kills == [1]
 
 
+def _assert_slept_then_killed(order, grace):
+    """Assert the grace was slept out before the kill.
+
+    Patching time.sleep is process-wide, so an unrelated background thread can
+    append its own ("sleep", x) while this runs. Assert on the entries this
+    test is about rather than on exact list equality, which would flake.
+    """
+    assert ("sleep", grace) in order, order
+    assert "kill" in order, order
+    assert order.index(("sleep", grace)) < order.index("kill"), order
+
+
 def test_rank_crash_kill_sleeps_grace_before_kill(monkeypatch):
     """The grace must elapse BEFORE the kill so cleaner error paths win the race."""
     order = []
@@ -158,7 +170,7 @@ def test_rank_crash_kill_sleeps_grace_before_kill(monkeypatch):
     monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "2.5")
     assert hard_kill_on_rank_crash(world_size=2) is True
-    assert order == [("sleep", 2.5), "kill"]
+    _assert_slept_then_killed(order, 2.5)
 
 
 def test_rank_crash_kill_disabled_by_negative_grace(monkeypatch):
@@ -176,7 +188,7 @@ def test_rank_crash_kill_invalid_grace_uses_default(monkeypatch):
     monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "bogus")
     assert hard_kill_on_rank_crash(world_size=2) is True
-    assert order == [("sleep", 10.0), "kill"]
+    _assert_slept_then_killed(order, 10.0)
 
 
 def test_rank_crash_kill_never_raises(monkeypatch):

From b669ed22cd0e2656fe667f8e3e89475188e65852 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 17:53:08 +0000
Subject: [PATCH 08/10] [TRTLLM-13409][test] fix the sentinel guards; correct a
 false rationale on the deadline clamp

Five guard-quality defects from review. None changed runtime behavior on a
healthy run, but two recreated the conditions under which the predicate has
already regressed twice.

1. The AST guard enforced the wrong invariant. It required EVERY own `break`
   in the three loops to be preceded by the sentinel, but only a break
   exiting the outer `while True` terminates the event loop, and
   _executor_loop / _executor_loop_overlap already contain inner `for` loops.
   Worse, its failure message told the reader "every normal exit must set
   it", so a contributor who added an inner-loop break would have been
   instructed to set the sentinel while the loop was still running -- making
   every later rank-local crash look like a clean shutdown and silently
   disabling the kill, the same class as the previous blocker. Now: exactly
   one loop-terminating break per loop, located by recursing through
   if/try/with but stopping at nested for/while/def.

   Mutation-tested: inner `for _z in (): break` without the sentinel PASSES;
   a missing sentinel on the outer break FAILS; a second outer break FAILS.

2. The load-bearing reset was untested. Deleting
   `self._event_loop_completed = False` from _event_loop_wrapper left the
   whole suite green while a second loop run on the same executor misread a
   genuine crash as a clean shutdown (verified: run-2 events go from
   [watchdog, cleanup, cancel, kill] to [cleanup]). The one "initialization"
   test only grepped __init__ -- the site that is NOT load-bearing -- which
   invited deleting the real one as redundant. Added a behavioral
   two-invocation test plus a source assertion on the wrapper.

3. The rationale on the max(0.0, ...) deadline clamp was false. The comment
   and docstring claimed it was the only thing preventing a negative sleep
   from raising into the blanket except and dropping the kill. It is not:
   _wait_out_kill_grace guards with `remaining > 0`, and Event.wait() returns
   immediately for a negative timeout (probe: _wait_out_kill_grace(-100, None)
   -> True, zero sleep calls). A false "this guard protects X" comment is how
   the previous regressions got through, and someone trusting it could drop
   the real guard as redundant. Corrected both, moved the load-bearing note
   onto the `> 0` guard, and added a test asserting sleep is never called
   with a negative argument.

4. Hardening the two grace-ordering tests against the process-wide time.sleep
   patch weakened them: membership-only assertions accepted sleeping the
   grace TWICE before killing -- exactly the crash + 2*grace bug this series
   introduced with its two independent timers. Pinned the counts to 1 while
   still tolerating an unrelated thread's sleep. The handover test now pins
   time.monotonic and asserts the exact remaining duration instead of racing
   a 0.3s real sleep against a 0.55s bound on a loaded CI node.

5. The AST guard checked neither the assigned value nor the target object:
   `self._event_loop_completed = False` before the break passed, and so did
   `self.dist._event_loop_completed = True`. A value inversion turns every
   clean shutdown into a whole-job SIGKILL and was invisible. Now requires
   Constant True assigned to an attribute of Name('self'); both mutants fail.

Also documents why _executor_loop_pp sets the sentinel before its Stage-5
drain: reaching there means every rank observed should_stop_processing, so
the drain consumes only a rank-local queue and a raise in it strands nobody.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../_torch/pyexecutor/hang_detector.py        |  18 +-
 tensorrt_llm/_torch/pyexecutor/py_executor.py |   6 +
 .../executor/test_hang_detector_kill.py       | 230 +++++++++++++-----
 3 files changed, 186 insertions(+), 68 deletions(-)

diff --git a/tensorrt_llm/_torch/pyexecutor/hang_detector.py b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
index 1254b21361ec..2d9b940ac8a4 100644
--- a/tensorrt_llm/_torch/pyexecutor/hang_detector.py
+++ b/tensorrt_llm/_torch/pyexecutor/hang_detector.py
@@ -116,10 +116,14 @@ def _remaining_kill_grace(grace: float, deadline: Optional[float]) -> float:
     ``deadline`` (a ``time.monotonic()`` stamp) lets a kill that was already
     armed elsewhere keep its ORIGINAL fire time when it is handed over to
     another waiter, so the handover cannot push the kill out by a second
-    grace. Clamped at 0: a deadline already in the past means fire now, and
-    a negative sleep would raise into the caller's blanket except and drop
-    the kill entirely -- exactly in the case (cleanup outlasted the grace)
-    the watchdog exists for.
+    grace.
+
+    The ``max(0.0, ...)`` is belt-and-braces only: it keeps the return value
+    meaningful as "time left" for callers and logs. It is NOT what stops a
+    negative sleep -- ``_wait_out_kill_grace`` does that with its
+    ``remaining > 0`` guard (and ``Event.wait`` returns immediately for a
+    negative timeout anyway). Do not drop that guard on the strength of this
+    clamp.
     """
     if deadline is None:
         return grace
@@ -131,6 +135,12 @@ def _wait_out_kill_grace(remaining: float, cancelled: Optional[threading.Event])
 
     ``cancelled`` makes the wait interruptible so the timer can be handed
     over to another waiter instead of two clocks running at once.
+
+    The ``remaining > 0`` guard is load-bearing: a deadline already in the
+    past must fire the kill now, and ``time.sleep`` of a negative duration
+    would raise into ``hard_kill_on_rank_crash``'s blanket except and drop
+    the kill -- precisely in the case (cleanup outlasted the grace) the
+    watchdog exists for.
     """
     if cancelled is None:
         if remaining > 0:
diff --git a/tensorrt_llm/_torch/pyexecutor/py_executor.py b/tensorrt_llm/_torch/pyexecutor/py_executor.py
index 704252a54914..d7ee462d687a 100644
--- a/tensorrt_llm/_torch/pyexecutor/py_executor.py
+++ b/tensorrt_llm/_torch/pyexecutor/py_executor.py
@@ -2887,6 +2887,12 @@ def handle_executed_batches(executed_batch_num: int):
                 self.iter_counter += 1
 
             # Stage 5: Handle remaining executed batches in the queue.
+            # Note: _event_loop_completed was already set at the break above,
+            # so a raise in this drain is classified as benign teardown rather
+            # than a peer-stranding crash. That is deliberate: reaching here
+            # means every rank observed should_stop_processing, so no peer is
+            # parked in a collective waiting on this one -- this drain only
+            # consumes from a rank-local queue.
             while self.unhandled_batch_counter > 0:
                 with nvtx_range("get_executed_batch"):
                     executed_batch = self.executed_batch_response_queue.get()
diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index f3ba84dd5a3f..ef1a6d60fa8e 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -152,14 +152,17 @@ def test_rank_crash_kill_fires_for_multi_rank(monkeypatch):
 
 
 def _assert_slept_then_killed(order, grace):
-    """Assert the grace was slept out before the kill.
+    """Assert the grace was slept out EXACTLY ONCE before EXACTLY ONE kill.
 
     Patching time.sleep is process-wide, so an unrelated background thread can
-    append its own ("sleep", x) while this runs. Assert on the entries this
-    test is about rather than on exact list equality, which would flake.
+    append its own ("sleep", x) while this runs; asserting exact list equality
+    would flake on that. But the counts must still be pinned: sleeping the
+    grace twice before killing (i.e. crash + 2*grace) is precisely the bug
+    class this PR series introduced with its two independent timers, and a
+    membership-only check accepts it.
     """
-    assert ("sleep", grace) in order, order
-    assert "kill" in order, order
+    assert order.count(("sleep", grace)) == 1, order
+    assert order.count("kill") == 1, order
     assert order.index(("sleep", grace)) < order.index("kill"), order
 
 
@@ -279,39 +282,51 @@ def test_kill_keeps_original_deadline_on_handover(monkeypatch):
     """Handing the kill over must not restart the grace clock.
 
     The caller cancels the watchdog once cleanup returns and carries the kill
-    itself; passing the watchdog's deadline keeps the kill at crash + grace
-    instead of crash + 2*grace. Uses a real (short) sleep rather than patching
-    time.sleep process-wide, so no background thread can busy-spin into the
-    assertion.
+    itself; passing the watchdog's deadline must make it sleep the REMAINING
+    time, not a fresh grace. Asserted on the exact duration handed to sleep
+    (with monotonic pinned) rather than on wall-clock, so the margin does not
+    depend on scheduling luck on a loaded CI node.
     """
-    kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    order = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hd_module.time, "sleep", lambda s: order.append(("sleep", s)))
+    monkeypatch.setattr(hd_module.time, "monotonic", lambda: 1000.0)
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
-    # Nearly all of the 30s grace has already been burned by the watchdog.
-    t0 = time.monotonic()
-    assert hard_kill_on_rank_crash(world_size=2, deadline=t0 + 0.3) is True
-    elapsed = time.monotonic() - t0
-    assert kills == [1]
-    # Slept out the REMAINING 0.3s, not a fresh 30s grace.
-    assert elapsed == pytest.approx(0.3, abs=0.25)
+    # 29.5s of the 30s grace has already been burned by the watchdog.
+    assert hard_kill_on_rank_crash(world_size=2, deadline=1000.5) is True
+    # Exactly one 0.5s sleep, then exactly one kill -- never a second grace.
+    assert order.count(("sleep", 0.5)) == 1, order
+    assert not any(s == ("sleep", 30.0) for s in order), order
+    _assert_slept_then_killed(order, 0.5)
 
 
 def test_kill_fires_immediately_when_deadline_already_passed(monkeypatch):
-    """A deadline in the past must fire now, not raise into the blanket except.
+    """A deadline in the past must fire the kill now, and never sleep negative.
 
-    Without the max(0.0, ...) clamp this sleeps a negative duration, raises,
-    and hard_kill_on_rank_crash returns False -- silently skipping the kill in
-    exactly the case the watchdog exists for (cleanup outlasted the grace).
+    time.sleep of a negative duration raises into hard_kill_on_rank_crash's
+    blanket except, which would return False and silently skip the kill --
+    exactly in the case the watchdog exists for (cleanup outlasted the grace).
     """
-    kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    order = []
+    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hd_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
     t0 = time.monotonic()
     assert hard_kill_on_rank_crash(world_size=2, deadline=t0 - 100.0) is True
-    assert kills == [1]
-    assert time.monotonic() - t0 < 1.0
+    assert order == ["kill"], order
+    assert not [s for s in order if isinstance(s, tuple) and s[1] < 0], order
+
+
+def test_wait_out_kill_grace_never_sleeps_negative(monkeypatch):
+    """The `remaining > 0` guard, not the deadline clamp, is what protects here."""
+    slept = []
+    monkeypatch.setattr(hd_module.time, "sleep", lambda s: slept.append(s))
+    assert hd_module._wait_out_kill_grace(-100.0, None) is True
+    assert slept == []
+    # The cancellable path must also return promptly, not wait forever.
+    assert hd_module._wait_out_kill_grace(-100.0, threading.Event()) is True
 
 
 # --------------------------------------------------------------------------
@@ -612,53 +627,140 @@ def _executor_loop_ast_nodes():
     }
 
 
-def test_every_executor_loop_break_sets_the_completion_sentinel():
+def _outer_while(fn):
+    """The `while True:` that IS the event loop (the function's own outermost)."""
+    import ast
+
+    for node in ast.walk(fn):
+        if isinstance(node, ast.While):
+            return node
+    return None
+
+
+def _loop_terminating_breaks(loop):
+    """(block, index, break_node) for every break that exits ``loop`` itself.
+
+    Recurses through if/try/with, but stops at nested for/while/def: a break
+    inside those binds to the inner construct, not to the event loop.
+    """
+    import ast
+
+    found = []
+
+    def visit(block):
+        for i, stmt in enumerate(block):
+            if isinstance(stmt, ast.Break):
+                found.append((block, i, stmt))
+            elif isinstance(
+                stmt, (ast.For, ast.AsyncFor, ast.While, ast.FunctionDef, ast.AsyncFunctionDef)
+            ):
+                continue  # binds to the inner construct
+            else:
+                for field in ("body", "orelse", "finalbody", "handlers"):
+                    inner = getattr(stmt, field, None)
+                    if isinstance(inner, list):
+                        if field == "handlers":
+                            for h in inner:
+                                visit(h.body)
+                        else:
+                            visit(inner)
+
+    visit(loop.body)
+    return found
+
+
+def _sets_sentinel_true(stmt):
+    """Exactly `self._event_loop_completed = True` -- object and value both checked."""
     import ast
 
+    if not isinstance(stmt, ast.Assign):
+        return False
+    if not (isinstance(stmt.value, ast.Constant) and stmt.value.value is True):
+        return False
+    return any(
+        isinstance(t, ast.Attribute)
+        and t.attr == "_event_loop_completed"
+        and isinstance(t.value, ast.Name)
+        and t.value.id == "self"
+        for t in stmt.targets
+    )
+
+
+def test_loop_terminating_break_sets_the_completion_sentinel():
+    """Only the break that exits the OUTER `while True` terminates the event loop.
+
+    Deliberately not "every break": these loops contain inner `for` loops, and
+    an inner break does not end the event loop. Demanding the sentinel there
+    would instruct a contributor to set it while the loop is still running,
+    which makes every later rank-local crash look like a clean shutdown and
+    silently disables the kill -- the same class of bug this predicate has
+    already regressed into twice.
+    """
+
     loops = _executor_loop_ast_nodes()
     assert set(loops) == {"_executor_loop", "_executor_loop_pp", "_executor_loop_overlap"}, (
         f"executor loops renamed or removed: {sorted(loops)}"
     )
 
     for name, fn in loops.items():
-        # Only breaks belonging to THIS function (not to a nested def) end
-        # the event loop.
-        nested = {
-            id(n)
-            for d in ast.walk(fn)
-            if isinstance(d, (ast.FunctionDef, ast.AsyncFunctionDef)) and d is not fn
-            for n in ast.walk(d)
-        }
-        own_breaks = [n for n in ast.walk(fn) if isinstance(n, ast.Break) and id(n) not in nested]
-        assert own_breaks, f"{name}: no break found -- did the loop exit change?"
-
-        checked = 0
-        for parent in ast.walk(fn):
-            for field in ("body", "orelse", "finalbody"):
-                block = getattr(parent, field, None)
-                if not isinstance(block, list):
-                    continue
-                for i, stmt in enumerate(block):
-                    if not isinstance(stmt, ast.Break) or id(stmt) in nested:
-                        continue
-                    checked += 1
-                    prev = block[i - 1] if i else None
-                    sets_sentinel = isinstance(prev, ast.Assign) and any(
-                        isinstance(t, ast.Attribute) and t.attr == "_event_loop_completed"
-                        for t in prev.targets
-                    )
-                    assert sets_sentinel, (
-                        f"{name}: the `break` at line {stmt.lineno} is not preceded "
-                        "by `self._event_loop_completed = True`. Every normal exit "
-                        "must set it, or _event_loop_wrapper treats a clean "
-                        "shutdown as a peer-stranding crash and SIGKILLs the job."
-                    )
-        assert checked == len(own_breaks)
-
-
-def test_completion_sentinel_is_initialized_false():
+        outer = _outer_while(fn)
+        assert outer is not None, f"{name}: no `while` loop found -- did the loop shape change?"
+
+        terminating = _loop_terminating_breaks(outer)
+        assert len(terminating) == 1, (
+            f"{name}: expected exactly 1 loop-terminating break, found "
+            f"{len(terminating)} at lines {[b.lineno for _, _, b in terminating]}. "
+            "A new normal-exit path must also set self._event_loop_completed = True."
+        )
+
+        block, idx, brk = terminating[0]
+        prev = block[idx - 1] if idx else None
+        assert _sets_sentinel_true(prev), (
+            f"{name}: the loop-terminating `break` at line {brk.lineno} is not "
+            "preceded by `self._event_loop_completed = True`. Without it "
+            "_event_loop_wrapper treats a clean shutdown as a peer-stranding "
+            "crash and SIGKILLs the job. (Inner-loop breaks must NOT set it.)"
+        )
+
+
+def test_completion_sentinel_is_reset_per_event_loop_run():
+    """The reset in _event_loop_wrapper is load-bearing, not redundant with __init__.
+
+    PyExecutor outlives a single loop run; without the reset a second run
+    starts with the sentinel left True by the first, so a genuine crash in it
+    is misread as a clean shutdown and no kill is armed.
+    """
     import inspect
 
     from tensorrt_llm._torch.pyexecutor import py_executor as pe
 
-    assert "self._event_loop_completed = False" in inspect.getsource(pe.PyExecutor.__init__)
+    assert "self._event_loop_completed = False" in inspect.getsource(
+        pe.PyExecutor._event_loop_wrapper
+    )
+
+
+def test_second_loop_run_still_kills_after_a_clean_first_run(monkeypatch):
+    """Behavioral guard for the reset above: run clean, then crash, on ONE executor."""
+    from tensorrt_llm._torch.pyexecutor import py_executor as pe
+
+    events = []
+    _stub_kill_paths(pe, monkeypatch, events)
+    ex = _bare_executor(pe, monkeypatch, world_size=4)
+    ex._executor_loop_cleanup = lambda: events.append("cleanup")
+
+    # Run 1: reaches the normal-exit break, leaving the sentinel True.
+    def clean():
+        ex._event_loop_completed = True
+
+    ex.event_loop = clean
+    ex._event_loop_wrapper()
+    assert events == ["cleanup"]
+    assert ex._event_loop_completed is True
+
+    # Run 2 on the SAME executor: a genuine crash must still arm the kill.
+    events.clear()
+    ex.event_loop = lambda: (_ for _ in ()).throw(ValueError("boom"))
+    with pytest.raises(ValueError, match="boom"):
+        ex._event_loop_wrapper()
+
+    assert events == [("watchdog", 4), "cleanup", "cancel", ("kill", 4, 1234.5)]

From e9696f54d56b23acc0ed3053e3f82fbf48e7997f Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Tue, 28 Jul 2026 17:57:16 +0000
Subject: [PATCH 09/10] [TRTLLM-13409][test] pin the AST guard to the `while
 True:` event loop, not walk order

_executor_loop_pp contains four while loops, one of them (the Stage-5
drain) a sibling of the event loop under the same with-block. Selecting
by ast.walk order happened to pick the right one, but a body reorder
would silently repoint the guard at the drain. Select by the `True`
test and assert uniqueness instead.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../executor/test_hang_detector_kill.py       | 24 +++++++++++++++----
 1 file changed, 19 insertions(+), 5 deletions(-)

diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index ef1a6d60fa8e..ae11170a9665 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -628,13 +628,27 @@ def _executor_loop_ast_nodes():
 
 
 def _outer_while(fn):
-    """The `while True:` that IS the event loop (the function's own outermost)."""
+    """The `while True:` that IS the event loop.
+
+    Selected by its `True` test, not by walk order: _executor_loop_pp contains
+    three other `while`s, one of them (the Stage-5 drain) a SIBLING of the
+    event loop, so relying on ast.walk ordering would silently point the guard
+    at the wrong loop if the body were ever reordered.
+    """
     import ast
 
-    for node in ast.walk(fn):
-        if isinstance(node, ast.While):
-            return node
-    return None
+    candidates = [
+        node
+        for node in ast.walk(fn)
+        if isinstance(node, ast.While)
+        and isinstance(node.test, ast.Constant)
+        and node.test.value is True
+    ]
+    assert len(candidates) == 1, (
+        f"{fn.name}: expected exactly one `while True:` (the event loop), "
+        f"found {len(candidates)} at lines {[c.lineno for c in candidates]}"
+    )
+    return candidates[0]
 
 
 def _loop_terminating_breaks(loop):

From 0a449fa3beabbcaa120eb0cb5361c43f6c873712 Mon Sep 17 00:00:00 2001
From: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
Date: Mon, 3 Aug 2026 02:15:18 +0000
Subject: [PATCH 10/10] [TRTLLM-13409][test] unify the hang_detector module
 alias after rebase

main and this branch each introduced an alias for the same module -- main
used `hang_detector_module`, this branch used `hd_module`. The rebase
resolved the import-block conflict in favour of main's name, but later
commits in this series reintroduced `hd_module.` call sites, leaving the
tip importing one alias and calling the other (19 undefined references,
NameError on collection).

Converge on main's `hang_detector_module` so the branch follows trunk
convention rather than the other way round.

Signed-off-by: JunyiXu-nv <219237550+JunyiXu-nv@users.noreply.github.com>
---
 .../executor/test_hang_detector_kill.py       | 38 +++++++++----------
 1 file changed, 19 insertions(+), 19 deletions(-)

diff --git a/tests/unittest/_torch/executor/test_hang_detector_kill.py b/tests/unittest/_torch/executor/test_hang_detector_kill.py
index ae11170a9665..b433ee585f65 100644
--- a/tests/unittest/_torch/executor/test_hang_detector_kill.py
+++ b/tests/unittest/_torch/executor/test_hang_detector_kill.py
@@ -138,14 +138,14 @@ def test_propagate_hard_kill_self_sigkills_without_mpi():
 def test_rank_crash_kill_single_rank_is_noop(monkeypatch):
     """No peers to unblock: the worker's own death already carries the error."""
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     assert hard_kill_on_rank_crash(world_size=1) is False
     assert kills == []
 
 
 def test_rank_crash_kill_fires_for_multi_rank(monkeypatch):
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
     assert hard_kill_on_rank_crash(world_size=4) is True
     assert kills == [1]
@@ -169,7 +169,7 @@ class this PR series introduced with its two independent timers, and a
 def test_rank_crash_kill_sleeps_grace_before_kill(monkeypatch):
     """The grace must elapse BEFORE the kill so cleaner error paths win the race."""
     order = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: order.append("kill"))
     monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "2.5")
     assert hard_kill_on_rank_crash(world_size=2) is True
@@ -178,7 +178,7 @@ def test_rank_crash_kill_sleeps_grace_before_kill(monkeypatch):
 
 def test_rank_crash_kill_disabled_by_negative_grace(monkeypatch):
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "-1")
     assert hard_kill_on_rank_crash(world_size=8) is False
     assert kills == []
@@ -187,7 +187,7 @@ def test_rank_crash_kill_disabled_by_negative_grace(monkeypatch):
 def test_rank_crash_kill_invalid_grace_uses_default(monkeypatch):
     """A malformed env value must not disable the kill (fail-safe default)."""
     order = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: order.append("kill"))
     monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "bogus")
     assert hard_kill_on_rank_crash(world_size=2) is True
@@ -200,7 +200,7 @@ def test_rank_crash_kill_never_raises(monkeypatch):
     def boom():
         raise RuntimeError("abort machinery broken")
 
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", boom)
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", boom)
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
     assert hard_kill_on_rank_crash(world_size=2) is False
 
@@ -215,7 +215,7 @@ def boom():
 def test_watchdog_kills_while_caller_blocks(monkeypatch):
     """The kill fires from the watchdog thread with no help from the caller."""
     killed = threading.Event()
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", killed.set)
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", killed.set)
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
 
     watchdog = start_rank_crash_kill_watchdog(world_size=2)
@@ -230,7 +230,7 @@ def test_watchdog_kills_while_caller_blocks(monkeypatch):
 
 def test_watchdog_not_armed_for_single_rank(monkeypatch):
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "0")
     assert start_rank_crash_kill_watchdog(world_size=1) is None
     assert kills == []
@@ -238,7 +238,7 @@ def test_watchdog_not_armed_for_single_rank(monkeypatch):
 
 def test_watchdog_not_armed_when_disabled(monkeypatch):
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "-1")
     assert start_rank_crash_kill_watchdog(world_size=8) is None
     assert kills == []
@@ -253,7 +253,7 @@ def test_watchdog_cancel_disarms_this_timer(monkeypatch):
     `crashed` predicate in _event_loop_wrapper.
     """
     kills = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: kills.append(1))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: kills.append(1))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
     watchdog = start_rank_crash_kill_watchdog(world_size=2)
@@ -274,7 +274,7 @@ def test_watchdog_cancel_disarms_this_timer(monkeypatch):
 
 def test_watchdog_deadline_is_grace_from_arming(monkeypatch):
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "5")
-    watchdog = hd_module.RankCrashKillWatchdog(world_size=2, grace=5.0)
+    watchdog = hang_detector_module.RankCrashKillWatchdog(world_size=2, grace=5.0)
     assert watchdog.deadline == pytest.approx(time.monotonic() + 5.0, abs=0.5)
 
 
@@ -288,9 +288,9 @@ def test_kill_keeps_original_deadline_on_handover(monkeypatch):
     depend on scheduling luck on a loaded CI node.
     """
     order = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
-    monkeypatch.setattr(hd_module.time, "sleep", lambda s: order.append(("sleep", s)))
-    monkeypatch.setattr(hd_module.time, "monotonic", lambda: 1000.0)
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
+    monkeypatch.setattr(hang_detector_module.time, "monotonic", lambda: 1000.0)
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
     # 29.5s of the 30s grace has already been burned by the watchdog.
@@ -309,8 +309,8 @@ def test_kill_fires_immediately_when_deadline_already_passed(monkeypatch):
     exactly in the case the watchdog exists for (cleanup outlasted the grace).
     """
     order = []
-    monkeypatch.setattr(hd_module, "propagate_hard_kill", lambda: order.append("kill"))
-    monkeypatch.setattr(hd_module.time, "sleep", lambda s: order.append(("sleep", s)))
+    monkeypatch.setattr(hang_detector_module, "propagate_hard_kill", lambda: order.append("kill"))
+    monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: order.append(("sleep", s)))
     monkeypatch.setenv(RANK_CRASH_KILL_GRACE_ENV, "30")
 
     t0 = time.monotonic()
@@ -322,11 +322,11 @@ def test_kill_fires_immediately_when_deadline_already_passed(monkeypatch):
 def test_wait_out_kill_grace_never_sleeps_negative(monkeypatch):
     """The `remaining > 0` guard, not the deadline clamp, is what protects here."""
     slept = []
-    monkeypatch.setattr(hd_module.time, "sleep", lambda s: slept.append(s))
-    assert hd_module._wait_out_kill_grace(-100.0, None) is True
+    monkeypatch.setattr(hang_detector_module.time, "sleep", lambda s: slept.append(s))
+    assert hang_detector_module._wait_out_kill_grace(-100.0, None) is True
     assert slept == []
     # The cancellable path must also return promptly, not wait forever.
-    assert hd_module._wait_out_kill_grace(-100.0, threading.Event()) is True
+    assert hang_detector_module._wait_out_kill_grace(-100.0, threading.Event()) is True
 
 
 # --------------------------------------------------------------------------