feat: [AMD Support via SemiAnalysis][4/4] Optional job observation and diagnostics

CLAUDE.md

@@ -20,6 +20,9 @@ uv run pytest tests/test_e2e.py -v
 # Run single test
 uv run pytest tests/test_e2e.py::TestH100Cluster::test_endpoint_allocation -v
 
+# Observe an existing allocation, streaming its shared log and returning its exit status
+srtctl wait 12345 --log-file outputs/12345/logs/sweep_12345.log
+
 # Auto-fix lint issues
 uv run ruff check --fix src/srtctl/
 uv run ruff format src/srtctl/

docs/installation.md

@@ -51,6 +51,9 @@ If you are trying to deploy onto Grace (GH200, GB200, etc.), you need to use the
 make setup ARCH=aarch64  # or ARCH=x86_64
 ```
 
+Observe an existing job with `srtctl wait JOB_ID --log-file PATH`. It streams
+the shared log and returns the terminal job status, without cancelling jobs.
+
 The setup will:
 
 1. Download NATS, ETCD, uv, and the Tachometer scraper for your compute-node architecture

src/srtctl/cli/do_sweep.py

@@ -548,11 +548,9 @@ def _run_post_eval(self, stop_event: threading.Event) -> int:
         # Pass through eval-related env vars. InferenceX writes multi-node
         # metadata from these variables in append_lm_eval_summary(). The recipe
         # extends this list with post_eval.passthrough_env.
-        # Post-eval replaces the configured benchmark runner with lm-eval, but
-        # it is still a benchmark process. Preserve the recipe's benchmark
-        # environment so integrations can pass runner-specific settings such
-        # as an additional artifact sink through this substituted path.
-        env_to_set = {key: self.runtime.format_string(value) for key, value in self.config.benchmark.env.items()}
+        # Preserve benchmark values verbatim, just like the custom runner:
+        # JSON metadata and shell literals are data, not runtime templates.
+        env_to_set = dict(self.config.benchmark.env)
         for var in [
             *self.config.post_eval.passthrough_env,
             "RUN_EVAL",

src/srtctl/cli/submit.py

@@ -2100,6 +2100,10 @@ def add_common_args(p):
     )
     add_override_args(preflight_parser)
 
+    wait_parser = subparsers.add_parser("wait", help="Wait for a submitted Slurm job and return its exit status")
+    wait_parser.add_argument("job_id", help="Slurm job ID returned by apply --json")
+    wait_parser.add_argument("--log-file", type=Path, help="Stream this shared-filesystem log while waiting")
+
     monitor_parser = subparsers.add_parser("monitor", help="Live dashboard for srt-slurm jobs", add_help=False)
     monitor_parser.add_argument("args", nargs=argparse.REMAINDER)
 
@@ -2406,6 +2410,23 @@ def restore_console() -> None:
         restore_console()
         return
 
+    if args.command == "wait":
+        from srtctl.core.slurm import wait_for_job
+
+        try:
+            result = wait_for_job(args.job_id, log_path=args.log_file)
+        except (ValueError, OSError) as error:
+            console.print(f"[bold red]Error:[/] {error}")
+            restore_console()
+            sys.exit(1)
+        except KeyboardInterrupt:
+            console.print("Stopped observing; the Slurm job has not been cancelled.")
+            restore_console()
+            sys.exit(130)
+        console.print(f"Job {result.job_id}: {result.state} ({result.exit_code}:{result.signal})")
+        restore_console()
+        sys.exit(result.returncode)
+
     if args.command == "monitor":
         from srtctl.cli.monitor import main as _monitor_main
 

src/srtctl/core/fingerprint.py

@@ -47,6 +47,8 @@
 FRAMEWORK_PACKAGES: dict[str, str] = {
     "vllm": "vllm",
     "sglang": "sglang",
+    "sglang-router": "sglang-router",
+    "amd-mori": "amd_mori",
     "tensorrt_llm": "tensorrt-llm",
     "dynamo": "ai-dynamo",
 }

src/srtctl/core/processes.py

@@ -380,7 +380,11 @@ def print_failure_details(self, tail_lines: int = 50) -> None:
                 # Tail the log file if available
                 if proc.log_file and proc.log_file.exists():
                     try:
-                        lines = proc.log_file.read_text().splitlines()
+                        # Backend logs can contain terminal progress output or
+                        # compiler diagnostics with bytes that are not valid
+                        # UTF-8. Failure reporting must remain available even
+                        # when the worker log is not clean text.
+                        lines = proc.log_file.read_text(errors="replace").splitlines()
                         if lines:
                             logger.error("\nLast %d lines of log:", tail_lines)
                             for line in lines[-tail_lines:]:

src/srtctl/core/slurm.py

@@ -13,18 +13,145 @@
 
 import logging
 import os
+import re
 import shlex
 import socket
 import subprocess
+import sys
+import time
 from collections.abc import Sequence
+from dataclasses import dataclass
 from pathlib import Path
+from typing import TextIO
 
 from .ip_utils import get_node_ip
 from .launch_plan import record_srun_command
 
 logger = logging.getLogger(__name__)
 
 
+@dataclass(frozen=True)
+class SlurmJobResult:
+    job_id: str
+    state: str
+    exit_code: int
+    signal: int
+
+    @property
+    def returncode(self) -> int:
+        if self.state == "COMPLETED" and self.exit_code == 0 and self.signal == 0:
+            return 0
+        return self.exit_code or (128 + self.signal if self.signal else 1)
+
+
+def wait_for_job(
+    job_id: str,
+    *,
+    log_path: Path | None = None,
+    output: TextIO | None = None,
+    poll_interval: float = 5.0,
+) -> SlurmJobResult:
+    """Observe an allocation without cancelling or resubmitting it.
+
+    An empty queue is not success: accounting can lag behind squeue. Require
+    the allocation's terminal controller/accounting record and exit status, continuing
+    through transient status-query failures and requeues.
+    """
+    if not re.fullmatch(r"[0-9]+", job_id) or int(job_id) < 1:
+        raise ValueError(f"Expected a single Slurm job ID, got {job_id!r}")
+    if poll_interval <= 0:
+        raise ValueError("poll_interval must be positive")
+    output = output if output is not None else sys.stdout
+    offset = 0
+    terminal_states = {
+        "BOOT_FAIL",
+        "CANCELLED",
+        "COMPLETED",
+        "DEADLINE",
+        "FAILED",
+        "NODE_FAIL",
+        "OUT_OF_MEMORY",
+        "PREEMPTED",
+        "REVOKED",
+        "TIMEOUT",
+    }
+
+    def stream_log() -> None:
+        nonlocal offset
+        if log_path is None:
+            return
+        try:
+            with log_path.open(encoding="utf-8", errors="replace") as stream:
+                if log_path.stat().st_size < offset:
+                    offset = 0
+                stream.seek(offset)
+                while chunk := stream.read(65536):
+                    output.write(chunk)
+                offset = stream.tell()
+                output.flush()
+        except FileNotFoundError:
+            pass  # Pending jobs do not have a log yet.
+
+    while True:
+        stream_log()
+        try:
+            queue = subprocess.run(
+                ["squeue", "--noheader", "--jobs", job_id, "--format=%i"],
+                capture_output=True,
+                text=True,
+                check=False,
+                timeout=15,
+            )
+            if queue.returncode == 0 and queue.stdout.strip():
+                time.sleep(poll_interval)
+                continue
+            # The controller retains recently finished allocations even when
+            # slurmdbd is unavailable. Require the exact job and a terminal
+            # state plus its explicit exit status; an empty queue is not proof.
+            controller = subprocess.run(
+                ["scontrol", "show", "job", "--oneliner", job_id],
+                capture_output=True,
+                text=True,
+                check=False,
+                timeout=15,
+            )
+            if controller.returncode == 0:
+                for line in controller.stdout.splitlines():
+                    fields = dict(re.findall(r"(?:^|\s)(JobId|JobState|ExitCode)=([^\s]+)", line))
+                    state = fields.get("JobState", "")
+                    code = fields.get("ExitCode", "")
+                    if (
+                        fields.get("JobId") == job_id
+                        and state in terminal_states
+                        and re.fullmatch(r"[0-9]+:[0-9]+", code)
+                    ):
+                        exit_code, signal = map(int, code.split(":"))
+                        stream_log()
+                        return SlurmJobResult(job_id, state, exit_code, signal)
+            accounting = subprocess.run(
+                ["sacct", "-X", "--noheader", "--parsable2", "--jobs", job_id, "--format=JobIDRaw,State%40,ExitCode"],
+                capture_output=True,
+                text=True,
+                check=False,
+                timeout=15,
+            )
+            if accounting.returncode == 0:
+                for line in accounting.stdout.splitlines():
+                    fields = [field.strip() for field in line.split("|")]
+                    if len(fields) < 3 or fields[0] != job_id or not fields[1]:
+                        continue
+                    state = fields[1].split()[0].rstrip("+")
+                    if state in terminal_states and re.fullmatch(r"[0-9]+:[0-9]+", fields[2]):
+                        exit_code, signal = map(int, fields[2].split(":"))
+                        stream_log()
+                        return SlurmJobResult(job_id, state, exit_code, signal)
+            else:
+                logger.warning("Waiting for accounting for job %s: %s", job_id, accounting.stderr.strip())
+        except subprocess.TimeoutExpired:
+            logger.warning("Slurm status query timed out for job %s; continuing to observe", job_id)
+        time.sleep(poll_interval)
+
+
 def _get_cluster_bash_preamble() -> str | None:
     """Look up the cluster-wide default_bash_preamble.
 

tests/test_benchmarks.py

@@ -1860,13 +1860,20 @@ def capture_srun(**kwargs):
         assert env_to_set["MODEL_NAME"] == "test-model"
 
     def test_benchmark_env_passthrough(self):
-        """Eval-only substitution preserves the configured benchmark env."""
+        """benchmark.env reaches eval verbatim (no template expansion); workflow variables still win."""
         import os
         import threading
         from unittest.mock import MagicMock, patch
 
+        metadata = '{\n  "name": "cache", "capacity": 128\n}'
         orch = self._make_orchestrator()
-        orch.config.benchmark.env["SRTCTL_LM_EVAL_RESULT_DIR"] = "/results/{job_id}/eval"
+        orch.config.benchmark.env.update(
+            {
+                "KV_OFFLOAD_BACKEND_METADATA": metadata,
+                "CLIENT_LITERAL": "${HOME}/{unresolved}",
+                "ISL": "recipe-value",
+            }
+        )
         stop = threading.Event()
 
         mock_proc = MagicMock()
@@ -1879,13 +1886,16 @@ def capture_srun(**kwargs):
             return mock_proc
 
         with (
-            patch.dict(os.environ, {"EVAL_ONLY": "false"}, clear=False),
+            patch.dict(os.environ, {"EVAL_ONLY": "false", "ISL": "1024"}, clear=False),
             patch("srtctl.cli.do_sweep.wait_for_port", return_value=True),
             patch("srtctl.cli.do_sweep.start_srun_process", side_effect=capture_srun),
         ):
             orch._run_post_eval(stop)
 
-        assert captured_kwargs["env_to_set"]["SRTCTL_LM_EVAL_RESULT_DIR"] == "/results/12345/eval"
+        env = captured_kwargs["env_to_set"]
+        assert env["KV_OFFLOAD_BACKEND_METADATA"] == metadata
+        assert env["CLIENT_LITERAL"] == "${HOME}/{unresolved}"
+        assert env["ISL"] == "1024"
 
     def test_eval_conc_from_env(self):
         """EVAL_CONC from env takes priority over benchmark concurrencies."""

tests/test_process_registry.py

@@ -205,6 +205,35 @@ def test_cleanup(self):
 
         mock_popen.terminate.assert_called_once()
 
+    def test_failure_details_tolerate_non_utf8_worker_logs(self, tmp_path, caplog):
+        """ROCm/compiler output must not hide the useful failure log tail."""
+        log_file = tmp_path / "worker.out"
+        log_file.write_bytes(b"valid line\ninvalid: \xff\xfe\nlast line\n")
+
+        mock_popen = MagicMock(spec=Popen)
+        mock_popen.poll.return_value = 137
+        mock_popen.returncode = 137
+        mock_popen.pid = 12345
+
+        registry = ProcessRegistry(job_id="test_job")
+        registry.add_process(
+            ManagedProcess(
+                name="decode_0",
+                popen=mock_popen,
+                log_file=log_file,
+                node="amd-worker",
+                critical=True,
+            )
+        )
+        assert registry.check_failures()
+
+        with caplog.at_level("ERROR"):
+            registry.print_failure_details()
+
+        assert "Could not read log file" not in caplog.text
+        assert "invalid: \ufffd\ufffd" in caplog.text
+        assert "last line" in caplog.text
+
 
 class TestTieredCleanup:
     """Cleanup signals a whole tier, waits for it, escalates, then moves to the next tier."""