nabinchha · Oct 30, 2026 · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026 · Oct 28, 2026
@@ -12,7 +12,7 @@ The submit environment needs:
 - `sbatch`, `squeue`, `sacct`, `scancel`, `scontrol`, and `sinfo`
 - Access to a shared workspace that is visible from the submit host and every eligible compute node
 
-Compute nodes need access to the same shared Python environment at the exact path used by the submit command. Local serving requires Enroot 3.5 or later for both digest-pinned OCI imports and existing SQSH inspection, and serving images must contain vLLM. Hosted-only runs require access to their external endpoints and do not need serving images or Enroot.
+Compute nodes need `srun`, `scancel`, and `scontrol`, plus access to the same shared Python environment at the exact path used by the submit command. Local serving requires Enroot 3.5 or later for both digest-pinned OCI imports and existing SQSH inspection, and serving images must contain vLLM. Hosted-only runs require access to their external endpoints and do not need serving images or Enroot.
 
 Your Slurm administrator can confirm the account, GPU partition, CPU image-build partition, GPU request mode, and shared filesystem paths for your cluster.
 

@@ -81,6 +81,20 @@ Inspect the attempt readiness record and Slurm stderr under the managed run dire
 
 ## A shard failed
 
+### Backend health is lost during generation
+
+After startup, the allocation checks each logical endpoint's aggregate backend health every second. Health probes use separate connection capacity so they can run while inference connections are saturated. If a required backend becomes unhealthy, including while its process remains alive, the monitor records `runtime_health_lost` in the attempt readiness evidence and fails the attempt. Inspect the backend logs before retrying; a healthy spare replica does not make the deployment ready when another required replica has failed.
+
+### Inspect routing and overload behavior
+
+Endpoint stderr contains structured `slurm_proxy_request` events. Each event identifies attempted upstream origins, outcomes, connection reuse, connection duration, time to response headers, total attempt duration, final HTTP status, and whether the response completed. Total attempt duration includes connection establishment and response streaming. A reused connection has a connection duration of zero; unavailable timings are `null`. Logs exclude request paths, authorization headers, prompts, and generated content.
+
+The allocation-local endpoint's `/metrics` response includes request and attempt counts, per-backend activity, connection opens/reuse/failures/evictions, retries by reason, final HTTP status counts, and timeouts. Evictions include expired, stale, and discarded connections; normal session shutdown is excluded. Timeouts include terminal attempts and interrupted response streams, even when no retry follows. These counters describe this endpoint process and reset when it restarts.
+
+During cleanup, health monitoring stops first. The controller sends `TERM` directly to each endpoint's Slurm task step using its recorded job and step IDs. Logical endpoints then stop accepting requests and drain active responses for up to 10 seconds while their `srun` launchers and serving backends remain available. The controller allows 12 seconds for draining and step exit before forcing remaining endpoints to stop and terminating other steps. Responses exceeding the drain deadline can be interrupted.
+
+### Retry a failed shard
+
 Use `status` to find the failed task and attempt. Batch stderr identifies the allocation runtime phase and exit status; synchronous step failures also report their durable stdout and stderr paths. Preview a sparse retry before submission:
 
 ```bash

@@ -3,6 +3,9 @@
 
 DD_MANAGED_PIDS=()
 DD_REQUIRED_PIDS=()
+DD_ENDPOINT_PIDS=()
+declare -A DD_ENDPOINT_STEP_FILES=()
+DD_HEALTH_PID=
 DD_CLEANUP_COMPLETE=0
 
 dd_start_runtime_timer() {
@@ -44,25 +47,66 @@ dd_wait_for_client() {
 
 dd_cleanup_steps() {
     ((DD_CLEANUP_COMPLETE == 0)) || return 0
-    local index pid position
-    local -a indices=()
-    ((${#DD_MANAGED_PIDS[@]} == 0)) || indices=("${!DD_MANAGED_PIDS[@]}")
-    for ((position = ${#indices[@]} - 1; position >= 0; position--)); do
-        index=${indices[position]}
-        pid=${DD_MANAGED_PIDS[index]}
+    if [[ -n ${DD_HEALTH_PID} ]]; then
+        dd_stop_steps 1 "${DD_HEALTH_PID}"
+    fi
+    dd_drain_endpoints
+    dd_stop_steps 1 "${DD_MANAGED_PIDS[@]+"${DD_MANAGED_PIDS[@]}"}"
+    DD_CLEANUP_COMPLETE=1
+}
+
+dd_drain_endpoints() {
+    local pid identity_file step_identity
+    for pid in "${DD_ENDPOINT_PIDS[@]+"${DD_ENDPOINT_PIDS[@]}"}"; do
+        kill -0 "${pid}" 2>/dev/null || continue
+        identity_file=${DD_ENDPOINT_STEP_FILES[${pid}]:-}
+        step_identity=
+        if [[ -f ${identity_file} && ! -L ${identity_file} ]]; then
+            IFS= read -r step_identity < "${identity_file}" || true
+        fi
+        if [[ ${step_identity} =~ ^([1-9][0-9]*)\.([0-9]+)$ && ${BASH_REMATCH[1]} == "${SLURM_JOB_ID:-}" ]]; then
+            # TERM to srun becomes KILL; signal only this endpoint's task step.
+            scancel --quiet --signal=TERM "${step_identity}" || true
+        fi
+    done
+    # Keep launchers and backends alive through the proxy's 10-second drain.
+    dd_wait_steps 12 "${DD_ENDPOINT_PIDS[@]+"${DD_ENDPOINT_PIDS[@]}"}"
+    dd_stop_steps 1 "${DD_ENDPOINT_PIDS[@]+"${DD_ENDPOINT_PIDS[@]}"}"
+}
+
+dd_stop_steps() {
+    local grace_seconds=$1
+    shift
+    (($# > 0)) || return 0
+    local pid
+    for pid in "$@"; do
         kill -0 "${pid}" 2>/dev/null && kill -TERM "${pid}" 2>/dev/null || true
     done
-    dd_sleep 0.2
-    for ((position = ${#indices[@]} - 1; position >= 0; position--)); do
-        index=${indices[position]}
-        pid=${DD_MANAGED_PIDS[index]}
+    dd_wait_steps "${grace_seconds}" "$@"
+    for pid in "$@"; do
         kill -0 "${pid}" 2>/dev/null && kill -KILL "${pid}" 2>/dev/null || true
     done
-    for ((position = ${#indices[@]} - 1; position >= 0; position--)); do
-        index=${indices[position]}
-        wait "${DD_MANAGED_PIDS[index]}" 2>/dev/null || true
+    for pid in "$@"; do
+        wait "${pid}" 2>/dev/null || true
+    done
+}
+
+dd_wait_steps() {
+    local grace_seconds=$1
+    shift
+    (($# > 0)) || return 0
+    local pid pending deadline=$((SECONDS + grace_seconds))
+    while ((SECONDS < deadline)); do
+        pending=0
+        for pid in "$@"; do
+            if kill -0 "${pid}" 2>/dev/null; then
+                pending=1
+                break
+            fi
+        done
+        ((pending != 0)) || break
+        dd_sleep 0.1
     done
-    DD_CLEANUP_COMPLETE=1
 }
 
 dd_stop_runtime_timer() {

@@ -17,6 +17,7 @@
 from data_designer.slurm.runtime.bootstrap import build_runtime_manifest
 from data_designer.slurm.runtime.context import load_allocation_context
 from data_designer.slurm.runtime.errors import SlurmRuntimeError, SlurmRuntimeErrorCode
+from data_designer.slurm.runtime.health import monitor_runtime_health
 from data_designer.slurm.runtime.logs import execution_log_directory
 from data_designer.slurm.runtime.models import AllocationContext
 from data_designer.slurm.runtime.ports import resolve_allocation_deployments
@@ -52,6 +53,8 @@ def main(arguments: Sequence[str] | None = None) -> int:
             _ready(parsed, os.environ)
         elif parsed.operation == "client":
             _client(parsed, os.environ)
+        elif parsed.operation == "monitor":
+            _monitor(parsed, os.environ)
         elif parsed.operation == "succeed":
             _succeed(parsed, os.environ)
         else:
@@ -73,7 +76,7 @@ def main(arguments: Sequence[str] | None = None) -> int:
 def _parse_arguments(arguments: Sequence[str] | None) -> argparse.Namespace:
     parser = argparse.ArgumentParser(prog="data-designer-slurm-runtime")
     subparsers = parser.add_subparsers(dest="operation", required=True)
-    for operation in ("prepare", "ready", "succeed", "fail"):
+    for operation in ("prepare", "ready", "succeed", "fail", "monitor"):
         subparser = subparsers.add_parser(operation)
         _add_context_arguments(subparser)
     prepare = subparsers.choices["prepare"]
@@ -150,6 +153,17 @@ def _ready(arguments: argparse.Namespace, environment: Mapping[str, str]) -> Non
     )
 
 
+def _monitor(arguments: argparse.Namespace, environment: Mapping[str, str]) -> None:
+    context, writer = _load_context(arguments, environment)
+    readiness = writer.load_readiness(context.shard.shard_id, context.attempt.attempt_id)
+    deployments = _resolve_deployments(context, environment)
+    monitor_runtime_health(
+        writer,
+        readiness,
+        {deployment.deployment_id: deployment.logical_endpoint.port for deployment in deployments},
+    )
+
+
 def _client(
     arguments: argparse.Namespace,
     environment: Mapping[str, str],
@@ -215,7 +229,18 @@ def _client(
 
 def _succeed(arguments: argparse.Namespace, environment: Mapping[str, str]) -> None:
     context, writer = _load_context(arguments, environment)
-    stopped_at = _write_stopped_readiness(context, writer)
+    # Cleanup has stopped the monitor, so its last persisted health evidence is final.
+    readiness = writer.load_readiness(context.shard.shard_id, context.attempt.attempt_id)
+    if readiness.state is ReadinessState.FAILED or any(
+        deployment.state is ReadinessState.FAILED
+        or (deployment.last_probe is not None and deployment.last_probe.outcome is ProbeOutcome.FAILURE)
+        for deployment in readiness.deployments
+    ):
+        raise SlurmRuntimeError(
+            SlurmRuntimeErrorCode.FINALIZATION_FAILED,
+            "cannot finalize a successful allocation after required inference service failure",
+        )
+    stopped_at = _write_stopped_readiness(context, writer, previous=readiness)
     attempt = writer.load_attempt(context.shard.shard_id, context.attempt.attempt_id)
     if attempt.candidate_output is None:
         raise SlurmRuntimeError(
@@ -316,6 +341,10 @@ def _write_failed_and_stopped_readiness(context: AllocationContext, writer: Slur
     previous = _load_optional_readiness(context, writer)
     if previous is None:
         return max(datetime.now(UTC), context.attempt.updated_at)
+    if previous.state is ReadinessState.STOPPED:
+        return _now(context.attempt, previous)
+    if previous.state is ReadinessState.FAILED:
+        return _write_stopped_readiness(context, writer, previous=previous)
     failed_at = _now(context.attempt, previous)
     failed = writer.write_readiness(
         AttemptReadiness(

@@ -96,6 +96,9 @@ dd_slurm_run_allocation() {
     dd_require_running
     DD_RUNTIME_PHASE=ready
     dd_run_bound_control_phase ready
+    if ((DD_EXPECTED_GPUS > 0)); then
+        dd_start_health_monitor
+    fi
 
     DD_RUNTIME_PHASE=client
     dd_runtime_log 'Starting Data Designer client...'
@@ -133,7 +136,7 @@ dd_run_bound_control_phase() {
 
 dd_verify_host_context() {
     local tool
-    for tool in bash sha256sum tar jq srun scontrol getent curl find; do
+    for tool in bash sha256sum tar jq srun scancel scontrol getent curl find; do
         command -v "${tool}" >/dev/null || {
             printf 'required allocation tool %q is unavailable\n' "${tool}" >&2
             return 69
@@ -234,9 +237,11 @@ dd_start_endpoints() {
     for step_id in "${DD_STEP_IDS[@]+"${DD_STEP_IDS[@]}"}"; do
         DD_STEP_STARTED_AT[${step_id}]=${SECONDS}
         dd_read_step "${DD_RUNTIME_MANIFEST}" "${step_id}"
-        rm -f -- "${DD_STEP_STDOUT}.listening"
-        dd_start_step "${DD_RUNTIME_MANIFEST}" "${step_id}"
+        rm -f -- "${DD_STEP_STDOUT}.listening" "${DD_STEP_STDOUT}.step-id"
+        dd_start_step "${DD_RUNTIME_MANIFEST}" "${step_id}" "${DD_STEP_STDOUT}.step-id"
         dd_register_required_pid "${DD_LAST_PID}"
+        DD_ENDPOINT_PIDS+=("${DD_LAST_PID}")
+        DD_ENDPOINT_STEP_FILES[${DD_LAST_PID}]=${DD_STEP_STDOUT}.step-id
         dd_require_running
     done
 }

@@ -0,0 +1,105 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+
+"""Continuous aggregate health checks for a running allocation."""
+
+from __future__ import annotations
+
+import http.client
+import json
+import time
+from collections.abc import Mapping
+from datetime import UTC, datetime
+
+from data_designer.slurm.runtime.errors import SlurmRuntimeError, SlurmRuntimeErrorCode
+from data_designer.slurm.state import (
+    AttemptReadiness,
+    DeploymentReadiness,
+    ProbeEvidence,
+    ProbeOutcome,
+    ReadinessState,
+    SlurmStateWriter,
+)
+
+_HEALTH_TIMEOUT_SECONDS = 6.0
+_MAXIMUM_HEALTH_BYTES = 4096
+
+
+def monitor_runtime_health(
+    writer: SlurmStateWriter,
+    readiness: AttemptReadiness,
+    endpoint_ports: Mapping[str, int],
+    *,
+    poll_interval_seconds: float = 1.0,
+) -> None:
+    """Publish health evidence until readiness stops or a required endpoint fails.
+
+    The allocation controller owns process termination and attempt finalization.
+    This monitor fails on service loss even if all serving processes remain alive.
+    """
+    if poll_interval_seconds <= 0:
+        raise ValueError("health poll interval must be positive")
+    if set(endpoint_ports) != {deployment.deployment_id for deployment in readiness.deployments}:
+        raise ValueError("health endpoints must match readiness deployments")
+    while readiness.state is ReadinessState.READY:
+        readiness = writer.write_readiness(probe_runtime_health(readiness, endpoint_ports))
+        if readiness.state is ReadinessState.FAILED:
+            raise SlurmRuntimeError(SlurmRuntimeErrorCode.STEP_FAILED, "required inference service lost health")
+        time.sleep(poll_interval_seconds)
+        readiness = writer.load_readiness(readiness.shard_id, readiness.attempt_id)
+
+
+def probe_runtime_health(readiness: AttemptReadiness, endpoint_ports: Mapping[str, int]) -> AttemptReadiness:
+    """Check every logical endpoint and preserve per-deployment backend counts."""
+    deployments: list[DeploymentReadiness] = []
+    for deployment in readiness.deployments:
+        count = _ready_backend_count(endpoint_ports[deployment.deployment_id], deployment.expected_backends)
+        healthy = count == deployment.expected_backends
+        observed_at = datetime.now(UTC)
+        deployments.append(
+            deployment.model_copy(
+                update={
+                    "state": ReadinessState.READY if healthy else ReadinessState.FAILED,
+                    "ready_backends": count,
+                    "last_probe": ProbeEvidence(
+                        observed_at=observed_at,
+                        outcome=ProbeOutcome.SUCCESS if healthy else ProbeOutcome.FAILURE,
+                        reason_code="runtime_healthy" if healthy else "runtime_health_lost",
+                        redacted_message=f"{count} of {deployment.expected_backends} backends ready",
+                    ),
+                }
+            )
+        )
+    return readiness.model_copy(
+        update={
+            "revision": readiness.revision + 1,
+            "updated_at": datetime.now(UTC),
+            "state": (
+                ReadinessState.READY
+                if all(deployment.state is ReadinessState.READY for deployment in deployments)
+                else ReadinessState.FAILED
+            ),
+            "deployments": tuple(deployments),
+        }
+    )
+
+
+def _ready_backend_count(port: int, expected: int) -> int:
+    connection = http.client.HTTPConnection("127.0.0.1", port, timeout=_HEALTH_TIMEOUT_SECONDS)
+    try:
+        connection.request("GET", "/health", headers={"Connection": "close"})
+        response = connection.getresponse()
+        body = response.read(_MAXIMUM_HEALTH_BYTES + 1)
+        if response.status not in {200, 503} or len(body) > _MAXIMUM_HEALTH_BYTES:
+            return 0
+        payload = json.loads(body)
+        if not isinstance(payload, dict) or type(payload.get("backends")) is not int or payload["backends"] != expected:
+            return 0
+        count = payload.get("backends_ready")
+        if type(count) is not int or not 0 <= count <= expected:
+            return 0
+        return count if response.status == 200 or count < expected else 0
+    except (OSError, http.client.HTTPException, ValueError):
+        return 0
+    finally:
+        connection.close()