docs: align user guide intro with the README description

docs/user/backends.md

@@ -119,7 +119,14 @@ also accepts:
 | `job_name` | workflow name | `salloc --job-name`. Falls back to the backend name, then is set to the workflow name at resolve time. |
 | `offload_task_logs` | `true` | Have `srun` write each task's `<task>.log` on the compute side (via `--output`) instead of streaming every line through the driver. Auto-falls back to streaming on an interactive TTY / `--tui`. Also toggled by `--offload-task-logs` / `--no-offload-task-logs` or `SFLOW_OFFLOAD_TASK_LOGS`. |
 
-> `time` accepts either an `"HH:MM:SS"` string or an integer number of minutes.
+> `time` accepts either an `HH:MM:SS` walltime or an integer number of minutes.
+> Quoting is optional: `time: 10:00:00` and `time: "10:00:00"` are equivalent.
+> A bare integer (`time: 5400`) means **minutes**, matching `sbatch --time`.
+>
+> Earlier sflow versions required the quotes. Unquoted, YAML 1.1 read `10:00:00`
+> as the base-60 integer `36000`, which Slurm then interpreted as 36000 *minutes*
+> — a 10-hour request silently became 25 days. sflow now parses config scalars
+> with YAML 1.2 rules, so the unquoted form means what it looks like.
 
 ### Cluster-specific flags (`extra_args`)
 

docs/user/configuration.md

@@ -23,12 +23,15 @@ Looking for a quick lookup of all config fields? See the [Quick Reference](./qui
 
 ## version
 
-Currently supported:
+Optional. Omit it and sflow uses `"0.1"`, the only value that has ever existed:
 
 ```yaml
-version: "0.1"
+version: "0.1"   # optional
 ```
 
+If you do declare it, it must be `"0.1"` — any other value is rejected. When
+several files are merged, they must not declare conflicting versions.
+
 ## variables
 
 Variables can be written as a **dict** or a **list** (they are normalized internally).

docs/user/intro.md

@@ -3,22 +3,26 @@ title: Introduction
 sidebar_position: 1
 ---
 
-`sflow` is a **declarative workflow descriptor** that separates _what to deploy_ from _where to deploy it_.
+## What is sflow
+
+A **declarative workflow descriptor for massive GPU clusters** that separates _what to deploy_ from _where to deploy it_.
 
 :::tip Find the right feature
 Not sure where to start? Open the [Feature Map](/feature-map) to choose a goal, see which sflow features apply, and jump to the relevant docs. Building with an AI coding agent? See [Agent Skills](/docs/agents/intro).
 :::
 
-An application's deployment steps are usually logically the same regardless of the underlying infrastructure. Take [NVIDIA Dynamo](https://github.com/ai-dynamo/dynamo) as an example: you start etcd and NATS, launch a frontend server, spin up workers that register to the frontend, and the service is up. That logical flow never changes — but making it actually run on Slurm, Docker Compose, or Kubernetes requires a different set of infrastructure-specific scripts, resource management, and networking tweaks each time, and the effort must be repeated for every new platform.
-
-`sflow` is trying to eliminate this duplication. You describe the workflow once in a portable YAML format — tasks, dependencies, resources, and launch methods — and `sflow` delegates execution to the target infrastructure through swappable backends, leveraging each platform's native ecosystem rather than reimplementing it (e.g. Kubernetes, Helm charts, Argo Workflows).
+**One semantic across every platform — backend agnostic by design.** A deployment's logic never changes: start etcd and NATS, launch a frontend, spin up workers, run the benchmark. Only the infrastructure glue does — and today that glue is rewritten from scratch for every platform. `sflow` consolidates it into a single portable YAML: tasks, dependencies, resources, and launch methods. The **same `sflow.yaml` runs on Docker, Slurm, and Kubernetes** — swap the backend fragment, rewrite nothing. Backends delegate to each platform's native ecosystem (`srun`/MPI, `docker run`, pods and MPI jobs) rather than reimplementing it.
 
-Pluggable extensions such as probes and artifacts integrate naturally without coupling your workflow to any specific platform. Write one `sflow.yaml` and run it across environments with minimal changes.
+**Cluster-level orchestration at scale.** Topology-aware node and GPU placement, multi-node replicas and sweeps, readiness/failure probes, and batch submission — so one descriptor drives hundreds of GPUs instead of a pile of hand-written bash.
 
-The current focus is **Slurm**, which — unlike Kubernetes or Docker — lacks a built-in workflow orchestration layer, making multi-step deployments especially cumbersome. As of the v0.3.0 release, the Docker and Kubernetes backends ship as well, alongside `local` and `slurm`. Kubernetes support is new in v0.3.0 and has known limitations (interactive `sflow run` only, `monitor:` not yet supported, tested on bare-metal Kubernetes and GKE) — see [Backends](./backends.md#kubernetes-backend).
+All four backends ship today: `local`, `docker`, `slurm`, and `kubernetes` (`k8s` / `k8s_mpi`). It is light enough to write and debug a recipe on your laptop and submit that same file to the cluster.
 
 ![sflow TUI](/img/sflow_tui.gif)
 
+Define _what to run_ in a `sflow.yaml` — tasks, dependencies, how to launch each task, and required resources. `sflow` executes the DAG in order, collects logs, and organizes outputs into a consistent directory structure. Example of a dynamo PD disaggregation LLM inference service workflow:
+
+![Workflow DAG Example](/img/workflow-dag.png)
+
 ## Docs versions
 
 The docs site version selector intentionally shows only maintained documentation streams:

docs/user/quick-reference.md

@@ -11,7 +11,7 @@ For detailed explanations and examples, see [Configuration](./configuration.md).
 
 | Field | Required | Type | Default | Description |
 |-------|----------|------|---------|-------------|
-| `version` | Yes | string | — | Schema version. Must be `"0.1"`. |
+| `version` | No | string | `"0.1"` | Schema version. If declared, must be `"0.1"`. |
 | `variables` | | dict / list | — | Global variables available to expressions and task env. |
 | `artifacts` | | dict / list | — | Named resources referenced by URI. |
 | `backends` | | dict / list | — | Compute backends (`local`, `slurm`, `docker`, `kubernetes`). |

examples/self_contained/kubernetes/multinode.yaml

@@ -3,21 +3,90 @@ version: "0.1"
 # Multi-node distributed training on Kubernetes via sflow's default device-plugin
 # GPU scheduling (scheduling: device_plugin -> nvidia.com/gpu limits).
 #
-# The reservation reserves 4 GPU nodes; the single task is split into 4 pods (one
-# per reserved node, leader = index 0) each requesting 8 GPUs via the device-plugin
-# limit. sflow injects SFLOW_TASK_NODE_INDEX and SFLOW_LEADER_ADDRESS
-# per pod (plus the shared SFLOW_TASK_ASSIGNED_NODE_IPS) so torchrun can rendezvous.
+# The reservation reserves NUM_NODES GPU nodes; the single task is split into one
+# pod per reserved node (leader = index 0), each requesting GPUS_PER_NODE GPUs via
+# the device-plugin limit. sflow injects SFLOW_TASK_NODE_INDEX and
+# SFLOW_LEADER_ADDRESS per pod so torchrun can rendezvous.
+#
+# The topology lives in variables and NOTHING restates it. torchrun's --nnodes must
+# equal the pod count or the c10d rendezvous blocks forever waiting for ranks that
+# will never join -- a silent hang, not an error. Deriving both from one variable is
+# why that cannot drift.
 #
 # `scheduling: device_plugin` is sflow's default; swap to `scheduling: dra` (a work in
 # progress) on a cluster with nvidia-dra-driver-gpu to request GPUs via DRA ResourceClaims.
 
+variables:
+  NUM_NODES:
+    description: "Nodes to reserve == the number of pods == torchrun --nnodes."
+    type: integer
+    value: 2
+  GPUS_PER_NODE:
+    description: "GPUs per node == torchrun --nproc_per_node."
+    type: integer
+    value: 2
+  TOTAL_GPUS:
+    description: "GPUs across the whole task; sflow divides it per pod."
+    type: integer
+    value: ${{ variables.NUM_NODES * variables.GPUS_PER_NODE }}
+
+artifacts:
+  # The recipe used to invoke a bare `train.py` that existed nowhere -- the run only
+  # ever got that far after the rendezvous, so the missing file was masked by the
+  # hang. Ship the smallest thing that PROVES the point of the example: every rank
+  # joined and NCCL actually carried a collective between the nodes.
+  - name: TRAIN_SCRIPT
+    uri: file://train.py
+    content: |
+      import os
+      import socket
+      import torch
+      import torch.distributed as dist
+
+      host = socket.gethostname()
+      node = os.environ.get("SFLOW_TASK_NODE_INDEX", "?")
+      local_rank = int(os.environ["LOCAL_RANK"])
+
+      def log(msg):
+          # Rank-tagged and flushed: torchrun interleaves every rank onto one stream,
+          # and an unflushed print is lost entirely if a peer dies mid-collective.
+          print(f"[node {node} | rank {os.environ.get('RANK', '?')} | {host}] {msg}", flush=True)
+
+      log(f"starting: local_rank={local_rank} master={os.environ.get('MASTER_ADDR')}:"
+          f"{os.environ.get('MASTER_PORT')} world_size={os.environ.get('WORLD_SIZE')}")
+
+      # Everything before this line is local. If the run hangs HERE, the rendezvous
+      # never completed -- which is exactly the --nnodes mismatch this recipe hit.
+      log("entering rendezvous (init_process_group)...")
+      dist.init_process_group("nccl")
+      rank, world = dist.get_rank(), dist.get_world_size()
+      torch.cuda.set_device(local_rank)
+      log(f"rendezvous OK: rank {rank}/{world} on {torch.cuda.get_device_name(local_rank)}")
+
+      # Sum of all ranks -- wrong if any rank is missing or NCCL silently no-ops.
+      t = torch.full((1,), float(rank), device="cuda")
+      dist.all_reduce(t)
+      expected = world * (world - 1) / 2
+      log(f"all_reduce -> {t.item():.0f} (expected {expected:.0f})")
+      assert t.item() == expected, f"all_reduce={t.item()} expected={expected}"
+
+      dist.barrier()
+      if rank == 0:
+          hosts = [None] * world
+          dist.all_gather_object(hosts, host)
+          log(f"OK: {world} ranks across {len(set(hosts))} host(s) agreed: {sorted(set(hosts))}")
+      else:
+          dist.all_gather_object([None] * world, host)
+      dist.destroy_process_group()
+      log("done")
+
 backends:
   - name: k8s
     type: kubernetes
     default: true
     namespace: default
-    nodes: 4
-    gpus_per_node: 8
+    nodes: ${{ variables.NUM_NODES }}
+    gpus_per_node: ${{ variables.GPUS_PER_NODE }}
     host_network: true
     scheduling: device_plugin
 
@@ -33,22 +102,23 @@ workflow:
       operator: trainer
       resources:
         nodes:
-          count: 4
+          count: ${{ variables.NUM_NODES }}
         gpus:
-          count: 32   # 8 GPUs per pod across 4 nodes
+          count: ${{ variables.TOTAL_GPUS }}
       script:
         # SFLOW_LEADER_ADDRESS is the leader (node 0) IP; SFLOW_TASK_NODE_INDEX is
-        # 0 on the leader and 1..N-1 on workers. Keep NNODES in sync with nodes.count.
+        # 0 on the leader and 1..N-1 on workers.
         - export MASTER_ADDR="$SFLOW_LEADER_ADDRESS"
         - export MASTER_PORT=12345
-        - export NNODES=4
-        - export NPROC_PER_NODE=8
+        - export NNODES=${{ variables.NUM_NODES }}
+        - export NPROC_PER_NODE=${{ variables.GPUS_PER_NODE }}
         - export NODE_RANK="$SFLOW_TASK_NODE_INDEX"
+        - nvidia-smi
         - >
           torchrun
           --nnodes=$NNODES
           --nproc_per_node=$NPROC_PER_NODE
           --node_rank=$NODE_RANK
           --master_addr=$MASTER_ADDR
           --master_port=$MASTER_PORT
-          train.py
+          ${{ artifacts.TRAIN_SCRIPT.path }}

src/sflow/cli/batch.py

@@ -22,6 +22,7 @@
 import yaml as _yaml
 
 from sflow.app.sflow import SflowApp
+from sflow.config.loader import safe_load
 from sflow.cli import DOCS_URL, app
 from sflow.cli._args import (  # split_list_arg re-exported for back-compat
     EnableTaskMonitorOption,
@@ -279,10 +280,9 @@ def _resolve_sbatch_extra_args(
     domain_map: dict[str, list[Any]] = {}
     for cfg_path in config_files:
         try:
-            import yaml as _yaml
 
             with open(cfg_path) as fh:
-                data = _yaml.safe_load(fh)
+                data = safe_load(fh)
             if data:
                 var_map.update(_build_var_map(data))
                 domain_map.update(extract_domains_from_raw_config(data))
@@ -1803,15 +1803,14 @@ def _derive_backend_int(
     runs when the regex returns None, so currently-resolving configs are unchanged and
     partial fragments (which the pipeline can't validate) keep their regex result.
     """
-    import yaml as _yaml
 
     merged_var_map: dict[str, Any] = {}
     all_data: list[dict] = []
 
     for f in config_files:
         try:
             with open(f) as fh:
-                raw = _yaml.safe_load(fh)
+                raw = safe_load(fh)
             if isinstance(raw, dict):
                 all_data.append(raw)
                 merged_var_map.update(_build_var_map(raw))
@@ -2108,7 +2107,6 @@ def _scan_sflow_yamls(paths: list[Path]) -> list[Path]:
     """
     import glob as _glob
 
-    import yaml as _yaml
 
     candidates: list[Path] = []
     for p in paths:
@@ -2132,8 +2130,8 @@ def _scan_sflow_yamls(paths: list[Path]) -> list[Path]:
     for f in candidates:
         try:
             with open(f) as fh:
-                data = _yaml.safe_load(fh)
-            if isinstance(data, dict) and "version" in data:
+                data = safe_load(fh)
+            if isinstance(data, dict) and "workflow" in data:
                 valid.append(f.resolve())
         except Exception:
             continue
@@ -2216,10 +2214,9 @@ def _run_bulk_submit(
         # Warn about CLI variable overrides
         if cli_var_keys:
             try:
-                import yaml as _yaml
 
                 with open(yaml_file) as fh:
-                    data = _yaml.safe_load(fh)
+                    data = safe_load(fh)
                 config_var_names: set[str] = set()
                 raw_vars = data.get("variables") or []
                 if isinstance(raw_vars, dict):
@@ -2326,10 +2323,9 @@ def _run_bulk_submit(
                 # (loading the YAML) when the backend has no resolvable ``nodes`` field.
                 row_nodes = _derive_nodes([yaml_file], cli_overrides=cli_set_var)
                 if row_nodes is None:
-                    import yaml as _yaml
 
                     with open(yaml_file) as fh:
-                        data = _yaml.safe_load(fh)
+                        data = safe_load(fh)
                     row_nodes = _first_node_column_int(
                         _build_var_map(data, cli_overrides=cli_set_var)
                     )

src/sflow/cli/compose.py

@@ -18,6 +18,7 @@
     ConfigLoader,
     _normalize_script_plain_mappings,
     merge_config_dicts,
+    safe_load,
 )
 from sflow.logging import configure_logging, get_logger
 from sflow.resolution import ExpressionResolver, resolve_variables_inline
@@ -85,7 +86,7 @@ def _compose_files(
     config_dicts: List[Dict[str, Any]] = []
     for path in files:
         with open(path, "r") as f:
-            data = yaml.safe_load(f)
+            data = safe_load(f)
         if data is None:
             raise ValueError(f"Configuration file is empty: {path}")
         _normalize_script_plain_mappings(data)