Add shard-count suggestion and render knobs used by Claude skills

The skills claimed manifest analyze emitted `suggested_shard_count`, that
slurm render picked up a rebased manifest automatically, and that render
could be run without a local rclone.conf. None of those matched the CLI.
Make the CLI match the documented behavior:

* `xfer manifest analyze` now emits `suggested_shard_count`,
  `shard_count_reasoning`, and `shard_count_assumptions` based on a
  10 TiB per-shard cap, `4 * array_concurrency` slack, and an optional
  core budget. New `--assumed-*` / `--max-shard-bytes-tb` flags let the
  user sharpen the suggestion.
* `xfer slurm render` gains `--manifest`, so rebased manifests can be
  consumed without clobbering `run/manifest.jsonl`, and its
  `--rclone-config` no longer requires a local file (it warns instead).
* Skill docs updated to match, plus pytest coverage for the shard-count
  heuristic.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: fluidnumericsJoe

.claude/skills/xfer-manifest-analyze/SKILL.md

@@ -23,23 +23,39 @@ uv run xfer manifest analyze \
   --out run/analyze.json
 ```
 
-Optionally, if the user already has a preferred base set of rclone flags they want to layer on top, pass `--base-flags "<flags>"`.
+Optional flags that tune the **shard-count suggestion** (not the rclone flag suggestion):
+
+| Flag                            | Default | What it does                                                                |
+| ------------------------------- | ------- | --------------------------------------------------------------------------- |
+| `--assumed-cpus-per-task`       | `4`     | Cores each worker will request. Matches `xfer slurm render` default.        |
+| `--assumed-array-concurrency`   | `64`    | Expected Slurm array concurrency. Matches `xfer slurm render` default.      |
+| `--assumed-core-budget`         | unset   | Total cores the partition will make available (supply from `sinfo`).         |
+| `--max-shard-bytes-tb`          | `10`    | Per-shard byte cap. No single shard should carry more than this.             |
+| `--base-flags "<flags>"`        | —       | Prepend the user's preferred rclone flags to the suggested ones.             |
+
+If the user already knows the transfer cluster's available core budget, pass it — the shard-count suggestion will be sharper. Otherwise the default (concurrency + bytes-only) is fine.
 
 ## Step 3 — Report
 
 Read `run/analyze.json` and report to the user:
 
-1. **Dataset shape**: total object count, total bytes, median size, p90/p99 sizes, and the histogram bin counts (power-of-2 edges).
+1. **Dataset shape**: total object count, total bytes, median size, p10/p90 sizes, and the histogram bin counts (power-of-2 edges).
 2. **Profile classification**: which profile the analyzer picked (`small_files`, `large_files`, or `mixed`) and the reasoning (e.g., ">70% of objects are under 1 MiB").
-3. **Suggested rclone flags**: the concrete string to pass to `--rclone-flags` for render. Typical examples:
+3. **Suggested rclone flags** (`suggested_flags`): the concrete string to pass to `--rclone-flags` for render. Typical examples:
    - small_files → `--transfers 64 --checkers 128 --fast-list`
    - large_files → `--transfers 16 --checkers 32 --buffer-size 256M`
    - mixed       → `--transfers 32 --checkers 64 --fast-list`
-4. **Suggested shard count**: derive from total object count / target objects-per-shard (aim for ~10k–50k objects per shard for small-file workloads, smaller for large-file). Cap at what the chosen transfer cluster can reasonably host in a job array. If you don't yet know the transfer cluster, give a range and defer to `xfer-manifest-shard`.
+4. **Suggested shard count** (`suggested_shard_count`, plus `shard_count_reasoning` and `shard_count_assumptions`). The heuristic:
+   - If `total_bytes` is below the per-shard cap (default 10 TiB), **1 shard** — don't shard small datasets.
+   - Otherwise `ceil(total_bytes / cap)` shards, upper-bounded by `4 × array_concurrency` and (if a core budget was supplied) `core_budget // cpus_per_task`.
+
+   Quote `shard_count_reasoning` verbatim back to the user so they can see the trade-offs.
 
 ## Step 4 — Persist for downstream skills
 
-`run/analyze.json` is the source of truth for flag/shard decisions. `xfer-manifest-shard` and `xfer-slurm-render` both read it. Don't re-derive flags by hand in those skills — point at this file.
+`run/analyze.json` is the source of truth for flag/shard decisions. `xfer-manifest-shard` reads `suggested_shard_count` and `xfer-slurm-render` reads `suggested_flags` — point at this file, don't re-derive.
+
+If the user's plan changes (different transfer cluster, different concurrency cap), re-run `xfer manifest analyze` with updated `--assumed-*` flags before calling `xfer-manifest-shard`.
 
 ## After this skill
 

.claude/skills/xfer-manifest-rebase/SKILL.md

@@ -40,7 +40,9 @@ Always write to a new file (don't overwrite `manifest.jsonl`). Keeping the origi
 
 ## Step 3 — Re-shard
 
-Sharding is derived from the manifest, so **re-shard after rebasing**:
+Sharding is derived from the manifest, so **re-shard after rebasing**. The existing `run/shards/` directory contains pre-rebase paths and must be replaced.
+
+Confirm with the user before removing the old shards (`run/shards` is small but removing it is irreversible locally):
 
 ```bash
 rm -rf run/shards
@@ -50,16 +52,26 @@ uv run xfer manifest shard \
   --num-shards <same-N-as-before>
 ```
 
-(Or invoke `xfer-manifest-shard`.) Byte balance won't change meaningfully, but the shard files need to carry the rebased paths or workers will try to copy from the wrong URI.
+(Or invoke `xfer-manifest-shard` with the rebased manifest as input.) Byte balance won't change meaningfully, but the shard files need to carry the rebased paths or workers will try to copy from the wrong URI.
+
+## Step 4 — Point `xfer slurm render` at the rebased manifest
+
+`xfer slurm render` reads `source_root` and `dest_root` from a manifest file. By default it reads `<run_dir>/manifest.jsonl`, which is intentionally left at the pre-rebase vantage point as an audit record. Pass `--manifest` to read the rebased file instead:
 
-## Step 4 — Point downstream skills at the rebased manifest
+```bash
+uv run xfer slurm render \
+  --run-dir run \
+  --manifest run/manifest.rebased.jsonl \
+  ...
+```
 
-When you invoke `xfer-slurm-render` next, pass `--run-dir run` but ensure `config.resolved.json` references the rebased manifest. If the user plans a fresh `xfer slurm render`, that's automatic (render reads from `run/shards/`).
+Without `--manifest`, render would use the original roots and every array task would target the wrong URI.
 
 ## Safety
 
 - Never delete the original manifest — always keep `run/manifest.jsonl` as an audit trail alongside `run/manifest.rebased.jsonl`.
 - Rebase is a remap, not a content migration. It does not move data. It only relabels what each shard points to.
+- Confirm before `rm -rf run/shards` — the user may want to move the old shards aside rather than delete them.
 
 ## After this skill
 

.claude/skills/xfer-manifest-shard/SKILL.md

@@ -13,22 +13,21 @@ Runs **locally on the workstation**. Pure file processing. No Slurm/SSH needed.
 
 ## Step 1 — Read the analyze output
 
-Read `run/analyze.json` (from `xfer-manifest-analyze`). Use its `suggested_shard_count` / profile as the starting point. If analyze hasn't been run yet, invoke `xfer-manifest-analyze` first — don't guess shard counts from the raw manifest.
+Read `run/analyze.json` (from `xfer-manifest-analyze`) and use `suggested_shard_count` directly as the shard count. The analyzer already factors in the 10 TiB/shard cap, the expected array concurrency, and (if supplied) the core budget.
 
-## Step 2 — Reconcile shard count with cluster resources
+If `run/analyze.json` doesn't exist yet, invoke `xfer-manifest-analyze` first — don't guess shard counts from the raw manifest.
 
-The right shard count depends on **both** rclone settings (from analyze) **and** the transfer cluster's available resources. Ask the user:
+## Step 2 — Decide whether to override
 
-1. Which cluster will run the transfer? (Same as build host, or different?)
-2. What's the target array concurrency — how many shards should run at once? Typical range: 32–256, capped by the partition's `MaxArraySize` and the throughput both S3 endpoints can handle.
-3. What's the partition's per-node core/memory budget?
+Only override `suggested_shard_count` if one of the inputs that fed it has changed since analyze ran:
 
-Rule of thumb:
-- Total shards ≈ max(suggested_shard_count_from_analyze, 4 × array_concurrency). This gives the scheduler enough slack to keep the array fully packed even as slow shards trail.
-- For small-files profiles (heavy listing, light bytes), bias toward **more shards** of fewer objects each.
-- For large-files profiles (heavy bytes, few objects), bias toward **fewer shards** with more bytes each — byte-balancing matters more than object count.
+- The transfer cluster is different from what analyze assumed (different core budget).
+- The array concurrency cap is different from what analyze assumed (defaults: `--assumed-array-concurrency=64`).
+- The user wants a different per-shard byte cap (default 10 TiB).
 
-State your recommendation with the reasoning, then confirm before running.
+In that case, **re-run `xfer-manifest-analyze`** with updated `--assumed-*` flags rather than hand-picking a new number here. Sharing the reasoning/assumptions via `run/analyze.json` is how downstream skills stay coherent.
+
+Show the user `suggested_shard_count` alongside `shard_count_reasoning` from analyze, then confirm before running.
 
 ## Step 3 — Run shard
 

.claude/skills/xfer-slurm-render/SKILL.md

@@ -57,16 +57,19 @@ Collect from the user (with defaults from `run/analyze.json` and the chosen part
 | `--rclone-image`       | `rclone/rclone:latest`                                  |
 | `--rclone-config`      | absolute path to rclone.conf **on the transfer cluster's compute nodes** (see note below) |
 | `--rclone-flags`       | `suggested_flags` from `run/analyze.json`               |
-| `--max-attempts`       | 3 (default)                                             |
+| `--max-attempts`       | 5 (default)                                             |
 | `--sbatch-extras`      | site-specific `--account=...`, `--qos=...`, etc.        |
 | `--pyxis-extra`        | extra `srun --container-*` flags if site requires them  |
+| `--manifest`           | optional; path to a specific manifest (pass `run/manifest.rebased.jsonl` if the rebase skill ran) |
 
-The `--rclone-config` path is baked into `sbatch_array.sh` and resolved **on the transfer cluster at job time**, not on the workstation. It must:
+The `--rclone-config` path is baked into `sbatch_array.sh` and resolved **on the transfer cluster at job time**, not at render time. Render itself no longer requires the file to exist on the workstation — it only prints a warning if the local path is missing, since the actual consumer is the compute node. Still:
 
-- Be an absolute path valid on that cluster's compute nodes (home dirs and shared paths differ between sites).
-- Exist with `0600` permissions before the job starts.
+- The path must be an absolute path valid on the cluster's compute nodes.
+- It must exist with `0600` permissions **on the cluster** before the job starts.
 
-If the user doesn't already have the config deployed to this cluster at a known path, **stop and invoke `xfer-rclone-config`** to set it up and record the path. Don't guess the path — a wrong value here means every array task will fail identically at container start.
+If the user doesn't already have the config deployed to the cluster at a known path, invoke `xfer-rclone-config` to set it up and record the path. A wrong value here means every array task will fail identically at container start, so double-check the path.
+
+Use `--manifest` whenever the user ran `xfer-manifest-rebase`: render reads `source_root` / `dest_root` from a manifest file, and the default path (`<run_dir>/manifest.jsonl`) is intentionally left at the pre-rebase vantage point as an audit record. Passing `--manifest run/manifest.rebased.jsonl` is how render picks up the rebased roots.
 
 ## Step 4 — Render
 
@@ -83,9 +86,10 @@ uv run xfer slurm render \
   --rclone-image <image> \
   --rclone-config <path-on-cluster> \
   --rclone-flags "<flags-from-analyze>" \
-  --max-attempts 3 \
+  --max-attempts 5 \
   --sbatch-extras '<multi-line sbatch directives>' \
-  --pyxis-extra '<extra pyxis flags>'
+  --pyxis-extra '<extra pyxis flags>' \
+  [--manifest run/manifest.rebased.jsonl]    # only after rebase
 ```
 
 ## Step 5 — Verify the outputs

pyproject.toml

@@ -30,5 +30,9 @@ select = ["E", "F", "I", "B", "UP"]
 dev = [
     "black>=26.1.0",
     "pre-commit>=4.5.1",
+    "pytest>=8.0.0",
 ]
 
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+

src/xfer/cli.py

@@ -467,6 +467,25 @@ def manifest_analyze(
         "--retries 10 --low-level-retries 20 --stats 600s --progress",
         help="Base rclone flags to always include",
     ),
+    assumed_cpus_per_task: int = typer.Option(
+        4,
+        min=1,
+        help="Cores per shard worker (matches slurm_render default). Used only for shard-count suggestion.",
+    ),
+    assumed_array_concurrency: int = typer.Option(
+        64,
+        min=1,
+        help="Expected Slurm array concurrency (matches slurm_render default). Used only for shard-count suggestion.",
+    ),
+    assumed_core_budget: Optional[int] = typer.Option(
+        None,
+        help="Total cores the transfer cluster's partition will make available. Used only for shard-count suggestion. If omitted, the core constraint is skipped.",
+    ),
+    max_shard_bytes_tb: int = typer.Option(
+        10,
+        min=1,
+        help="Per-shard byte cap in TiB (no single shard should exceed this).",
+    ),
 ) -> None:
     """
     Analyze manifest file sizes and suggest optimal rclone flags.
@@ -481,6 +500,7 @@ def manifest_analyze(
         format_histogram_data,
         human_bytes,
         suggest_rclone_flags_from_sizes,
+        suggest_shard_count,
     )
 
     # Read manifest and extract sizes
@@ -505,6 +525,13 @@ def manifest_analyze(
         stats = compute_file_size_stats(sizes)
         suggestion = suggest_rclone_flags_from_sizes(sizes)
         histogram = format_histogram_data(sizes)
+        shard_suggestion = suggest_shard_count(
+            stats.total_bytes,
+            cpus_per_task=assumed_cpus_per_task,
+            array_concurrency=assumed_array_concurrency,
+            core_budget=assumed_core_budget,
+            max_shard_bytes_tb=max_shard_bytes_tb,
+        )
 
         # Combine base flags with profile-specific flags
         combined_flags = f"{suggestion.flags} {base_flags}"
@@ -528,6 +555,9 @@ def manifest_analyze(
             "profile": suggestion.profile,
             "profile_explanation": suggestion.explanation,
             "suggested_flags": combined_flags,
+            "suggested_shard_count": shard_suggestion.num_shards,
+            "shard_count_reasoning": shard_suggestion.reasoning,
+            "shard_count_assumptions": shard_suggestion.assumptions,
             "histogram": histogram,
         }
 
@@ -882,9 +912,7 @@ def slurm_render(
     rclone_image: str = typer.Option(..., help="Container image containing rclone"),
     rclone_config: Path = typer.Option(
         ...,
-        exists=True,
-        dir_okay=False,
-        help="Host path to rclone.conf",
+        help="Absolute path to rclone.conf on the transfer cluster's compute nodes. Not required to exist on this host; a warning is emitted if it is missing locally.",
         resolve_path=True,
     ),
     rclone_conf_in_container: str = typer.Option(
@@ -911,6 +939,11 @@ def slurm_render(
     pyxis_extra: str = typer.Option(
         "", help="Extra pyxis flags (string placed after --container-mounts...)"
     ),
+    manifest: Optional[Path] = typer.Option(
+        None,
+        help="Manifest JSONL to read source/dest_root from. Defaults to <run_dir>/manifest.jsonl. Use this after `xfer manifest rebase` to point render at the rebased file.",
+        resolve_path=True,
+    ),
 ) -> None:
     """
     Render worker.sh, sbatch_array.sh, and submit.sh under run_dir.
@@ -919,12 +952,19 @@ def slurm_render(
     mkdirp(run_dir / "logs")
     mkdirp(run_dir / "state")
 
-    # If source/dest not provided, try to read first line of manifest.jsonl (if present)
+    if not rclone_config.exists():
+        eprint(
+            f"WARNING: rclone_config {rclone_config} does not exist on this host. "
+            "That is fine if the path is valid on the transfer cluster's compute nodes. "
+            "Verify before submitting."
+        )
+
+    # If source/dest not provided, try to read first line of the manifest (if present)
     if source_root is None or dest_root is None:
-        manifest = run_dir / "manifest.jsonl"
-        if manifest.exists():
+        manifest_path = manifest or (run_dir / "manifest.jsonl")
+        if manifest_path.exists():
             first = None
-            for ln in manifest.read_text(encoding="utf-8").splitlines():
+            for ln in manifest_path.read_text(encoding="utf-8").splitlines():
                 if ln.strip():
                     first = json.loads(ln)
                     break
@@ -934,7 +974,7 @@ def slurm_render(
 
     if not source_root or not dest_root:
         raise typer.BadParameter(
-            "source_root/dest_root not set and could not be read from run_dir/manifest.jsonl"
+            "source_root/dest_root not set and could not be read from the manifest (pass --manifest, --source-root, and/or --dest-root explicitly)"
         )
 
     # Write scripts

src/xfer/est.py

@@ -569,6 +569,90 @@ def suggest_rclone_flags_from_sizes(sizes: List[int]) -> RcloneFlagsSuggestion:
     )
 
 
+@dataclass
+class ShardCountSuggestion:
+    """Suggested shard count for `xfer manifest shard` based on bytes, cores, and concurrency."""
+
+    num_shards: int
+    reasoning: str
+    assumptions: Dict[str, Any]
+
+
+def suggest_shard_count(
+    total_bytes: int,
+    *,
+    cpus_per_task: int = 4,
+    array_concurrency: int = 64,
+    core_budget: Optional[int] = None,
+    max_shard_bytes_tb: int = 10,
+) -> ShardCountSuggestion:
+    """
+    Suggest a shard count for a transfer based on three constraints:
+
+    1. Bytes cap: no single shard should carry more than ``max_shard_bytes_tb`` TiB
+       of data (worker wall-clock dominates the array's long tail otherwise).
+    2. Concurrency cap: producing more than ``4 * array_concurrency`` shards is
+       wasteful (the scheduler only needs enough slack to keep the queue packed
+       as slow shards trail).
+    3. Core cap (optional): if ``core_budget`` is supplied, the array can't
+       usefully exceed ``core_budget // cpus_per_task`` concurrent workers, so
+       producing more shards just lengthens the queue.
+
+    Special case: if ``total_bytes`` is below the per-shard cap, return
+    ``num_shards=1`` — sharding isn't helpful.
+    """
+    tib = 1024**4
+    max_shard_bytes = max_shard_bytes_tb * tib
+
+    assumptions: Dict[str, Any] = {
+        "total_bytes": total_bytes,
+        "cpus_per_task": cpus_per_task,
+        "array_concurrency": array_concurrency,
+        "core_budget": core_budget,
+        "max_shard_bytes_tb": max_shard_bytes_tb,
+    }
+
+    if total_bytes < max_shard_bytes:
+        reasoning = (
+            f"total_bytes ({human_bytes(total_bytes)}) is below the per-shard cap "
+            f"({max_shard_bytes_tb} TiB); a single shard is sufficient."
+        )
+        return ShardCountSuggestion(
+            num_shards=1, reasoning=reasoning, assumptions=assumptions
+        )
+
+    shards_by_bytes = math.ceil(total_bytes / max_shard_bytes)
+    shards_by_concurrency = 4 * array_concurrency
+    shards_by_cores: Optional[int] = None
+    if core_budget is not None:
+        shards_by_cores = max(1, core_budget // cpus_per_task)
+
+    upper = shards_by_concurrency
+    if shards_by_cores is not None:
+        upper = min(upper, shards_by_cores)
+
+    num_shards = max(1, min(upper, max(shards_by_bytes, 1)))
+
+    reasoning_parts = [
+        f"shards_by_bytes={shards_by_bytes} "
+        f"(total {human_bytes(total_bytes)} / {max_shard_bytes_tb} TiB cap)",
+        f"shards_by_concurrency={shards_by_concurrency} (4 x {array_concurrency})",
+    ]
+    if shards_by_cores is not None:
+        reasoning_parts.append(
+            f"shards_by_cores={shards_by_cores} "
+            f"({core_budget} cores / {cpus_per_task} cpus_per_task)"
+        )
+    reasoning_parts.append(
+        f"chose max(1, min(upper={upper}, shards_by_bytes)) = {num_shards}"
+    )
+    reasoning = "; ".join(reasoning_parts)
+
+    return ShardCountSuggestion(
+        num_shards=num_shards, reasoning=reasoning, assumptions=assumptions
+    )
+
+
 def format_histogram_data(
     sizes: List[int],
 ) -> List[Dict[str, Any]]: