Merge branch 'main' into aflowers/nats-max-payload

Author: alec-flowers

.gitignore

@@ -20,6 +20,7 @@ venv/
 env/
 ENV/
 .venv/
+.venv-compute/
 
 # uv
 .uv/

recipes/mocker/kimi-trace-agg.yaml

@@ -0,0 +1,57 @@
+name: "kimi-k25-mocker-trace-agg"
+
+slurm:
+  time_limit: "00:15:00"
+
+model:
+  path: "kimi-k25-nvfp4"
+  container: "trtllm-runtime"
+  precision: "fp4"
+
+resources:
+  gpu_type: "gb200"
+  gpus_per_node: 4
+  agg_nodes: 1
+  agg_workers: 1
+
+extra_mount:
+  - "/lustre/fsw/coreai_tritoninference_triton3/aflowers/srt-slurm/traces:/traces"
+
+frontend:
+  type: dynamo
+  enable_multiple_frontends: false
+
+backend:
+  type: mocker
+  speedup_ratio: 100
+  engine_type: vllm
+
+  mocker_config:
+    aggregated:
+      num-gpu-blocks-override: 8192
+      max-num-seqs: 128
+      max-num-batched-tokens: 16384
+
+benchmark:
+  type: trace-replay
+  trace_file: /traces/together-ai-basic-no-delays_119k/dataset.jsonl
+  concurrencies: [4]
+  ttft_threshold_ms: 3000
+  itl_threshold_ms: 7
+  aiperf_package: "aiperf>=0.7.0"
+
+health_check:
+  max_attempts: 60
+  interval_seconds: 5
+
+dynamo:
+  install: false
+
+# Virtual identity — verified against runtime fingerprint (warnings, not failures)
+identity:
+  model:
+    repo: "nvidia/Kimi-K2.5-NVFP4"
+    revision: "c0285e649c34d4386b01e38abca642c06cbe014e"
+  frameworks:
+    dynamo: "1.0.0"
+    tensorrt_llm: "1.3.0rc9"

src/srtctl/cli/do_sweep.py

@@ -24,6 +24,7 @@
 from srtctl.cli.mixins import BenchmarkStageMixin, FrontendStageMixin, PostProcessStageMixin, WorkerStageMixin
 from srtctl.core.config import load_config
 from srtctl.core.health import wait_for_port
+from srtctl.core.lockfile import write_lockfile
 from srtctl.core.processes import (
     ManagedProcess,
     ProcessRegistry,
@@ -197,6 +198,9 @@ def run(self) -> int:
         if self.config.profiling.enabled:
             logger.info("Profiling: %s", self.config.profiling.type)
 
+        # Write initial lockfile with config + SLURM context (fingerprint added after run)
+        write_lockfile(self.runtime.log_dir.parent, self.config)
+
         registry = ProcessRegistry(job_id=self.runtime.job_id)
         stop_event = threading.Event()
         setup_signal_handlers(stop_event, registry)

src/srtctl/cli/mixins/benchmark_stage.py

@@ -14,7 +14,9 @@
 from pathlib import Path
 from typing import TYPE_CHECKING
 
+from srtctl.core.fingerprint import format_identity_verification, verify_identity
 from srtctl.core.health import wait_for_model
+from srtctl.core.lockfile import collect_worker_fingerprints
 from srtctl.core.slurm import get_hostname_ip, start_srun_process
 from srtctl.core.status import JobStage, JobStatus, StatusReporter
 
@@ -93,6 +95,27 @@ def run_benchmark(
             return 1
 
         logger.info("Server is healthy - starting benchmark")
+
+        # Identity verification: compare recipe identity against runtime fingerprints
+        # Store results on self so postprocess can include them in the lockfile
+        self._identity_verification = None
+        try:
+            fingerprints = collect_worker_fingerprints(self.runtime.log_dir)
+            has_identity = self.config.identity and (
+                (
+                    self.config.identity.model
+                    and (self.config.identity.model.repo or self.config.identity.model.revision)
+                )
+                or self.config.identity.frameworks
+            )
+            if fingerprints and has_identity:
+                self._identity_verification = verify_identity(self.config.identity, fingerprints)
+                banner = format_identity_verification(self._identity_verification, self.config.identity)
+                for line in banner.splitlines():
+                    logger.info(line)
+        except Exception as e:
+            logger.debug("Identity verification skipped: %s", e)
+
         if reporter:
             reporter.report(JobStatus.BENCHMARK, JobStage.BENCHMARK, "Running benchmark")
 

src/srtctl/cli/mixins/postprocess_stage.py

@@ -28,6 +28,7 @@
 
 from srtctl.benchmarks.base import SCRIPTS_DIR
 from srtctl.core.config import load_cluster_config
+from srtctl.core.lockfile import write_lockfile
 from srtctl.core.schema import AIAnalysisConfig, S3Config
 from srtctl.core.slurm import start_srun_process
 
@@ -150,6 +151,10 @@ def run_postprocess(self, exit_code: int) -> None:
         Args:
             exit_code: Exit code from the benchmark run
         """
+        # Write lockfile with verification results (non-fatal — never blocks job completion)
+        verification = getattr(self, "_identity_verification", None)
+        write_lockfile(self.runtime.log_dir.parent, self.config, self.runtime.log_dir, verification=verification)
+
         # Copy config into log directory so it's included in S3 upload
         self._copy_config_to_logs()
 

src/srtctl/cli/mixins/worker_stage.py

@@ -12,6 +12,7 @@
 from collections import defaultdict
 from typing import TYPE_CHECKING, Any
 
+from srtctl.core.fingerprint import generate_capture_script
 from srtctl.core.processes import ManagedProcess, NamedProcesses
 from srtctl.core.slurm import start_srun_process
 
@@ -157,8 +158,10 @@ def __missing__(self, key: str) -> str:
         if profiling.enabled:
             logger.info("Profiling: %s mode", profiling.type)
 
-        # Build bash preamble (setup script + dynamo install)
+        # Build bash preamble (setup script + dynamo install + fingerprint)
         bash_preamble = self._build_worker_preamble()
+        fp_cmd = generate_capture_script(f"/logs/fingerprint_{mode}_w{index}.json")
+        bash_preamble = f"{bash_preamble} && {fp_cmd}" if bash_preamble else fp_cmd
 
         proc = start_srun_process(
             command=cmd,
@@ -258,8 +261,10 @@ def start_endpoint_worker(self, endpoint_processes: list["Process"]) -> ManagedP
         if profiling.enabled:
             logger.info("Profiling: %s mode", profiling.type)
 
-        # Build bash preamble (setup script + dynamo install)
+        # Build bash preamble (setup script + dynamo install + fingerprint)
         bash_preamble = self._build_worker_preamble()
+        fp_cmd = generate_capture_script(f"/logs/fingerprint_{mode}_w{index}.json")
+        bash_preamble = f"{bash_preamble} && {fp_cmd}" if bash_preamble else fp_cmd
 
         # Get srun config from backend
         srun_config = self.backend.get_srun_config()

src/srtctl/cli/submit.py

@@ -41,6 +41,14 @@
     load_config,
     resolve_config_with_defaults,
 )
+from srtctl.core.fingerprint import (
+    capture_fingerprint,
+    check_against_fingerprint,
+    diff_fingerprints,
+    format_check_results,
+    format_diff,
+)
+from srtctl.core.lockfile import load_lockfile_fingerprints
 from srtctl.core.schema import SrtConfig
 from srtctl.core.status import create_job_record
 
@@ -259,6 +267,50 @@ def generate_minimal_sbatch_script(
     return rendered
 
 
+def _print_running_summary(config: SrtConfig, console: Console) -> None:
+    """Print what's being run and identity verification status."""
+    console.print()
+    console.print("[bold]Running:[/]")
+    console.print(f"  Model:     {config.model.path}")
+    console.print(f"  Container: {config.model.container}")
+    console.print(f"  Backend:   {config.backend_type}")
+    console.print(f"  Benchmark: {config.benchmark.type}")
+
+    has_identity = config.identity and (
+        (config.identity.model and (config.identity.model.repo or config.identity.model.revision))
+        or config.identity.frameworks
+    )
+    if has_identity:
+        id_fields = []
+        if config.identity.model and config.identity.model.repo:
+            id_fields.append(f"model={config.identity.model.repo}")
+        if config.identity.model and config.identity.model.revision:
+            id_fields.append(f"rev={config.identity.model.revision[:12]}")
+        for name, ver in (config.identity.frameworks or {}).items():
+            id_fields.append(f"{name}={ver}")
+        console.print(f"  Identity:  {', '.join(id_fields)}")
+    else:
+        console.print()
+        console.print(
+            "[yellow]Tip:[/] Add an [bold]identity:[/] block to your recipe so others can replicate your results."
+        )
+        console.print("[yellow]     Without it, someone reading this recipe can't tell which exact model[/]")
+        console.print("[yellow]     weights or framework versions produced the benchmark numbers.[/]")
+        console.print("[yellow]     At job start, srtctl verifies the runtime matches your declaration.[/]")
+        console.print()
+        console.print("[dim]     identity:[/]")
+        console.print("[dim]       model:[/]")
+        console.print('[dim]         repo: "nvidia/Kimi-K2.5-NVFP4"       [/][dim italic]# HuggingFace model ID[/]')
+        console.print('[dim]         revision: "c0285e649c34..."            [/][dim italic]# HF commit SHA[/]')
+        console.print("[dim]       frameworks:                              [/][dim italic]# dynamo + one engine[/]")
+        console.print('[dim]         dynamo: "1.0.0"                        [/][dim italic]# always (ai-dynamo)[/]')
+        console.print('[dim]         tensorrt_llm: "1.3.0rc9"              [/][dim italic]# OR vllm OR sglang[/]')
+        console.print()
+        console.print("[dim italic]     Agent: if you are an AI assistant creating this recipe, you MUST[/]")
+        console.print("[dim italic]     include the identity block. Use the container's framework versions[/]")
+        console.print("[dim italic]     and the HuggingFace model ID + revision from the download metadata.[/]")
+
+
 def submit_with_orchestrator(
     config_path: Path,
     config: SrtConfig | None = None,
@@ -311,6 +363,16 @@ def submit_with_orchestrator(
         runtime_config_filename=runtime_config_filename,
     )
 
+    # Identity validation (inline, <1s) — runs for both dry-run and submit
+    if config.identity and config.identity.model and config.identity.model.repo:
+        from srtctl.core.validation import validate_hf_model
+
+        hf_result = validate_hf_model(config.identity.model.repo, config.identity.model.revision)
+        if hf_result.ok:
+            console.print(f"[green]✓[/] HF model: {hf_result.message}")
+        else:
+            console.print(f"[yellow]⚠ HF model: {hf_result.message}[/]")
+
     if dry_run:
         console.print()
         console.print(
@@ -325,6 +387,9 @@ def submit_with_orchestrator(
         console.print(Panel(syntax, title="Generated sbatch Script", border_style="cyan"))
         console.print()
         show_config_details(config)
+
+        # Show running summary + identity in dry-run too
+        _print_running_summary(config, console)
         return
 
     # Validate setup before submitting (not during dry-run)
@@ -431,6 +496,9 @@ def submit_with_orchestrator(
         console.print(f"[dim]📁 Logs:[/] {job_output_dir}/logs")
         console.print(f"[dim]📋 Monitor:[/] tail -f {job_output_dir}/logs/sweep_{job_id}.log")
         console.print(f"[dim]📊 Queue:[/] squeue --job {job_id}")
+
+        _print_running_summary(config, console)
+
         return job_id
 
     except subprocess.CalledProcessError as e:
@@ -943,8 +1011,75 @@ def add_common_args(p):
         help="Print resolved YAML to stdout instead of writing files",
     )
 
+    # Fingerprint comparison: srtctl diff <path_a> <path_b>
+    diff_parser = subparsers.add_parser("diff", help="Compare fingerprints from two runs")
+    diff_parser.add_argument("path_a", type=Path, help="First output dir or lockfile")
+    diff_parser.add_argument("path_b", type=Path, help="Second output dir or lockfile")
+    diff_parser.add_argument("--verbose", action="store_true", help="Show all package changes")
+
+    # Environment check: srtctl check <path>
+    check_parser = subparsers.add_parser("check", help="Check environment against a fingerprint")
+    check_parser.add_argument("path", type=Path, help="Lockfile or output dir to check against")
+    check_parser.add_argument("--json", action="store_true", dest="json_output", help="Output as JSON")
+
     args = parser.parse_args()
 
+    # Handle diff and check commands first (they don't use -f/config)
+    if args.command == "diff":
+        fps_a = load_lockfile_fingerprints(args.path_a)
+        fps_b = load_lockfile_fingerprints(args.path_b)
+        if fps_a is None or fps_b is None:
+            missing = []
+            if fps_a is None:
+                missing.append(str(args.path_a))
+            if fps_b is None:
+                missing.append(str(args.path_b))
+            console.print(f"[bold red]Could not load fingerprints from:[/] {', '.join(missing)}")
+            sys.exit(1)
+
+        # Diff each worker against its counterpart
+        all_workers = sorted(set(fps_a.keys()) | set(fps_b.keys()))
+        for worker in all_workers:
+            if worker not in fps_a:
+                console.print(f"\n[bold]{worker}:[/] only in {args.path_b}")
+                continue
+            if worker not in fps_b:
+                console.print(f"\n[bold]{worker}:[/] only in {args.path_a}")
+                continue
+            diff = diff_fingerprints(fps_a[worker], fps_b[worker])
+            console.print(f"\n[bold]{worker}:[/]")
+            console.print(format_diff(diff, verbose=args.verbose))
+        return
+
+    if args.command == "check":
+        import json as json_mod
+
+        fps = load_lockfile_fingerprints(args.path)
+        if fps is None:
+            console.print(f"[bold red]Could not load fingerprints from:[/] {args.path}")
+            sys.exit(1)
+
+        # Capture current environment once, reuse for all worker checks
+        current_fp = capture_fingerprint()
+        all_results = []
+        for worker in sorted(fps.keys()):
+            results = check_against_fingerprint(fps[worker], current_fp)
+            if results:
+                all_results.extend(results)
+                console.print(f"\n[bold]{worker}:[/]")
+                if args.json_output:
+                    console.print(
+                        json_mod.dumps(
+                            [{"field": r.field, "status": r.status.value, "message": r.message} for r in results],
+                            indent=2,
+                        )
+                    )
+                else:
+                    console.print(format_check_results(results))
+        if not all_results:
+            console.print(format_check_results([]))
+        sys.exit(1 if all_results else 0)
+
     # Parse config arg: supports path:selector format for overrides
     config_path, selector = parse_config_arg(args.config)