From 1c4a6c0d7d0a98859cc4d9f5ed4bb6da0de02585 Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 04:53:32 +0000
Subject: [PATCH 1/9] fix(slurm): drop dead QUANTUI_RESULTS_DIR export from the
 sbatch template

The template exported QUANTUI_RESULTS_DIR=<staging>/results and created that
directory, but the batch worker never calls save_result() or list_results();
results reach History only when the app ingests the staging dir after the
job ends, using the app's own results dir. The export only left an empty
results/ folder in every staging dir. (M-JOBDIRS JD.9)

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/backends/cluster_config.py | 2 --
 quantui/backends/slurm.py          | 2 --
 2 files changed, 4 deletions(-)

diff --git a/quantui/backends/cluster_config.py b/quantui/backends/cluster_config.py
index 17ceaeb..fd7bfd6 100644
--- a/quantui/backends/cluster_config.py
+++ b/quantui/backends/cluster_config.py
@@ -150,8 +150,6 @@ def default_staging_root() -> Path:
 echo "Working directory: $(pwd)"
 
 export OMP_NUM_THREADS="${{SLURM_CPUS_PER_TASK:-{cores}}}"
-export QUANTUI_RESULTS_DIR="{results_dir}"
-mkdir -p "$QUANTUI_RESULTS_DIR"
 
 {worker_command}
 
diff --git a/quantui/backends/slurm.py b/quantui/backends/slurm.py
index 5333745..459d004 100644
--- a/quantui/backends/slurm.py
+++ b/quantui/backends/slurm.py
@@ -212,7 +212,6 @@ def _write_slurm_script(
 
         optional = "\n" + "\n".join(extra) if extra else ""
         worker_command = self._worker_command(request_path, staging_dir)
-        results_dir = staging_dir / "results"
         content = cfg.SLURM_SCRIPT_TEMPLATE.format(
             job_name=job_name,
             partition=self.partition,
@@ -222,7 +221,6 @@ def _write_slurm_script(
             output_file=str(staging_dir / "slurm-%j.out"),
             error_file=str(staging_dir / "slurm-%j.err"),
             optional_directives=optional,
-            results_dir=str(results_dir),
             worker_command=worker_command,
         )
         output_path.write_text(content)

From 64ad26b9af57c3ea398ccec3f5cc1e91cf865502 Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 04:58:47 +0000
Subject: [PATCH 2/9] feat(slurm): per-job dirs with attempt subdirs so reruns
 never overwrite

M-JOBDIRS JD.2/JD.3/JD.4 (backend)/JD.8.

- Each SLURM submission gets a readable job dir under the staging root,
  named <label>_<calc>_<method>_<basis> (or a sanitized --job-name), with
  _2, _3... for a new calculation that reuses a name. The same name is the
  SLURM --job-name (truncated to 40 chars) so squeue matches the dir.
- submit.slurm now creates its own attempt-NN_job<SLURM id>/ dir (numbered
  max+1 under a flock, mkdir without -p) and the worker writes into it via
  a new --attempt-dir flag. A hand-run `sbatch submit.slurm` therefore gets
  a fresh attempt and never touches an earlier attempt's files. SLURM's own
  slurm-%j.out/.err stay in the job dir because SLURM will not create a
  missing directory for them.
- JobRecord gains job_dir, attempts and ingested_attempts; staging_path
  resolves to the tracked job's attempt dir, so the monitor, ingest and
  terminal-state checks read the right attempt unchanged.
- SlurmBackend.resubmit() runs the job's submit.slurm again as a new
  attempt and points the record at the new SLURM job.
- The worker only updates the registry record when its SLURM_JOB_ID is the
  tracked one, so a hand-run attempt cannot flip the record's status.
- Apptainer runs bind the job dir when it lives outside $HOME.
- Legacy records (no job_dir) keep staging_root/<request_id>/ and behave as
  before.

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app_slurm.py               |   2 +-
 quantui/backends/cluster_config.py |  37 ++-
 quantui/backends/registry.py       | 144 +++++++++-
 quantui/backends/slurm.py          | 113 ++++++--
 quantui/backends/slurm_utils.py    |  45 +++
 quantui/backends/worker.py         |  54 +++-
 quantui/cli.py                     |   2 +-
 tests/test_backends_slurm.py       |  33 ++-
 tests/test_jobdirs.py              | 423 +++++++++++++++++++++++++++++
 9 files changed, 799 insertions(+), 54 deletions(-)
 create mode 100644 tests/test_jobdirs.py

diff --git a/quantui/app_slurm.py b/quantui/app_slurm.py
index c1294b9..ea996d8 100644
--- a/quantui/app_slurm.py
+++ b/quantui/app_slurm.py
@@ -194,7 +194,7 @@ def submit_slurm_run(app: Any) -> None:
     _append_run_stdout(
         app,
         f"\n📤 Submitted batch job {slurm_id} (request {request_id})\n"
-        f"Staging directory: {record.staging_dir if record else '?'}\n",
+        f"Job directory: {(record.job_dir or record.staging_dir) if record else '?'}\n",
     )
     threading.Thread(
         target=_monitor_slurm_job,
diff --git a/quantui/backends/cluster_config.py b/quantui/backends/cluster_config.py
index fd7bfd6..75777c9 100644
--- a/quantui/backends/cluster_config.py
+++ b/quantui/backends/cluster_config.py
@@ -9,6 +9,7 @@
 
 import logging
 import os
+import shlex
 from pathlib import Path
 
 logger = logging.getLogger(__name__)
@@ -130,6 +131,40 @@ def default_staging_root() -> Path:
     os.path.expanduser("~/quantui-gpu.sif"),
 )
 
+# Shell block that gives every run of submit.slurm its own attempt dir
+# (M-JOBDIRS). Numbering is max(existing NN) + 1 under a lock, and ``mkdir``
+# without ``-p`` makes the script fail rather than write into an existing
+# directory — an earlier attempt's files are never overwritten, whether the
+# job came from QuantUI's Resubmit or a hand-run ``sbatch submit.slurm``.
+# ``$JOB_DIR`` is set by the line build_attempt_setup() prepends.
+_ATTEMPT_SETUP_BODY = r"""exec 9>"$JOB_DIR/.attempt.lock"
+if command -v flock >/dev/null 2>&1; then flock 9; fi
+attempt_n=0
+for d in "$JOB_DIR"/attempt-*; do
+  [ -d "$d" ] || continue
+  k="${d##*/attempt-}"
+  k="${k%%_*}"
+  case "$k" in
+    ''|*[!0-9]*) continue ;;
+  esac
+  k=$((10#$k))
+  if [ "$k" -gt "$attempt_n" ]; then attempt_n=$k; fi
+done
+attempt_n=$((attempt_n + 1))
+ATTEMPT_DIR="$JOB_DIR/$(printf 'attempt-%02d_job%s' "$attempt_n" "${SLURM_JOB_ID:-manual$$}")"
+mkdir "$ATTEMPT_DIR"
+exec 9>&-
+ln -sfn "${ATTEMPT_DIR##*/}" "$JOB_DIR/latest" 2>/dev/null || true
+cd "$ATTEMPT_DIR"
+echo "Attempt directory: $ATTEMPT_DIR"
+"""
+
+
+def build_attempt_setup(job_dir: str) -> str:
+    """Return the attempt-dir shell block for a job dir (shell-quoted)."""
+    return f"\nJOB_DIR={shlex.quote(job_dir)}\n{_ATTEMPT_SETUP_BODY}"
+
+
 # SLURM batch script template. ``{worker_command}`` is the full command line
 # run inside the allocation (Apptainer-wrapped when configured).
 SLURM_SCRIPT_TEMPLATE = """#!/bin/bash
@@ -150,7 +185,7 @@ def default_staging_root() -> Path:
 echo "Working directory: $(pwd)"
 
 export OMP_NUM_THREADS="${{SLURM_CPUS_PER_TASK:-{cores}}}"
-
+{attempt_setup}
 {worker_command}
 
 echo "Job completed at: $(date -u +%Y-%m-%dT%H:%M:%SZ)"
diff --git a/quantui/backends/registry.py b/quantui/backends/registry.py
index 28285f6..f320d50 100644
--- a/quantui/backends/registry.py
+++ b/quantui/backends/registry.py
@@ -2,14 +2,24 @@
 On-disk job registry for execution backends (M-CLUSTER2 CL2.1).
 
 Each submitted calculation gets ``<jobs_root>/<request_id>.json`` plus a
-companion staging directory under ``staging_root/<request_id>/`` for live logs
-and progress files while the job runs.
+directory under ``staging_root`` for live logs, progress files and results.
+
+Two layouts exist:
+
+* **Job dir (M-JOBDIRS)** — ``staging_root/<job name>/`` holds ``request.json``
+  and ``submit.slurm``; every run of that script creates its own
+  ``attempt-NN_job<SLURM id>/`` subdirectory, so a retry never overwrites an
+  earlier attempt's files. :attr:`JobRecord.staging_path` resolves to the
+  attempt dir of the currently tracked SLURM job.
+* **Legacy** — ``staging_root/<request_id>/`` holds everything directly.
+  Records written before M-JOBDIRS have no ``job_dir`` and keep this behavior.
 """
 
 from __future__ import annotations
 
 import json
 import logging
+import re
 import time
 from dataclasses import asdict, dataclass, field
 from datetime import datetime, timezone
@@ -25,10 +35,21 @@
 _ACTIVE_STATUSES = frozenset({"queued", "pending", "running", "submitted"})
 
 
+_ATTEMPT_DIR_RE = re.compile(r"^attempt-(\d+)_job(.+)$")
+
+
 def _utc_now() -> str:
     return datetime.now(timezone.utc).replace(microsecond=0).isoformat()
 
 
+def parse_attempt_dir_name(name: str) -> Optional[tuple[int, str]]:
+    """Return ``(attempt number, SLURM job id)`` for an attempt dir name, else None."""
+    m = _ATTEMPT_DIR_RE.match(name)
+    if m is None:
+        return None
+    return int(m.group(1)), m.group(2)
+
+
 @dataclass
 class JobRecord:
     request_id: str
@@ -43,6 +64,15 @@ class JobRecord:
     result_dir: Optional[str] = None
     resources: Dict[str, Any] = field(default_factory=dict)
     error: Optional[Dict[str, Any]] = None
+    # M-JOBDIRS — None on legacy records (staging_dir holds everything).
+    job_dir: Optional[str] = None
+    # One entry per sbatch submission made through QuantUI:
+    # {"slurm_job_id", "submitted_at", "source"}. Hand-run sbatch attempts are
+    # not listed here; they are found on disk (see attempt_dirs()).
+    attempts: List[Dict[str, Any]] = field(default_factory=list)
+    # Names of attempt dirs (or, for legacy records, the staging dir) already
+    # saved to History, so a result is never ingested twice.
+    ingested_attempts: List[str] = field(default_factory=list)
 
     def to_dict(self) -> Dict[str, Any]:
         return asdict(self)
@@ -62,15 +92,54 @@ def from_dict(cls, data: Dict[str, Any]) -> JobRecord:
             result_dir=data.get("result_dir"),
             resources=dict(data.get("resources") or {}),
             error=data.get("error"),
+            job_dir=data.get("job_dir"),
+            attempts=list(data.get("attempts") or []),
+            ingested_attempts=list(data.get("ingested_attempts") or []),
         )
 
     @property
     def request_obj(self) -> CalculationRequest:
         return CalculationRequest.from_dict(self.request)
 
+    @property
+    def job_path(self) -> Optional[Path]:
+        return Path(self.job_dir) if self.job_dir else None
+
+    def attempt_dirs(self) -> List[Path]:
+        """Attempt dirs on disk, oldest first (empty for legacy records)."""
+        job_path = self.job_path
+        if job_path is None or not job_path.is_dir():
+            return []
+        found = []
+        for child in job_path.iterdir():
+            parsed = parse_attempt_dir_name(child.name)
+            if parsed is not None and child.is_dir() and not child.is_symlink():
+                found.append((parsed[0], child.name, child))
+        return [path for _n, _name, path in sorted(found)]
+
+    def attempt_dir_for(self, slurm_job_id: Optional[str]) -> Optional[Path]:
+        """The attempt dir created by SLURM job *slurm_job_id*, if it exists yet."""
+        if not slurm_job_id:
+            return None
+        for path in self.attempt_dirs():
+            parsed = parse_attempt_dir_name(path.name)
+            if parsed is not None and parsed[1] == str(slurm_job_id):
+                return path
+        return None
+
     @property
     def staging_path(self) -> Path:
-        return Path(self.staging_dir)
+        """Where the tracked run writes ``live.log`` / ``result.json``.
+
+        Legacy records: the staging dir. Job-dir records: the attempt dir of
+        the tracked SLURM job, or the job dir itself while that job has not
+        started yet (the attempt dir is created by the batch script).
+        """
+        job_path = self.job_path
+        if job_path is None:
+            return Path(self.staging_dir)
+        attempt = self.attempt_dir_for(self.slurm_job_id)
+        return attempt if attempt is not None else job_path
 
     @property
     def live_log_path(self) -> Path:
@@ -102,6 +171,22 @@ def staging_dir_for(self, request_id: str) -> Path:
         path.mkdir(parents=True, exist_ok=True)
         return path
 
+    def new_job_dir(self, name: str) -> Path:
+        """Create a fresh job dir named *name* (``_2``, ``_3``… on collision).
+
+        Never reuses an existing directory: each new calculation gets its own.
+        """
+        n = 1
+        while True:
+            candidate = name if n == 1 else f"{name}_{n}"
+            path = safe_join(self.staging_root, candidate)
+            try:
+                path.mkdir(parents=True, exist_ok=False)
+            except FileExistsError:
+                n += 1
+                continue
+            return path
+
     def create(
         self,
         request: CalculationRequest,
@@ -109,8 +194,19 @@ def create(
         *,
         resources: Optional[Dict[str, Any]] = None,
         status: str = "queued",
+        job_name: Optional[str] = None,
     ) -> JobRecord:
-        staging = self.staging_dir_for(request.request_id)
+        """Register *request*.
+
+        With *job_name* the record gets a job dir (M-JOBDIRS layout); without
+        it, a legacy ``staging_root/<request_id>/`` dir.
+        """
+        if job_name:
+            staging = self.new_job_dir(job_name)
+            job_dir: Optional[str] = str(staging)
+        else:
+            staging = self.staging_dir_for(request.request_id)
+            job_dir = None
         now = _utc_now()
         record = JobRecord(
             request_id=request.request_id,
@@ -122,6 +218,7 @@ def create(
             created_at=now,
             updated_at=now,
             resources=dict(resources or {}),
+            job_dir=job_dir,
         )
         self.save(record)
         return record
@@ -184,6 +281,45 @@ def update_status(
         self.save(record)
         return record
 
+    def start_attempt(
+        self, request_id: str, slurm_job_id: str, *, source: str = "submit"
+    ) -> Optional[JobRecord]:
+        """Point *request_id* at a newly submitted SLURM job.
+
+        Used for the first submission and for every Resubmit: the record
+        tracks the new job (status ``submitted``, previous error cleared) and
+        the submission is appended to :attr:`JobRecord.attempts`.
+        """
+        record = self.load(request_id)
+        if record is None:
+            return None
+        record.status = "submitted"
+        record.slurm_job_id = slurm_job_id
+        record.error = None
+        record.attempts.append(
+            {
+                "slurm_job_id": slurm_job_id,
+                "submitted_at": _utc_now(),
+                "source": source,
+            }
+        )
+        self.save(record)
+        return record
+
+    def mark_ingested(
+        self, request_id: str, attempt_key: str, *, result_dir: Optional[str] = None
+    ) -> Optional[JobRecord]:
+        """Record that *attempt_key* has been saved to History."""
+        record = self.load(request_id)
+        if record is None:
+            return None
+        if attempt_key not in record.ingested_attempts:
+            record.ingested_attempts.append(attempt_key)
+        if result_dir is not None:
+            record.result_dir = result_dir
+        self.save(record)
+        return record
+
     def _submit_meta_path(self) -> Path:
         return self.jobs_root / ".slurm_submit_meta.json"
 
diff --git a/quantui/backends/slurm.py b/quantui/backends/slurm.py
index 459d004..de6d5b8 100644
--- a/quantui/backends/slurm.py
+++ b/quantui/backends/slurm.py
@@ -32,11 +32,14 @@
 from .registry import JobRegistry
 from .slurm_errors import format_error_for_student
 from .slurm_utils import (
+    SLURM_JOB_NAME_MAX_LEN,
     SlurmJobAccounting,
+    default_job_name,
     estimate_slurm_resources,
     parse_sacct_accounting,
     parse_sacct_states,
     parse_slurm_job_id,
+    sanitize_job_name,
 )
 
 logger = logging.getLogger(__name__)
@@ -118,27 +121,27 @@ def dispatch(
         if email is not None:
             resolved_events = validate_mail_events(mail_events)
 
+        # M-JOBDIRS: the job dir name doubles as the SLURM job name, so
+        # squeue output matches the directory on disk.
+        name = sanitize_job_name(job_name or "") or default_job_name(request)
         record = self.registry.create(
             request,
             self.backend_id,
             resources=resources,
             status="queued",
+            job_name=name,
         )
-        staging = record.staging_path
-        request_path = staging / "request.json"
+        job_dir = record.job_path
+        assert job_dir is not None  # create(job_name=...) always sets it
+        request_path = job_dir / "request.json"
         request_path.write_text(json.dumps(request.to_dict(), indent=2))
 
-        label = (
-            job_name or f"{request.molecule.get('label', 'quantui')}_{request.method}"
-        )
-        label = "".join(c if c.isalnum() or c in "-_" else "_" for c in label)[:40]
-
-        slurm_script = staging / "submit.slurm"
+        slurm_script = job_dir / "submit.slurm"
         self._write_slurm_script(
             slurm_script,
-            job_name=label,
+            job_name=job_dir.name[:SLURM_JOB_NAME_MAX_LEN],
             request_path=request_path,
-            staging_dir=staging,
+            job_dir=job_dir,
             resources=resources,
             depends_on=depends_on,
             email=email,
@@ -161,15 +164,52 @@ def dispatch(
             )
             raise
 
-        self.registry.update_status(
-            request.request_id,
-            "submitted",
-            slurm_job_id=slurm_job_id,
-        )
+        self.registry.start_attempt(request.request_id, slurm_job_id, source="submit")
         self.registry.record_slurm_submit()
         return request.request_id
 
-    def _worker_command(self, request_path: Path, staging_dir: Path) -> str:
+    def resubmit(self, request_id: str) -> str:
+        """Run a finished job's ``submit.slurm`` again as a new attempt.
+
+        The new attempt gets its own ``attempt-NN_job<id>/`` dir inside the
+        same job dir (earlier attempts are untouched), and the record switches
+        to tracking the new SLURM job. Returns the new SLURM job id.
+
+        Raises ``ValueError`` when the job cannot be resubmitted (unknown,
+        still active, or a legacy record without a job dir) and
+        ``RuntimeError`` when ``sbatch`` fails.
+        """
+        record = self.registry.load(request_id)
+        if record is None:
+            raise ValueError(f"Job {request_id} is not in your registry.")
+        if record.status.lower() in _ACTIVE_RECORD:
+            raise ValueError(
+                f"Job {request_id} is still active ({record.status}). "
+                "Wait for it to finish or cancel it first."
+            )
+        job_dir = record.job_path
+        script = job_dir / "submit.slurm" if job_dir is not None else None
+        if script is None or not script.exists():
+            raise ValueError(
+                "This job was submitted before per-job folders existed, so it "
+                "cannot be resubmitted in place. Submit it again from the "
+                "Calculate tab."
+            )
+
+        active_slurm = sum(
+            1
+            for active in self.registry.list_active()
+            if active.backend_id == self.backend_id
+        )
+        check_concurrent_job_limit(active_slurm)
+        check_submit_cooldown(self.registry.seconds_since_last_slurm_submit())
+
+        slurm_job_id = self._submit_to_slurm(script)
+        self.registry.start_attempt(request_id, slurm_job_id, source="resubmit")
+        self.registry.record_slurm_submit()
+        return slurm_job_id
+
+    def _worker_command(self, request_path: Path, job_dir: Path) -> str:
         # AUDIT F21 — every path here is inserted into shell text (this
         # string is embedded verbatim into the generated sbatch script),
         # not passed as an argv list, so an unquoted path containing a
@@ -178,16 +218,25 @@ def _worker_command(self, request_path: Path, staging_dir: Path) -> str:
         # "--request /tmp/audit" plus a stray "folder/request.json"
         # argument. shlex.quote makes every one of these shell-safe
         # regardless of what it contains.
+        #
+        # "$ATTEMPT_DIR" is the one deliberate exception: it is a shell
+        # variable set by the attempt-setup block at run time, so it is
+        # double-quoted (expanded by the shell, never word-split) instead.
         py = shlex.quote(sys.executable)
         request_arg = shlex.quote(str(request_path))
-        inner = f"{py} -m quantui.backends.worker --request {request_arg}"
+        inner = (
+            f"{py} -m quantui.backends.worker --request {request_arg} "
+            '--attempt-dir "$ATTEMPT_DIR"'
+        )
         if self.use_apptainer:
             image = shlex.quote(self.apptainer_image)
-            staging_arg = shlex.quote(str(staging_dir))
-            return (
-                f'apptainer exec --nv --bind "$HOME:$HOME" --pwd {staging_arg} '
-                f"{image} {inner}"
-            )
+            binds = '--bind "$HOME:$HOME"'
+            # A job root outside $HOME (e.g. scratch) must be bound too, or
+            # the container cannot see request.json or the attempt dir.
+            if not _is_within(job_dir, Path.home()):
+                job_arg = shlex.quote(str(job_dir))
+                binds += f" --bind {job_arg}:{job_arg}"
+            return f'apptainer exec --nv {binds} --pwd "$ATTEMPT_DIR" {image} {inner}'
         return inner
 
     def _write_slurm_script(
@@ -196,7 +245,7 @@ def _write_slurm_script(
         *,
         job_name: str,
         request_path: Path,
-        staging_dir: Path,
+        job_dir: Path,
         resources: Dict[str, int | str],
         depends_on: str | None,
         email: str | None,
@@ -211,16 +260,20 @@ def _write_slurm_script(
             extra.append(f"#SBATCH --mail-type={events_str}")
 
         optional = "\n" + "\n".join(extra) if extra else ""
-        worker_command = self._worker_command(request_path, staging_dir)
+        worker_command = self._worker_command(request_path, job_dir)
         content = cfg.SLURM_SCRIPT_TEMPLATE.format(
             job_name=job_name,
             partition=self.partition,
             cores=resources["cores"],
             memory=resources["memory_gb"],
             walltime=resources["walltime"],
-            output_file=str(staging_dir / "slurm-%j.out"),
-            error_file=str(staging_dir / "slurm-%j.err"),
+            # SLURM opens these before the script runs and will not create
+            # a missing directory, so they live in the job dir (already
+            # unique per job via %j), not in the attempt dir.
+            output_file=str(job_dir / "slurm-%j.out"),
+            error_file=str(job_dir / "slurm-%j.err"),
             optional_directives=optional,
+            attempt_setup=cfg.build_attempt_setup(str(job_dir)),
             worker_command=worker_command,
         )
         output_path.write_text(content)
@@ -626,6 +679,14 @@ def _record_cancel_failure(self, record, *, code: str, message: str) -> None:
         )
 
 
+def _is_within(path: Path, parent: Path) -> bool:
+    try:
+        path.resolve().relative_to(parent.resolve())
+    except (ValueError, OSError):
+        return False
+    return True
+
+
 def _map_slurm_status(slurm_status: str) -> str:
     upper = slurm_status.upper()
     if upper in ("PENDING", "CONFIGURING"):
diff --git a/quantui/backends/slurm_utils.py b/quantui/backends/slurm_utils.py
index 834eccd..9dd5a92 100644
--- a/quantui/backends/slurm_utils.py
+++ b/quantui/backends/slurm_utils.py
@@ -21,6 +21,51 @@ class SlurmJobAccounting:
     elapsed: str = ""
 
 
+# Short calc-type tags for job dir names (M-JOBDIRS JD.2).
+_CALC_TYPE_TAGS = {
+    "single_point": "sp",
+    "geometry_opt": "opt",
+    "frequency": "freq",
+    "tddft": "tddft",
+    "nmr": "nmr",
+    "pes_scan": "scan",
+    "reorganization_energy": "reorg",
+}
+
+_JOB_NAME_MAX_LEN = 80
+# SLURM itself accepts longer names, but squeue's default format column is
+# narrow; the job dir keeps the full name.
+SLURM_JOB_NAME_MAX_LEN = 40
+
+
+def sanitize_job_name(name: str) -> str:
+    """Make *name* safe as a job dir name and SLURM ``--job-name``.
+
+    Characters other than letters, digits, ``-`` and ``_`` become ``_``;
+    leading/trailing separators are stripped. Returns ``""`` when nothing
+    usable is left.
+    """
+    cleaned = re.sub(r"[^A-Za-z0-9_-]+", "_", name.strip())
+    cleaned = re.sub(r"_+", "_", cleaned).strip("_-")
+    return cleaned[:_JOB_NAME_MAX_LEN].rstrip("_-")
+
+
+def default_job_name(request: CalculationRequest) -> str:
+    """``<label>_<calc tag>_<method>_<basis>``, e.g. ``H2O_opt_B3LYP_def2-SVP``.
+
+    Uses the same unsafe-character rule as History result dirs
+    (``results_storage._safe_name``: ``*`` → ``x``), so ``6-31G*`` stays
+    distinguishable from ``6-31G``.
+    """
+    from quantui.results_storage import _safe_name
+
+    label = str(request.molecule.get("label") or "quantui")
+    tag = _CALC_TYPE_TAGS.get(request.calc_type, request.calc_type)
+    parts = [label, tag, request.method, request.basis]
+    name = "_".join(_safe_name(str(p)) for p in parts if p)
+    return sanitize_job_name(name) or "quantui"
+
+
 def format_walltime(hours: float) -> str:
     total_seconds = int(hours * 3600)
     h = total_seconds // 3600
diff --git a/quantui/backends/worker.py b/quantui/backends/worker.py
index 239826b..bf8c6a0 100644
--- a/quantui/backends/worker.py
+++ b/quantui/backends/worker.py
@@ -24,6 +24,7 @@
 import argparse
 import json
 import logging
+import os
 import sys
 import time
 from pathlib import Path
@@ -282,6 +283,30 @@ def _maybe_run_preopt(
         return molecule, None
 
 
+def _update_tracked_record(request: CalculationRequest, status: str, **kwargs) -> None:
+    """Update the registry record, but only for the SLURM job it tracks.
+
+    A hand-run ``sbatch submit.slurm`` (M-JOBDIRS) runs under a SLURM job id
+    the record has never seen; letting it flip the record's status would
+    make the app ingest or report the wrong attempt. Those attempts are
+    picked up from disk instead. Outside SLURM (no ``SLURM_JOB_ID``) or
+    before the record has a job id, the update goes through as before.
+    """
+    registry = JobRegistry()
+    current_job = os.environ.get("SLURM_JOB_ID")
+    if current_job:
+        record = registry.load(request.request_id)
+        if record is not None and record.slurm_job_id not in (None, current_job):
+            logger.info(
+                "SLURM job %s is not the tracked attempt (%s); "
+                "leaving the registry record unchanged.",
+                current_job,
+                record.slurm_job_id,
+            )
+            return
+    registry.update_status(request.request_id, status, **kwargs)
+
+
 def _error_result(
     request: CalculationRequest,
     staging_dir: Path,
@@ -298,8 +323,7 @@ def _error_result(
         "technical_message": message,
         "retryable": retryable,
     }
-    registry = JobRegistry()
-    registry.update_status(request.request_id, "error", error=err)
+    _update_tracked_record(request, "error", error=err)
     _write_progress(staging_dir, "error", message, 100.0)
     return CalculationResult(
         request_id=request.request_id,
@@ -674,11 +698,17 @@ def _build_payload(
     raise ValueError(f"unsupported calc_type {calc_type!r}")
 
 
-def run_worker_request(request_path: Path) -> CalculationResult:
+def run_worker_request(
+    request_path: Path, attempt_dir: Path | None = None
+) -> CalculationResult:
+    """Run one batch request.
+
+    Outputs go to *attempt_dir* (M-JOBDIRS: one dir per run of
+    ``submit.slurm``), or, for legacy staging dirs, next to ``request.json``.
+    """
     data = json.loads(request_path.read_text(encoding="utf-8"))
     request = CalculationRequest.from_dict(data)
-    staging_dir = request_path.parent
-    registry = JobRegistry()
+    staging_dir = attempt_dir if attempt_dir is not None else request_path.parent
     calc_type = request.calc_type
 
     _append_log(staging_dir, f"Worker starting for {request.request_id} ({calc_type})")
@@ -753,7 +783,7 @@ def run_worker_request(request_path: Path) -> CalculationResult:
     payload = _build_payload(calc_type, outcome, staging_dir, request)
     write_worker_result(staging_dir, payload)
 
-    registry.update_status(request.request_id, "success", result_dir=str(staging_dir))
+    _update_tracked_record(request, "success", result_dir=str(staging_dir))
     _write_progress(staging_dir, "finalizing", "Calculation complete", 100.0)
 
     warnings: list[str] = []
@@ -785,10 +815,20 @@ def main(argv: list[str] | None = None) -> int:
         type=Path,
         help="Path to request.json inside the job staging directory",
     )
+    parser.add_argument(
+        "--attempt-dir",
+        type=Path,
+        default=None,
+        help=(
+            "Directory for this attempt's outputs (created by submit.slurm). "
+            "Defaults to the directory holding request.json."
+        ),
+    )
     args = parser.parse_args(argv)
 
     logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
-    outcome = run_worker_request(args.request.expanduser().resolve())
+    attempt_dir = args.attempt_dir.expanduser().resolve() if args.attempt_dir else None
+    outcome = run_worker_request(args.request.expanduser().resolve(), attempt_dir)
     return 0 if outcome.status == "success" else 1
 
 
diff --git a/quantui/cli.py b/quantui/cli.py
index a9dffad..747f6ca 100644
--- a/quantui/cli.py
+++ b/quantui/cli.py
@@ -373,7 +373,7 @@ def _cmd_submit(args: argparse.Namespace) -> int:
         submitted_count += 1
         record = backend.registry.load(request_id)
         slurm_job_id = record.slurm_job_id if record else None
-        staging = record.staging_path if record else "?"
+        staging = (record.job_dir or record.staging_dir) if record else "?"
         print(
             f"{request_path}: submitted request_id={request_id} "
             f"slurm_job_id={slurm_job_id or '?'} staging={staging}"
diff --git a/tests/test_backends_slurm.py b/tests/test_backends_slurm.py
index 26c2f36..60583df 100644
--- a/tests/test_backends_slurm.py
+++ b/tests/test_backends_slurm.py
@@ -97,9 +97,10 @@ def test_dispatch_writes_registry_and_scripts(
         assert record is not None
         assert record.slurm_job_id == "555666"
         assert record.status == "submitted"
-        assert (record.staging_path / "request.json").exists()
-        assert (record.staging_path / "submit.slurm").exists()
-        slurm_text = (record.staging_path / "submit.slurm").read_text()
+        assert record.job_path is not None
+        assert (record.job_path / "request.json").exists()
+        assert (record.job_path / "submit.slurm").exists()
+        slurm_text = (record.job_path / "submit.slurm").read_text()
         assert "#SBATCH" in slurm_text
         assert "quantui.backends.worker" in slurm_text
 
@@ -176,14 +177,16 @@ class TestSlurmBackendWorkerCommandQuoting:
 
     def test_request_path_with_space_stays_one_argument(self, slurm_backend):
         request_path = Path("/tmp/audit folder/request.json")
-        staging_dir = Path("/tmp/audit folder/staging")
-        cmd = slurm_backend._worker_command(request_path, staging_dir)
+        job_dir = Path("/tmp/audit folder/job")
+        cmd = slurm_backend._worker_command(request_path, job_dir)
         tokens = shlex.split(cmd)
-        assert tokens[-2] == "--request"
-        assert tokens[-1] == str(request_path)
-        assert (
-            len(tokens) == tokens.index("--request") + 2
-        ), f"request path split into extra shell tokens: {tokens}"
+        idx = tokens.index("--request")
+        assert tokens[idx + 1] == str(request_path)
+        # The attempt dir is a shell variable expanded at run time (M-JOBDIRS).
+        assert tokens[idx + 2 :] == [
+            "--attempt-dir",
+            "$ATTEMPT_DIR",
+        ], f"request path split into extra shell tokens: {tokens}"
 
     def test_apptainer_branch_quotes_staging_and_request_paths(self, tmp_path):
         registry = JobRegistry(
@@ -196,12 +199,14 @@ def test_apptainer_branch_quotes_staging_and_request_paths(self, tmp_path):
             apptainer_image="/opt/images/quantui image.sif",
         )
         request_path = Path("/tmp/audit folder/request.json")
-        staging_dir = Path("/tmp/audit folder/staging")
-        cmd = backend._worker_command(request_path, staging_dir)
+        job_dir = Path("/tmp/audit folder/job")
+        cmd = backend._worker_command(request_path, job_dir)
         tokens = shlex.split(cmd)
-        assert str(staging_dir) in tokens
+        # A job dir outside $HOME is bound into the container as one token.
+        assert f"{job_dir}:{job_dir}" in tokens
+        assert tokens[tokens.index("--pwd") + 1] == "$ATTEMPT_DIR"
         assert "/opt/images/quantui image.sif" in tokens
-        assert tokens[-1] == str(request_path)
+        assert tokens[tokens.index("--request") + 1] == str(request_path)
 
     def test_dispatch_with_space_in_staging_path_produces_parseable_script(
         self, mock_slurm_env, tmp_path
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
new file mode 100644
index 0000000..ebe4793
--- /dev/null
+++ b/tests/test_jobdirs.py
@@ -0,0 +1,423 @@
+"""
+M-JOBDIRS — SLURM job dirs with per-attempt subdirs.
+
+Covers the job dir layout (naming, collisions, ``--job-name``), the attempt
+setup block inside ``submit.slurm`` (run through a real ``bash``), Resubmit,
+the worker's ``--attempt-dir`` / tracked-record guard, and legacy records.
+"""
+
+import json
+import os
+import shlex
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+import pytest
+
+from quantui.backends import cluster_config as cfg
+from quantui.backends.base import CalculationRequest
+from quantui.backends.registry import JobRegistry, parse_attempt_dir_name
+from quantui.backends.slurm import SlurmBackend
+from quantui.backends.slurm_utils import default_job_name, sanitize_job_name
+
+needs_bash = pytest.mark.skipif(
+    sys.platform == "win32" or shutil.which("bash") is None,
+    reason="submit.slurm is a bash script",
+)
+
+
+def _request(rid: str = "jd001", **overrides) -> CalculationRequest:
+    fields = dict(
+        request_id=rid,
+        calc_type="geometry_opt",
+        method="B3LYP",
+        basis="def2-SVP",
+        charge=0,
+        multiplicity=1,
+        molecule={
+            "atoms": ["H", "H"],
+            "coords": [[0, 0, 0], [0, 0, 0.74]],
+            "label": "H2",
+        },
+    )
+    fields.update(overrides)
+    return CalculationRequest(**fields)
+
+
+@pytest.fixture
+def registry(tmp_path):
+    return JobRegistry(jobs_root=tmp_path / "jobs", staging_root=tmp_path / "staging")
+
+
+@pytest.fixture
+def backend(registry):
+    return SlurmBackend(registry=registry, partition="test", use_apptainer=False)
+
+
+def _sbatch_returning(*job_ids):
+    """Patch subprocess.run so successive sbatch calls return *job_ids*."""
+    outputs = iter(job_ids)
+
+    def _run(args, **kwargs):
+        result = subprocess.CompletedProcess(args, 0)
+        result.stdout = f"Submitted batch job {next(outputs)}\n"
+        result.stderr = ""
+        return result
+
+    return patch("quantui.backends.slurm.subprocess.run", side_effect=_run)
+
+
+def _run_attempt_setup(job_dir: Path, slurm_job_id: str) -> subprocess.CompletedProcess:
+    script = "set -euo pipefail\n" + cfg.build_attempt_setup(str(job_dir))
+    env = {**os.environ, "SLURM_JOB_ID": slurm_job_id}
+    return subprocess.run(
+        ["bash", "-c", script], env=env, capture_output=True, text=True
+    )
+
+
+# ---------------------------------------------------------------------------
+# Naming (JD.2)
+# ---------------------------------------------------------------------------
+
+
+class TestJobNames:
+    def test_default_name_is_label_calc_method_basis(self):
+        assert default_job_name(_request()) == "H2_opt_B3LYP_def2-SVP"
+
+    def test_default_name_keeps_star_basis_distinct(self):
+        name = default_job_name(_request(calc_type="single_point", basis="6-31G*"))
+        assert name == "H2_sp_B3LYP_6-31Gx"
+
+    def test_missing_label_falls_back(self):
+        req = _request(molecule={"atoms": ["H"], "coords": [[0, 0, 0]]})
+        assert default_job_name(req).startswith("quantui_opt_")
+
+    @pytest.mark.parametrize(
+        "raw, expected",
+        [
+            ("my water run", "my_water_run"),
+            ("../../etc/passwd", "etc_passwd"),
+            ("  __weird!!name__ ", "weird_name"),
+            ("***", ""),
+        ],
+    )
+    def test_sanitize(self, raw, expected):
+        assert sanitize_job_name(raw) == expected
+
+    def test_sanitize_caps_length(self):
+        assert len(sanitize_job_name("x" * 500)) <= 80
+
+
+# ---------------------------------------------------------------------------
+# Registry layout (JD.2 / JD.8)
+# ---------------------------------------------------------------------------
+
+
+class TestRegistryLayout:
+    def test_job_dir_collision_gets_suffix(self, registry):
+        a = registry.create(_request("a"), "cluster_slurm", job_name="run")
+        b = registry.create(_request("b"), "cluster_slurm", job_name="run")
+        assert Path(a.job_dir).name == "run"
+        assert Path(b.job_dir).name == "run_2"
+
+    def test_staging_path_follows_tracked_attempt(self, registry):
+        rec = registry.create(_request(), "cluster_slurm", job_name="run")
+        job = Path(rec.job_dir)
+        rec = registry.start_attempt(rec.request_id, "101")
+        # Queued: the attempt dir does not exist yet.
+        assert rec.staging_path == job
+        (job / "attempt-01_job101").mkdir()
+        (job / "attempt-02_job102").mkdir()
+        assert rec.staging_path == job / "attempt-01_job101"
+        rec = registry.start_attempt(rec.request_id, "102", source="resubmit")
+        assert rec.staging_path == job / "attempt-02_job102"
+        assert rec.live_log_path == job / "attempt-02_job102" / "live.log"
+        assert [a["source"] for a in rec.attempts] == ["submit", "resubmit"]
+
+    def test_attempt_dirs_sorted_numerically_and_skip_latest_link(self, registry):
+        rec = registry.create(_request(), "cluster_slurm", job_name="run")
+        job = Path(rec.job_dir)
+        for name in ("attempt-10_job9", "attempt-02_job5", "notes"):
+            (job / name).mkdir()
+        if sys.platform != "win32":
+            (job / "latest").symlink_to("attempt-10_job9")
+        assert [p.name for p in rec.attempt_dirs()] == [
+            "attempt-02_job5",
+            "attempt-10_job9",
+        ]
+
+    def test_parse_attempt_dir_name(self):
+        assert parse_attempt_dir_name("attempt-03_job812345") == (3, "812345")
+        assert parse_attempt_dir_name("latest") is None
+
+    def test_legacy_record_unchanged(self, registry):
+        """Records without job_name keep staging_root/<request_id>/ (JD.8)."""
+        rec = registry.create(_request("legacy1"), "cluster_slurm")
+        assert rec.job_dir is None
+        assert rec.staging_path == registry.staging_root / "legacy1"
+        assert rec.attempt_dirs() == []
+
+    def test_old_record_json_loads(self, registry):
+        """A record file written before M-JOBDIRS has none of the new keys."""
+        rec = registry.create(_request("old1"), "cluster_slurm")
+        data = rec.to_dict()
+        for key in ("job_dir", "attempts", "ingested_attempts"):
+            data.pop(key)
+        (registry.jobs_root / "old1.json").write_text(json.dumps(data))
+        loaded = registry.load("old1")
+        assert loaded.job_dir is None
+        assert loaded.attempts == []
+        assert loaded.ingested_attempts == []
+
+
+# ---------------------------------------------------------------------------
+# Dispatch (JD.2 / JD.3)
+# ---------------------------------------------------------------------------
+
+
+class TestDispatch:
+    def test_dispatch_creates_named_job_dir(self, backend):
+        with _sbatch_returning("5001"):
+            rid = backend.dispatch(_request())
+        rec = backend.registry.load(rid)
+        job = Path(rec.job_dir)
+        assert job.name == "H2_opt_B3LYP_def2-SVP"
+        assert (job / "request.json").exists()
+        script = (job / "submit.slurm").read_text()
+        assert "#SBATCH --job-name=H2_opt_B3LYP_def2-SVP" in script
+        assert f'#SBATCH --output="{job / "slurm-%j.out"}"' in script
+        assert "attempt-%02d_job%s" in script
+        assert '--attempt-dir "$ATTEMPT_DIR"' in script
+        assert rec.slurm_job_id == "5001"
+        assert rec.attempts[0]["slurm_job_id"] == "5001"
+
+    def test_same_config_twice_gets_separate_dirs(self, backend, monkeypatch):
+        monkeypatch.setattr(
+            "quantui.backends.slurm.check_submit_cooldown", lambda _s: None
+        )
+        with _sbatch_returning("1", "2"):
+            a = backend.dispatch(_request("a"))
+            b = backend.dispatch(_request("b"))
+        names = [Path(backend.registry.load(r).job_dir).name for r in (a, b)]
+        assert names == ["H2_opt_B3LYP_def2-SVP", "H2_opt_B3LYP_def2-SVP_2"]
+
+    def test_job_name_override_is_sanitized_and_truncated_for_slurm(self, backend):
+        long_name = "my very long descriptive job name " * 3
+        with _sbatch_returning("7"):
+            rid = backend.dispatch(_request(), job_name=long_name)
+        job = Path(backend.registry.load(rid).job_dir)
+        assert job.name == sanitize_job_name(long_name)
+        script = (job / "submit.slurm").read_text()
+        assert f"#SBATCH --job-name={job.name[:40]}\n" in script
+
+    def test_apptainer_binds_job_dir_outside_home(self, registry, monkeypatch):
+        monkeypatch.setattr(Path, "home", classmethod(lambda cls: Path("/nonexistent")))
+        backend = SlurmBackend(
+            registry=registry,
+            partition="test",
+            use_apptainer=True,
+            apptainer_image="/img.sif",
+        )
+        job = registry.staging_root / "x"
+        tokens = shlex.split(backend._worker_command(job / "request.json", job))
+        assert f"{job}:{job}" in tokens
+
+    def test_apptainer_skips_extra_bind_under_home(self, registry, monkeypatch):
+        monkeypatch.setattr(
+            Path, "home", classmethod(lambda cls: registry.staging_root.parent)
+        )
+        backend = SlurmBackend(
+            registry=registry,
+            partition="test",
+            use_apptainer=True,
+            apptainer_image="/img.sif",
+        )
+        job = registry.staging_root / "x"
+        cmd = backend._worker_command(job / "request.json", job)
+        assert cmd.count("--bind") == 1
+
+
+# ---------------------------------------------------------------------------
+# Attempt setup block (JD.3) — real bash
+# ---------------------------------------------------------------------------
+
+
+@needs_bash
+class TestAttemptSetupScript:
+    def test_each_run_gets_a_new_numbered_attempt(self, tmp_path):
+        job = tmp_path / "job dir with space"
+        job.mkdir()
+        assert _run_attempt_setup(job, "101").returncode == 0
+        (job / "attempt-01_job101" / "result.json").write_text("first")
+        assert _run_attempt_setup(job, "102").returncode == 0
+        assert sorted(p.name for p in job.iterdir() if p.name.startswith("att")) == [
+            "attempt-01_job101",
+            "attempt-02_job102",
+        ]
+        assert (job / "attempt-01_job101" / "result.json").read_text() == "first"
+        assert os.readlink(job / "latest") == "attempt-02_job102"
+
+    def test_numbering_continues_after_gaps(self, tmp_path):
+        job = tmp_path / "job"
+        (job / "attempt-05_job9").mkdir(parents=True)
+        assert _run_attempt_setup(job, "10").returncode == 0
+        assert (job / "attempt-06_job10").is_dir()
+
+    def test_mkdir_clash_aborts(self, tmp_path):
+        job = tmp_path / "job"
+        job.mkdir()
+        script = (
+            "set -euo pipefail\n"
+            + cfg.build_attempt_setup(str(job)).replace(
+                'mkdir "$ATTEMPT_DIR"', 'mkdir "$ATTEMPT_DIR"; mkdir "$ATTEMPT_DIR"'
+            )
+            + "echo SHOULD_NOT_RUN\n"
+        )
+        result = subprocess.run(
+            ["bash", "-c", script],
+            env={**os.environ, "SLURM_JOB_ID": "1"},
+            capture_output=True,
+            text=True,
+        )
+        assert result.returncode != 0
+        assert "SHOULD_NOT_RUN" not in result.stdout
+
+
+# ---------------------------------------------------------------------------
+# Resubmit (JD.4)
+# ---------------------------------------------------------------------------
+
+
+class TestResubmit:
+    def test_resubmit_tracks_new_job_and_clears_error(self, backend, monkeypatch):
+        monkeypatch.setattr(
+            "quantui.backends.slurm.check_submit_cooldown", lambda _s: None
+        )
+        with _sbatch_returning("100", "200"):
+            rid = backend.dispatch(_request())
+            backend.registry.update_status(
+                rid, "error", error={"code": "SLURM_TERMINAL", "user_message": "x"}
+            )
+            new_id = backend.resubmit(rid)
+        rec = backend.registry.load(rid)
+        assert new_id == "200"
+        assert rec.slurm_job_id == "200"
+        assert rec.status == "submitted"
+        assert rec.error is None
+        assert [a["slurm_job_id"] for a in rec.attempts] == ["100", "200"]
+        assert rec.attempts[1]["source"] == "resubmit"
+
+    def test_resubmit_rejects_active_job(self, backend):
+        with _sbatch_returning("100"):
+            rid = backend.dispatch(_request())
+        with pytest.raises(ValueError, match="still active"):
+            backend.resubmit(rid)
+
+    def test_resubmit_rejects_legacy_record(self, backend):
+        backend.registry.create(_request("legacy"), "cluster_slurm", status="error")
+        with pytest.raises(ValueError, match="before per-job folders"):
+            backend.resubmit("legacy")
+
+    def test_resubmit_unknown_job(self, backend):
+        with pytest.raises(ValueError, match="not in your registry"):
+            backend.resubmit("nope")
+
+
+# ---------------------------------------------------------------------------
+# Worker (JD.3 / JD.11 guard)
+# ---------------------------------------------------------------------------
+
+
+class TestWorkerAttemptDir:
+    @pytest.fixture
+    def job(self, tmp_path, monkeypatch):
+        monkeypatch.setenv("QUANTUI_JOBS_DIR", str(tmp_path / "jobs"))
+        monkeypatch.setenv("QUANTUI_STAGING_DIR", str(tmp_path / "staging"))
+        registry = JobRegistry()
+        rec = registry.create(
+            _request("w1", calc_type="not_a_calc"), "cluster_slurm", job_name="w"
+        )
+        job = Path(rec.job_dir)
+        (job / "request.json").write_text(json.dumps(rec.request))
+        registry.start_attempt("w1", "300")
+        return registry, job
+
+    def test_outputs_go_to_attempt_dir(self, job):
+        from quantui.backends.worker import run_worker_request
+
+        _registry, job_dir = job
+        attempt = job_dir / "attempt-01_job300"
+        attempt.mkdir()
+        run_worker_request(job_dir / "request.json", attempt)
+        assert (attempt / "live.log").exists()
+        assert (attempt / "progress.json").exists()
+        assert not (job_dir / "live.log").exists()
+
+    def test_tracked_job_updates_record(self, job, monkeypatch):
+        from quantui.backends.worker import run_worker_request
+
+        registry, job_dir = job
+        monkeypatch.setenv("SLURM_JOB_ID", "300")
+        attempt = job_dir / "attempt-01_job300"
+        attempt.mkdir()
+        run_worker_request(job_dir / "request.json", attempt)
+        assert registry.load("w1").status == "error"
+
+    def test_hand_run_job_leaves_record_alone(self, job, monkeypatch):
+        from quantui.backends.worker import run_worker_request
+
+        registry, job_dir = job
+        monkeypatch.setenv("SLURM_JOB_ID", "999")
+        attempt = job_dir / "attempt-02_job999"
+        attempt.mkdir()
+        run_worker_request(job_dir / "request.json", attempt)
+        rec = registry.load("w1")
+        assert rec.status == "submitted"
+        assert rec.slurm_job_id == "300"
+
+
+# ---------------------------------------------------------------------------
+# End to end: the real submit.slurm, run twice by hand
+# ---------------------------------------------------------------------------
+
+
+@needs_bash
+def test_submit_script_rerun_by_hand_keeps_both_attempts(tmp_path, monkeypatch):
+    """Generate a real submit.slurm and run it twice with bash, as a user
+    would with ``sbatch submit.slurm``. The first (failed) attempt's files
+    must survive the second run untouched."""
+    monkeypatch.setenv("QUANTUI_JOBS_DIR", str(tmp_path / "jobs"))
+    monkeypatch.setenv("QUANTUI_STAGING_DIR", str(tmp_path / "staging"))
+    backend = SlurmBackend(registry=JobRegistry(), partition="t", use_apptainer=False)
+    # An unsupported calc type fails fast inside the worker, no PySCF needed.
+    with _sbatch_returning("41"):
+        rid = backend.dispatch(_request("e2e", calc_type="not_a_calc"))
+    job = Path(backend.registry.load(rid).job_dir)
+    script = job / "submit.slurm"
+
+    env = {**os.environ, "PYTHONPATH": str(Path(__file__).resolve().parents[1])}
+    first = subprocess.run(
+        ["bash", str(script)],
+        env={**env, "SLURM_JOB_ID": "41"},
+        capture_output=True,
+        text=True,
+        cwd=tmp_path,
+    )
+    assert first.returncode != 0  # worker reports the unsupported calc type
+    first_log = (job / "attempt-01_job41" / "live.log").read_text()
+
+    second = subprocess.run(
+        ["bash", str(script)],
+        env={**env, "SLURM_JOB_ID": "42"},
+        capture_output=True,
+        text=True,
+        cwd=tmp_path,
+    )
+    assert second.returncode != 0
+    assert (job / "attempt-02_job42" / "live.log").exists()
+    assert (job / "attempt-01_job41" / "live.log").read_text() == first_log
+    # The hand-run job (42) is not the tracked one (41): record untouched by it.
+    assert backend.registry.load(rid).slurm_job_id == "41"

From 7c5a57d008f8ca9f0e66d388dce14e8a003956ef Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:02:07 +0000
Subject: [PATCH 3/9] feat(history): mark SLURM results in History with job id
 and attempt
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

M-JOBDIRS JD.5/JD.6.

- Ingest now passes provenance extras to save_result on all three paths
  (generic, frequency, reorganization energy): execution_backend="slurm"
  plus slurm.{job_id, attempt, job_dir, attempt_dir, request_id}. For a
  job-dir record the attempt number and job id come from the attempt dir
  name, so a hand-run attempt is labelled with the job that produced it.
- History list labels get a "🖥 SLURM <job>·a<attempt>" marker, next to
  the existing calibration marker.
- The saved-result card gets a "Ran on" row: SLURM batch, job id, attempt,
  and the attempt (or job) folder path.
- Additive: no schema bump. Local results and results ingested before this
  change have no provenance keys and render exactly as before.

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app_formatters.py        | 45 ++++++++++++++++
 quantui/app_runflow.py           |  5 +-
 quantui/backends/slurm_ingest.py | 56 ++++++++++++++++---
 tests/test_jobdirs.py            | 93 ++++++++++++++++++++++++++++++++
 4 files changed, 190 insertions(+), 9 deletions(-)

diff --git a/quantui/app_formatters.py b/quantui/app_formatters.py
index 5436b75..f635569 100644
--- a/quantui/app_formatters.py
+++ b/quantui/app_formatters.py
@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import html
 from pathlib import Path
 from typing import Any, Optional
 
@@ -815,6 +816,49 @@ def format_reorg_result(r: Any) -> str:
     )
 
 
+def slurm_history_marker(data: dict[str, Any]) -> str:
+    """Short History-list marker for a SLURM result, e.g. ``🖥 SLURM 812345·a2 ``.
+
+    Empty for local results and for results saved before SLURM provenance
+    was recorded (M-JOBDIRS JD.6).
+    """
+    if data.get("execution_backend") != "slurm":
+        return ""
+    info = data.get("slurm") or {}
+    job_id = info.get("job_id")
+    attempt = info.get("attempt")
+    text = "🖥 SLURM"
+    if job_id:
+        text += f" {job_id}"
+        if attempt:
+            text += f"·a{attempt}"
+    return text + " "
+
+
+def _slurm_provenance_row(data: dict[str, Any]) -> str:
+    """History-card row naming the SLURM job, attempt and job folder."""
+    if data.get("execution_backend") != "slurm":
+        return ""
+    info = data.get("slurm") or {}
+    parts = ["SLURM batch"]
+    if info.get("job_id"):
+        parts.append(f"job {html.escape(str(info['job_id']))}")
+    if info.get("attempt"):
+        parts.append(f"attempt {int(info['attempt'])}")
+    value = " &middot; ".join(parts)
+    folder = info.get("attempt_dir") or info.get("job_dir")
+    if folder:
+        value += (
+            f'<br><span style="font-family:monospace;font-size:12px;'
+            f'color:{_theme.css.TEXT_MUTED_LIGHT}">{html.escape(str(folder))}</span>'
+        )
+    return (
+        f'<tr><td style="padding:3px 18px 3px 0;color:{_theme.css.TEXT_LABEL};'
+        f'vertical-align:top">Ran on</td>'
+        f'<td style="color:{_theme.css.TEXT_HEADING}">{value}</td></tr>'
+    )
+
+
 def format_past_result(data: dict[str, Any], result_dir: Optional[Path] = None) -> str:
     """Format a saved result.json payload as an HTML result card."""
     import base64 as _b64
@@ -912,6 +956,7 @@ def format_past_result(data: dict[str, Any], result_dir: Optional[Path] = None)
     # Shared 'extra' rows (correlation breakdown / solvent / device / dipole /
     # Mulliken) — same builder as the live card so the two never drift.
     _extra = _result_extra_rows(lambda k, d=None: data.get(k, d))
+    _extra += _slurm_provenance_row(data)
 
     # Embed thumbnail if saved
     _thumb_html = ""
diff --git a/quantui/app_runflow.py b/quantui/app_runflow.py
index 0589bea..77e0be1 100644
--- a/quantui/app_runflow.py
+++ b/quantui/app_runflow.py
@@ -2667,6 +2667,7 @@ def refresh_results_browser(app: Any) -> None:
         from quantui import list_results, load_result
     except ImportError:
         return
+    from quantui.app_formatters import slurm_history_marker
     from quantui.app_history import (
         apply_history_filter,
         entry_date,
@@ -2696,12 +2697,14 @@ def refresh_results_browser(app: Any) -> None:
             # comes from result.json's ``calibration_run_id`` extras field
             # written by the worker.
             calib_marker = "🔧 " if data.get("calibration_run_id") else ""
+            # SLURM batch results carry job id + attempt (M-JOBDIRS JD.6).
+            slurm_marker = slurm_history_marker(data)
             formula = data.get("formula", "?")
             method = data.get("method", "?")
             basis = data.get("basis", "?")
             label = (
                 f"{ts}  ·  [{calc_badge}]  "
-                f"{calib_marker}{formula}  "
+                f"{calib_marker}{slurm_marker}{formula}  "
                 f"{method}/{basis}"
             )
             entries.append(
diff --git a/quantui/backends/slurm_ingest.py b/quantui/backends/slurm_ingest.py
index bfb30b5..1675a7a 100644
--- a/quantui/backends/slurm_ingest.py
+++ b/quantui/backends/slurm_ingest.py
@@ -11,7 +11,7 @@
 from types import SimpleNamespace
 from typing import Any
 
-from .registry import JobRecord
+from .registry import JobRecord, parse_attempt_dir_name
 from .worker_payload import molecule_from_dict
 
 # Sidecar files the batch worker may write for Analysis replay parity.
@@ -25,6 +25,29 @@
 )
 
 
+def slurm_provenance(record: JobRecord, staging: Path) -> dict[str, Any]:
+    """``result.json`` extras that mark a History entry as a SLURM run.
+
+    M-JOBDIRS JD.5: stored under ``execution_backend`` / ``slurm`` so the
+    History list and card can tell SLURM results from local ones. Additive —
+    local results and results ingested before this simply lack the keys.
+    For a job-dir record, *staging* is the attempt dir and its name gives the
+    attempt number and the SLURM job id that produced it (which, for a
+    hand-run ``sbatch``, differs from the record's tracked job).
+    """
+    parsed = parse_attempt_dir_name(staging.name) if record.job_dir else None
+    return {
+        "execution_backend": "slurm",
+        "slurm": {
+            "job_id": parsed[1] if parsed else record.slurm_job_id,
+            "attempt": parsed[0] if parsed else None,
+            "job_dir": record.job_dir or record.staging_dir,
+            "attempt_dir": str(staging) if parsed else None,
+            "request_id": record.request_id,
+        },
+    }
+
+
 def _copy_staging_sidecars(staging: Path, saved_dir: Path) -> None:
     for name in _STAGING_SIDECAR_FILES:
         src = staging / name
@@ -97,7 +120,11 @@ def _copy_trajectory(staging: Path, saved_dir: Path, payload: dict[str, Any]) ->
 
 
 def _ingest_frequency(
-    staging: Path, payload: dict[str, Any], record: JobRecord, log_text: str
+    staging: Path,
+    payload: dict[str, Any],
+    record: JobRecord,
+    log_text: str,
+    extras: dict[str, Any],
 ) -> Path:
     from quantui import save_result
     from quantui.results_storage import save_molden
@@ -109,6 +136,7 @@ def _ingest_frequency(
         pyscf_log=log_text,
         calc_type="frequency",
         spectra=spectra,
+        extras=extras,
     )
     ir = spectra.get("ir") or {}
     freqs = ir.get("frequencies_cm1")
@@ -135,7 +163,10 @@ def _ingest_frequency(
 
 
 def _ingest_reorganization_energy(
-    payload: dict[str, Any], record: JobRecord, log_text: str
+    payload: dict[str, Any],
+    record: JobRecord,
+    log_text: str,
+    extras: dict[str, Any],
 ) -> Path:
     from quantui import save_result
 
@@ -180,6 +211,7 @@ def _ingest_reorganization_energy(
         pyscf_log=log_text,
         calc_type="reorganization_energy",
         spectra=payload.get("spectra") or {},
+        extras=extras,
     )
     _finalize_history_entry(saved_dir)
     return saved_dir
@@ -198,9 +230,16 @@ def _ingest_with_sidecars(
     return saved_dir
 
 
-def ingest_staging_success(record: JobRecord, log_text: str = "") -> Path:
-    """Read ``result.json`` from staging and save under ``results/``."""
-    staging = record.staging_path
+def ingest_staging_success(
+    record: JobRecord, log_text: str = "", *, attempt_dir: Path | None = None
+) -> Path:
+    """Read ``result.json`` from staging and save under ``results/``.
+
+    *attempt_dir* selects a specific attempt (used for hand-run attempts the
+    record does not track); by default the record's tracked staging path.
+    """
+    staging = attempt_dir if attempt_dir is not None else record.staging_path
+    extras = slurm_provenance(record, staging)
     result_path = staging / "result.json"
     if not result_path.exists():
         raise FileNotFoundError(f"Missing staging result: {result_path}")
@@ -211,10 +250,10 @@ def ingest_staging_success(record: JobRecord, log_text: str = "") -> Path:
     from quantui import save_result
 
     if calc_type == "frequency":
-        return _ingest_frequency(staging, payload, record, log_text)
+        return _ingest_frequency(staging, payload, record, log_text, extras)
 
     if calc_type == "reorganization_energy":
-        return _ingest_reorganization_energy(payload, record, log_text)
+        return _ingest_reorganization_energy(payload, record, log_text, extras)
 
     result = _basic_result(payload, record)
     spectra = payload.get("spectra")
@@ -223,6 +262,7 @@ def ingest_staging_success(record: JobRecord, log_text: str = "") -> Path:
         pyscf_log=log_text,
         calc_type=calc_type,
         spectra=spectra if spectra is not None else {},
+        extras=extras,
     )
 
     return _ingest_with_sidecars(staging, saved_dir, payload)
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index ebe4793..1399e34 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -421,3 +421,96 @@ def test_submit_script_rerun_by_hand_keeps_both_attempts(tmp_path, monkeypatch):
     assert (job / "attempt-01_job41" / "live.log").read_text() == first_log
     # The hand-run job (42) is not the tracked one (41): record untouched by it.
     assert backend.registry.load(rid).slurm_job_id == "41"
+
+
+# ---------------------------------------------------------------------------
+# History provenance (JD.5 / JD.6)
+# ---------------------------------------------------------------------------
+
+
+class TestHistoryProvenance:
+    @pytest.mark.parametrize(
+        "calc_type", ["single_point", "frequency", "reorganization_energy"]
+    )
+    def test_legacy_record_ingest_is_tagged_slurm(
+        self, tmp_path, monkeypatch, calc_type
+    ):
+        """All three save_result paths carry the provenance extras."""
+        from quantui.backends.slurm_ingest import ingest_staging_success
+        from tests.slurm_ingest_helpers import (
+            make_staging_record,
+            patch_results_root,
+            sample_payload,
+        )
+
+        patch_results_root(tmp_path, monkeypatch)
+        record, _staging = make_staging_record(
+            tmp_path, sample_payload(calc_type), calc_type=calc_type
+        )
+        record.slurm_job_id = "4242"
+        saved = ingest_staging_success(record)
+        data = json.loads((saved / "result.json").read_text())
+        assert data["execution_backend"] == "slurm"
+        assert data["slurm"]["job_id"] == "4242"
+        assert data["slurm"]["attempt"] is None
+        assert data["slurm"]["request_id"] == record.request_id
+
+    def test_attempt_dir_ingest_records_attempt_and_its_job_id(
+        self, tmp_path, monkeypatch, registry
+    ):
+        from quantui.backends.slurm_ingest import ingest_staging_success
+        from tests.slurm_ingest_helpers import patch_results_root, sample_payload
+
+        patch_results_root(tmp_path, monkeypatch)
+        rec = registry.create(_request(), "cluster_slurm", job_name="run")
+        rec = registry.start_attempt(rec.request_id, "100")
+        # A hand-run second attempt the record does not track.
+        attempt = Path(rec.job_dir) / "attempt-02_job555"
+        attempt.mkdir()
+        (attempt / "result.json").write_text(json.dumps(sample_payload("single_point")))
+        saved = ingest_staging_success(rec, attempt_dir=attempt)
+        info = json.loads((saved / "result.json").read_text())["slurm"]
+        assert info["job_id"] == "555"
+        assert info["attempt"] == 2
+        assert info["attempt_dir"] == str(attempt)
+        assert info["job_dir"] == rec.job_dir
+
+    def test_history_marker_and_card_row(self):
+        from quantui.app_formatters import format_past_result, slurm_history_marker
+
+        data = {
+            "calc_type": "single_point",
+            "formula": "H2",
+            "method": "RHF",
+            "basis": "STO-3G",
+            "energy_hartree": -1.1,
+            "energy_ev": -30.0,
+            "converged": True,
+            "execution_backend": "slurm",
+            "slurm": {
+                "job_id": "812345",
+                "attempt": 2,
+                "job_dir": "/scratch/u/run",
+                "attempt_dir": "/scratch/u/run/attempt-02_job812345",
+            },
+        }
+        assert slurm_history_marker(data) == "🖥 SLURM 812345·a2 "
+        card = format_past_result(data)
+        assert "Ran on" in card
+        assert "job 812345 &middot; attempt 2" in card
+        assert "/scratch/u/run/attempt-02_job812345" in card
+
+    def test_local_result_has_no_marker_or_row(self):
+        from quantui.app_formatters import format_past_result, slurm_history_marker
+
+        data = {
+            "calc_type": "single_point",
+            "formula": "H2",
+            "method": "RHF",
+            "basis": "STO-3G",
+            "energy_hartree": -1.1,
+            "energy_ev": -30.0,
+            "converged": True,
+        }
+        assert slurm_history_marker(data) == ""
+        assert "Ran on" not in format_past_result(data)

From 2d350a68b601bc5789cdc34cb07702dfc8cf41bc Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:03:56 +0000
Subject: [PATCH 4/9] feat(slurm): pick up hand-run attempts on refresh; never
 ingest twice

M-JOBDIRS JD.11.

- A Cluster Jobs refresh now saves every finished attempt that is not yet
  in History: hand-run `sbatch submit.slurm` attempts (which the app never
  monitored) and tracked successes nobody reconnected to. The attempt the
  app is currently monitoring, or whose tracked job is still active, is
  left alone. result.json is written only on success, so failed attempts
  stay out of History.
- JobRecord.ingested_attempts records what has been saved; ingest_attempt()
  saves and records in one place. A hand-run attempt does not retarget the
  record's result_dir.
- Reconnecting (View progress) to a finished job whose result is already in
  History now shows it without saving a duplicate entry. Legacy records
  are covered too: a result_dir that already points at a History entry
  counts as ingested (this duplicate-on-reconnect predates M-JOBDIRS).

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app_slurm.py             |  68 +++++++++++++---
 quantui/backends/slurm_ingest.py |  62 +++++++++++++-
 tests/test_jobdirs.py            | 136 +++++++++++++++++++++++++++++++
 3 files changed, 254 insertions(+), 12 deletions(-)

diff --git a/quantui/app_slurm.py b/quantui/app_slurm.py
index ea996d8..6f193af 100644
--- a/quantui/app_slurm.py
+++ b/quantui/app_slurm.py
@@ -13,6 +13,7 @@
 import logging
 import threading
 import time
+from pathlib import Path
 from typing import Any
 
 from IPython.display import HTML
@@ -298,6 +299,36 @@ def _slurm_job_error_cell(record: JobRecord) -> str:
     return f"{html.escape(code)}: {html.escape(message)}"
 
 
+def ingest_new_slurm_attempts(app: Any) -> list[Path]:
+    """Save finished, not-yet-ingested attempts to History (M-JOBDIRS JD.11).
+
+    Picks up hand-run ``sbatch submit.slurm`` attempts (which the app never
+    monitored) and tracked successes nobody reconnected to. The attempt the
+    app is monitoring right now is left to the monitor, and a tracked
+    attempt whose job is still active is skipped.
+    """
+    from quantui.backends.slurm_ingest import ingest_attempt, uningested_attempts
+
+    registry = ensure_job_registry(app)
+    monitored = getattr(app, "_slurm_active_request_id", None)
+    saved: list[Path] = []
+    for record in list_slurm_jobs(app):
+        for attempt in uningested_attempts(record):
+            tracked = attempt == record.staging_path
+            if tracked and (
+                record.request_id == monitored
+                or record.status.lower() in _ACTIVE_SLURM_STATUSES
+            ):
+                continue
+            try:
+                saved.append(ingest_attempt(registry, record, attempt))
+            except Exception:  # noqa: BLE001 — one bad attempt must not block others
+                logger.exception("Failed to ingest SLURM attempt %s", attempt)
+                continue
+            record = registry.load(record.request_id) or record
+    return saved
+
+
 def refresh_slurm_jobs_tab(app: Any) -> None:
     """Re-render the Cluster Jobs tab from the on-disk registry."""
     summary = getattr(app, "_slurm_jobs_summary_html", None)
@@ -314,6 +345,21 @@ def refresh_slurm_jobs_tab(app: Any) -> None:
         except Exception:  # noqa: BLE001 — UI refresh must not crash the app
             logger.exception("Failed to refresh SLURM registry statuses")
 
+    new_saved = ingest_new_slurm_attempts(app)
+    if new_saved:
+        from quantui.app_runflow import refresh_results_browser
+
+        try:
+            refresh_results_browser(app)
+        except Exception:  # noqa: BLE001
+            logger.exception("Failed to refresh History after SLURM ingest")
+        status = getattr(app, "_slurm_jobs_status_html", None)
+        if status is not None:
+            status.value = (
+                f'<span style="color:{_theme.css.TEXT_STRONG};font-size:12px">'
+                f"Saved {len(new_saved)} finished cluster run(s) to History.</span>"
+            )
+
     records = list_slurm_jobs(app)
     active = active_slurm_job_count(app)
     limit = max_concurrent_slurm_jobs()
@@ -555,31 +601,31 @@ def _ingest_terminal_job(app: Any, record: Any) -> None:
 
 
 def _ingest_success(app: Any, record: Any) -> None:
+    # Reload: the record passed in may predate an ingest done elsewhere
+    # (a Cluster Jobs refresh, or an earlier reconnect).
+    record = app._job_registry.load(record.request_id) or record
     staging = record.staging_path
     result_path = staging / "result.json"
-    log_path = record.live_log_path
     if not result_path.exists():
         app.run_status.value = "SLURM job finished but result.json is missing."
         return
 
     payload = json.loads(result_path.read_text(encoding="utf-8"))
-    log_text = (
-        log_path.read_text(encoding="utf-8", errors="replace")
-        if log_path.exists()
-        else ""
-    )
 
     try:
         from quantui.app_runflow import refresh_results_browser
         from quantui.backends.slurm_ingest import (
+            already_ingested,
             completion_summary_html,
-            ingest_staging_success,
+            ingest_attempt,
         )
 
-        saved_dir = ingest_staging_success(record, log_text)
-        app._job_registry.update_status(
-            record.request_id, "success", result_dir=str(saved_dir)
-        )
+        if already_ingested(record, staging) and record.result_dir:
+            # Never save the same run to History twice (e.g. View progress
+            # on a finished job, or a refresh that already picked it up).
+            saved_dir = Path(record.result_dir)
+        else:
+            saved_dir = ingest_attempt(app._job_registry, record)
         app._last_result_dir = saved_dir
         refresh_results_browser(app)
         calc_type = payload.get("calc_type", record.calc_type)
diff --git a/quantui/backends/slurm_ingest.py b/quantui/backends/slurm_ingest.py
index 1675a7a..c38199c 100644
--- a/quantui/backends/slurm_ingest.py
+++ b/quantui/backends/slurm_ingest.py
@@ -11,7 +11,7 @@
 from types import SimpleNamespace
 from typing import Any
 
-from .registry import JobRecord, parse_attempt_dir_name
+from .registry import JobRecord, JobRegistry, parse_attempt_dir_name
 from .worker_payload import molecule_from_dict
 
 # Sidecar files the batch worker may write for Analysis replay parity.
@@ -268,6 +268,66 @@ def ingest_staging_success(
     return _ingest_with_sidecars(staging, saved_dir, payload)
 
 
+def already_ingested(record: JobRecord, staging: Path) -> bool:
+    """True when the run in *staging* has already been saved to History.
+
+    Job-dir records list ingested attempt dirs by name. Legacy records
+    predate that list; for them, a ``result_dir`` that points at an existing
+    History entry (the worker sets it to the staging dir; the app replaces
+    it with the saved dir after ingest) means the tracked run was ingested.
+    """
+    if staging.name in record.ingested_attempts:
+        return True
+    if record.job_dir is None and record.result_dir:
+        saved = Path(record.result_dir)
+        return saved != staging and (saved / "result.json").exists()
+    return False
+
+
+def uningested_attempts(record: JobRecord) -> list[Path]:
+    """Finished attempt dirs of a job-dir record not yet saved to History.
+
+    ``result.json`` is written only when a run succeeds, so its presence
+    marks a finished, successful attempt (M-JOBDIRS: successes only).
+    Includes hand-run ``sbatch submit.slurm`` attempts the record never
+    tracked (JD.11). Legacy records return ``[]``: their single run is
+    ingested by the monitor / reconnect path, as before.
+    """
+    if record.job_dir is None:
+        return []
+    return [
+        d
+        for d in record.attempt_dirs()
+        if (d / "result.json").exists() and not already_ingested(record, d)
+    ]
+
+
+def ingest_attempt(
+    registry: JobRegistry, record: JobRecord, attempt_dir: Path | None = None
+) -> Path:
+    """Save one attempt to History and record it as ingested.
+
+    *attempt_dir* defaults to the record's tracked staging path. The
+    record's ``result_dir`` is updated only for the tracked attempt, so it
+    keeps pointing at the History entry of the run the record reports on.
+    """
+    staging = attempt_dir if attempt_dir is not None else record.staging_path
+    log_path = staging / "live.log"
+    log_text = (
+        log_path.read_text(encoding="utf-8", errors="replace")
+        if log_path.exists()
+        else ""
+    )
+    saved_dir = ingest_staging_success(record, log_text, attempt_dir=staging)
+    tracked = staging == record.staging_path
+    registry.mark_ingested(
+        record.request_id,
+        staging.name,
+        result_dir=str(saved_dir) if tracked else None,
+    )
+    return saved_dir
+
+
 def completion_summary_html(saved_dir: Path, payload: dict[str, Any]) -> str:
     calc_type = payload.get("calc_type", "single_point")
     energy = float(payload.get("energy_hartree", float("nan")))
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index 1399e34..98236ed 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -514,3 +514,139 @@ def test_local_result_has_no_marker_or_row(self):
         }
         assert slurm_history_marker(data) == ""
         assert "Ran on" not in format_past_result(data)
+
+
+# ---------------------------------------------------------------------------
+# Hand-run attempts reach History; nothing is ingested twice (JD.11)
+# ---------------------------------------------------------------------------
+
+
+def _finished_attempt(job_dir: Path, name: str) -> Path:
+    from tests.slurm_ingest_helpers import sample_payload
+
+    attempt = job_dir / name
+    attempt.mkdir()
+    (attempt / "result.json").write_text(json.dumps(sample_payload("single_point")))
+    (attempt / "live.log").write_text(f"log of {name}\n")
+    return attempt
+
+
+def _jobs_tab_app(registry, **overrides):
+    from types import SimpleNamespace
+
+    fields = dict(
+        _job_registry=registry,
+        _slurm_jobs_summary_html=SimpleNamespace(value=""),
+        _slurm_jobs_table_html=SimpleNamespace(value=""),
+        _slurm_jobs_select=SimpleNamespace(options=[], value="", disabled=True),
+        _slurm_jobs_status_html=SimpleNamespace(value=""),
+        _slurm_active_request_id=None,
+    )
+    fields.update(overrides)
+    return SimpleNamespace(**fields)
+
+
+class TestHandRunAttempts:
+    @pytest.fixture
+    def job(self, tmp_path, monkeypatch, registry):
+        from tests.slurm_ingest_helpers import patch_results_root
+
+        results = patch_results_root(tmp_path, monkeypatch)
+        rec = registry.create(_request(), "cluster_slurm", job_name="run")
+        registry.start_attempt(rec.request_id, "100")
+        registry.update_status(rec.request_id, "error")
+        return registry, rec.request_id, Path(rec.job_dir), results
+
+    def test_uningested_attempts_lists_only_finished_new_ones(self, job):
+        from quantui.backends.slurm_ingest import uningested_attempts
+
+        registry, rid, job_dir, _ = job
+        (job_dir / "attempt-01_job100").mkdir()  # failed: no result.json
+        done = _finished_attempt(job_dir, "attempt-02_job555")
+        assert uningested_attempts(registry.load(rid)) == [done]
+
+    def test_ingest_attempt_marks_ingested_without_retargeting_record(self, job):
+        from quantui.backends.slurm_ingest import ingest_attempt, uningested_attempts
+
+        registry, rid, job_dir, results = job
+        done = _finished_attempt(job_dir, "attempt-02_job555")
+        saved = ingest_attempt(registry, registry.load(rid), done)
+        assert (saved / "pyscf.log").read_text() == "log of attempt-02_job555\n"
+        rec = registry.load(rid)
+        assert rec.ingested_attempts == ["attempt-02_job555"]
+        assert rec.result_dir is None  # hand-run attempt: record not retargeted
+        assert uningested_attempts(rec) == []
+        assert len(list(results.iterdir())) == 1
+
+    @patch("quantui.app_slurm.is_slurm_available", return_value=False)
+    def test_refresh_ingests_hand_run_attempt_once(self, _avail, job):
+        from quantui.app_slurm import refresh_slurm_jobs_tab
+
+        registry, _rid, job_dir, results = job
+        _finished_attempt(job_dir, "attempt-02_job555")
+        app = _jobs_tab_app(registry)
+        with patch("quantui.app_runflow.refresh_results_browser") as refresh:
+            refresh_slurm_jobs_tab(app)
+            refresh_slurm_jobs_tab(app)
+        assert refresh.call_count == 1
+        assert "Saved 1 finished cluster run" in app._slurm_jobs_status_html.value
+        [entry] = list(results.iterdir())
+        info = json.loads((entry / "result.json").read_text())["slurm"]
+        assert (info["job_id"], info["attempt"]) == ("555", 2)
+
+    @patch("quantui.app_slurm.is_slurm_available", return_value=False)
+    def test_refresh_leaves_monitored_attempt_to_the_monitor(self, _avail, job):
+        from quantui.app_slurm import ingest_new_slurm_attempts
+
+        registry, rid, job_dir, results = job
+        registry.update_status(rid, "success")
+        _finished_attempt(job_dir, "attempt-01_job100")  # the tracked job
+        app = _jobs_tab_app(registry, _slurm_active_request_id=rid)
+        assert ingest_new_slurm_attempts(app) == []
+        app._slurm_active_request_id = None
+        assert len(ingest_new_slurm_attempts(app)) == 1
+
+    def test_view_on_finished_job_does_not_duplicate_history(self, job):
+        """Reconnecting to an ingested job shows it, but saves nothing new."""
+        from quantui.app_slurm import _ingest_success
+
+        registry, rid, job_dir, results = job
+        registry.update_status(rid, "success")
+        _finished_attempt(job_dir, "attempt-01_job100")
+        app = _jobs_tab_app(
+            registry,
+            run_status=_Value(),
+            run_output=_Sink(),
+            result_output=_Sink(),
+        )
+        with patch("quantui.app_runflow.refresh_results_browser"):
+            _ingest_success(app, registry.load(rid))
+            _ingest_success(app, registry.load(rid))
+        assert len(list(results.iterdir())) == 1
+        rec = registry.load(rid)
+        assert rec.result_dir == str(next(results.iterdir()))
+
+    def test_legacy_record_already_ingested_is_detected(self, tmp_path, registry):
+        from quantui.backends.slurm_ingest import already_ingested
+
+        rec = registry.create(_request("leg"), "cluster_slurm")
+        staging = rec.staging_path
+        rec.result_dir = str(staging)  # worker's value: not yet ingested
+        assert already_ingested(rec, staging) is False
+        saved = tmp_path / "results" / "x"
+        saved.mkdir(parents=True)
+        (saved / "result.json").write_text("{}")
+        rec.result_dir = str(saved)  # app's value after ingest
+        assert already_ingested(rec, staging) is True
+
+
+class _Value:
+    value = ""
+
+
+class _Sink:
+    def append_stdout(self, _text):
+        pass
+
+    def append_display_data(self, _obj):
+        pass

From 84b57c04a8510ff5e65a9724a0e23e44e52fec5d Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:05:05 +0000
Subject: [PATCH 5/9] feat(slurm): Resubmit button on the Cluster Jobs tab

M-JOBDIRS JD.4 (UI).

Resubmit runs the selected finished or failed job's submit.slurm again as
a new attempt in the same job folder (SlurmBackend.resubmit), then opens
the Calculate tab monitoring the new attempt, the same as View progress.
Earlier attempts' files are kept, and checkpoints let the new attempt pick
up where the failed one stopped. It is refused while another calculation
is running or monitored, for jobs still active, and for jobs submitted
before per-job folders existed (with a message saying why).

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app.py          |  7 ++++++
 quantui/app_builders.py | 12 ++++++++-
 quantui/app_slurm.py    | 48 +++++++++++++++++++++++++++++++++++
 tests/test_jobdirs.py   | 55 +++++++++++++++++++++++++++++++++++++++++
 4 files changed, 121 insertions(+), 1 deletion(-)

diff --git a/quantui/app.py b/quantui/app.py
index 13df928..d996d48 100644
--- a/quantui/app.py
+++ b/quantui/app.py
@@ -415,6 +415,9 @@
 from quantui.app_slurm import (
     on_slurm_jobs_remove_clicked as _slurm_on_jobs_remove_clicked,
 )
+from quantui.app_slurm import (
+    on_slurm_jobs_resubmit_clicked as _slurm_on_jobs_resubmit_clicked,
+)
 from quantui.app_slurm import (
     on_slurm_jobs_view_clicked as _slurm_on_jobs_view_clicked,
 )
@@ -1491,6 +1494,7 @@ class QuantUIApp:
         _slurm_jobs_view_btn: Any
         _slurm_jobs_cancel_btn: Any
         _slurm_jobs_remove_btn: Any
+        _slurm_jobs_resubmit_btn: Any
         _slurm_jobs_status_html: Any
         slurm_jobs_tab_panel: Any
         _slurm_jobs_tab_index: int | None
@@ -2650,6 +2654,9 @@ def _wire_callbacks(self) -> None:
         self._slurm_jobs_remove_btn.on_click(
             self._safe_cb(lambda _btn: _slurm_on_jobs_remove_clicked(self, _btn))
         )
+        self._slurm_jobs_resubmit_btn.on_click(
+            self._safe_cb(lambda _btn: _slurm_on_jobs_resubmit_clicked(self, _btn))
+        )
         self.cancel_btn.on_click(self._safe_cb(self._on_cancel))
         self.basis_fix_btn.on_click(self._safe_cb(self._on_basis_fix))
         self.charge_mult_suggest_btn.on_click(
diff --git a/quantui/app_builders.py b/quantui/app_builders.py
index 0c95c1c..da7f522 100644
--- a/quantui/app_builders.py
+++ b/quantui/app_builders.py
@@ -3456,6 +3456,15 @@ def build_slurm_jobs_tab(app: Any, *, layout_fn: Any) -> None:
         layout=layout_fn(width="120px"),
         tooltip="Cancel the selected active cluster job",
     )
+    app._slurm_jobs_resubmit_btn = widgets.Button(
+        description="Resubmit",
+        icon="redo",
+        layout=layout_fn(width="120px"),
+        tooltip=(
+            "Run a finished or failed job again in the same job folder, as a "
+            "new attempt. Earlier attempts' files are kept."
+        ),
+    )
     app._slurm_jobs_remove_btn = widgets.Button(
         description="Remove",
         icon="trash",
@@ -3465,7 +3474,7 @@ def build_slurm_jobs_tab(app: Any, *, layout_fn: Any) -> None:
     app._slurm_jobs_status_html = widgets.HTML(
         value=(
             f'<span style="font-size:12px;color:{_theme.css.TEXT_SUBTLE}">'
-            "Select a job and use View progress or Cancel.</span>"
+            "Select a job and use View progress, Cancel, or Resubmit.</span>"
         )
     )
 
@@ -3485,6 +3494,7 @@ def build_slurm_jobs_tab(app: Any, *, layout_fn: Any) -> None:
                 [
                     app._slurm_jobs_view_btn,
                     app._slurm_jobs_cancel_btn,
+                    app._slurm_jobs_resubmit_btn,
                     app._slurm_jobs_remove_btn,
                 ],
                 layout=layout_fn(gap="8px", margin="6px 0"),
diff --git a/quantui/app_slurm.py b/quantui/app_slurm.py
index 6f193af..52df04f 100644
--- a/quantui/app_slurm.py
+++ b/quantui/app_slurm.py
@@ -493,6 +493,54 @@ def on_slurm_jobs_cancel_clicked(app: Any, _btn: Any = None) -> None:
         )
 
 
+def resubmit_slurm_job(app: Any, request_id: str) -> tuple[bool, str]:
+    """Resubmit a finished job as a new attempt in its job dir (M-JOBDIRS JD.4).
+
+    Returns ``(ok, user message)``. On success the Calculate tab starts
+    monitoring the new attempt.
+    """
+    if getattr(app, "_calc_running", False):
+        return False, (
+            "A calculation is already running or being monitored. "
+            "Wait for it to finish before resubmitting."
+        )
+    ensure_job_registry(app)
+    backend = slurm_backend_for_app(app)
+    try:
+        new_job_id = backend.resubmit(request_id)
+    except (ValueError, SecurityError) as exc:
+        return False, str(exc)
+    except RuntimeError as exc:
+        return False, f"Resubmit failed: {exc}"
+    record = app._job_registry.load(request_id)
+    folder = (record.job_dir if record else None) or "?"
+    return True, (
+        f"Resubmitted as SLURM job {new_job_id}. Its attempt folder appears "
+        f"in {folder} when the job starts; earlier attempts are kept."
+    )
+
+
+def on_slurm_jobs_resubmit_clicked(app: Any, _btn: Any = None) -> None:
+    select = getattr(app, "_slurm_jobs_select", None)
+    request_id = select.value if select is not None else None
+    if not request_id:
+        return
+    ok, message = resubmit_slurm_job(app, request_id)
+    refresh_slurm_jobs_tab(app)
+    _update_slurm_jobs_tab_title(app)
+    status = getattr(app, "_slurm_jobs_status_html", None)
+    if status is not None:
+        color = _theme.css.TEXT_STRONG if ok else _theme.css.ACCENT_ERROR
+        status.value = (
+            f'<span style="color:{color};font-size:12px">{html.escape(message)}</span>'
+        )
+    if ok:
+        attach_slurm_job(app, request_id)
+        go_to = getattr(app, "_go_to_calculate_tab", None)
+        if callable(go_to):
+            go_to()
+
+
 def on_slurm_jobs_remove_clicked(app: Any, _btn: Any = None) -> None:
     select = getattr(app, "_slurm_jobs_select", None)
     request_id = select.value if select is not None else None
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index 98236ed..8521923 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -650,3 +650,58 @@ def append_stdout(self, _text):
 
     def append_display_data(self, _obj):
         pass
+
+
+# ---------------------------------------------------------------------------
+# Resubmit button (JD.4 UI)
+# ---------------------------------------------------------------------------
+
+
+class TestResubmitButton:
+    @pytest.fixture
+    def app(self, backend, monkeypatch):
+        monkeypatch.setattr(
+            "quantui.backends.slurm.check_submit_cooldown", lambda _s: None
+        )
+        monkeypatch.setattr(
+            "quantui.app_slurm.slurm_backend_for_app", lambda _app: backend
+        )
+        monkeypatch.setattr("quantui.app_slurm.is_slurm_available", lambda: False)
+        with _sbatch_returning("100"):
+            rid = backend.dispatch(_request())
+        backend.registry.update_status(rid, "error")
+        app = _jobs_tab_app(backend.registry, _calc_running=False)
+        app._slurm_jobs_select.value = rid
+        return app, rid
+
+    def test_resubmit_starts_monitoring_new_attempt(self, app, backend):
+        from quantui.app_slurm import on_slurm_jobs_resubmit_clicked
+
+        app_ns, rid = app
+        with (
+            _sbatch_returning("200"),
+            patch("quantui.app_slurm.attach_slurm_job") as attach,
+        ):
+            on_slurm_jobs_resubmit_clicked(app_ns)
+        attach.assert_called_once_with(app_ns, rid)
+        assert "Resubmitted as SLURM job 200" in app_ns._slurm_jobs_status_html.value
+        assert backend.registry.load(rid).slurm_job_id == "200"
+
+    def test_resubmit_blocked_while_running(self, app, backend):
+        from quantui.app_slurm import resubmit_slurm_job
+
+        app_ns, rid = app
+        app_ns._calc_running = True
+        ok, message = resubmit_slurm_job(app_ns, rid)
+        assert not ok
+        assert "already running" in message
+        assert backend.registry.load(rid).slurm_job_id == "100"
+
+    def test_resubmit_legacy_record_explains(self, app, backend):
+        from quantui.app_slurm import resubmit_slurm_job
+
+        app_ns, _rid = app
+        backend.registry.create(_request("legacy"), "cluster_slurm", status="error")
+        ok, message = resubmit_slurm_job(app_ns, "legacy")
+        assert not ok
+        assert "before per-job folders" in message

From 70103b6c7a419f019848d197a2dbafcd5e23ff90 Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:07:04 +0000
Subject: [PATCH 6/9] feat(settings): configurable SLURM job folder root

M-JOBDIRS JD.1.

- New compute.slurm_job_root user setting (blank = ~/.quantui/staging).
  Precedence: QUANTUI_STAGING_DIR > setting > default, resolved in
  cluster_config.default_staging_root(), so the app, the CLI and the batch
  worker agree.
- System Settings shows a "SLURM job folder" field when SLURM is
  available. It accepts an absolute path (created if missing), shows an
  error for relative or unusable paths, and is locked with a note when
  QUANTUI_STAGING_DIR is set. A change applies to new submissions; jobs
  already in the registry keep their absolute folder paths.

Useful on clusters with small home quotas: point it at scratch or a
project directory (Apptainer runs bind it automatically, see the job-dir
commit).

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app.py                     | 15 ++++++
 quantui/app_builders.py            | 42 +++++++++++++++++
 quantui/app_slurm.py               | 40 ++++++++++++++++
 quantui/backends/cluster_config.py | 23 ++++++++-
 quantui/user_settings.py           | 16 +++++++
 tests/test_jobdirs.py              | 75 ++++++++++++++++++++++++++++++
 6 files changed, 210 insertions(+), 1 deletion(-)

diff --git a/quantui/app.py b/quantui/app.py
index d996d48..2ed7c14 100644
--- a/quantui/app.py
+++ b/quantui/app.py
@@ -406,6 +406,9 @@
 from quantui.app_runflow import (
     update_scan_widgets as _run_update_scan_widgets,
 )
+from quantui.app_slurm import (
+    on_slurm_job_root_changed as _slurm_on_job_root_changed,
+)
 from quantui.app_slurm import (
     on_slurm_jobs_cancel_clicked as _slurm_on_jobs_cancel_clicked,
 )
@@ -571,6 +574,7 @@
 from quantui.app_xyz_input import (
     on_xyz_fill_table as _xyz_on_fill_table,
 )
+from quantui.backends import cluster_config as _cluster_cfg
 from quantui.backends.dispatch import is_slurm_available
 from quantui.cancellation import CalcCancelled as _CalcCancelled
 
@@ -1443,6 +1447,8 @@ class QuantUIApp:
         density_fit_enabled_cb: Any
         freq_parallel_enabled_cb: Any
         execution_backend_dd: Any
+        slurm_job_root_txt: Any
+        slurm_job_root_note: Any
         quantum_engine_dd: Any
         quantum_engine_note: Any
         engine_capability_html: Any
@@ -2128,6 +2134,9 @@ def _build_status_panel(self) -> None:
             execution_backend=self._user_settings.compute.execution_backend,
             slurm_available=is_slurm_available(),
             quantum_engine=self._user_settings.compute.quantum_engine,
+            slurm_job_root=self._user_settings.compute.slurm_job_root,
+            slurm_job_root_env_locked=_cluster_cfg.staging_root_env_configured(),
+            slurm_job_root_effective=str(_cluster_cfg.default_staging_root()),
         )
 
     # ── Welcome header ────────────────────────────────────────────────────
@@ -2541,6 +2550,12 @@ def _wire_callbacks(self) -> None:
         self.execution_backend_dd.observe(
             self._safe_cb(self._on_execution_backend_changed), names="value"
         )
+        self.slurm_job_root_txt.observe(
+            self._safe_cb(
+                lambda change: _slurm_on_job_root_changed(self, change["new"])
+            ),
+            names="value",
+        )
         self.quantum_engine_dd.observe(
             self._safe_cb(self._on_quantum_engine_changed), names="value"
         )
diff --git a/quantui/app_builders.py b/quantui/app_builders.py
index da7f522..270f317 100644
--- a/quantui/app_builders.py
+++ b/quantui/app_builders.py
@@ -128,6 +128,9 @@ def build_status_panel(
     execution_backend: str = "local",
     slurm_available: bool = False,
     quantum_engine: str = "auto",
+    slurm_job_root: str = "",
+    slurm_job_root_env_locked: bool = False,
+    slurm_job_root_effective: str = "",
 ) -> None:
     """Build the Status tab panel."""
     cores, mem_gb = get_session_resources_fn()
@@ -407,6 +410,44 @@ def _render_status(gpu_state: Any) -> str:
             "Use the Cluster Jobs tab to monitor and cancel runs.</div>"
         )
 
+    # SLURM job folder root (M-JOBDIRS JD.1). Built unconditionally so the
+    # app can wire it; only shown when SLURM is available.
+    app.slurm_job_root_txt = widgets.Text(
+        value=slurm_job_root_effective if slurm_job_root_env_locked else slurm_job_root,
+        placeholder="~/.quantui/staging (default)",
+        continuous_update=False,
+        disabled=slurm_job_root_env_locked,
+        layout=layout_fn(width="420px"),
+    )
+    if slurm_job_root_env_locked:
+        _root_note_text = (
+            "Set by <code>QUANTUI_STAGING_DIR</code> in the container or "
+            "session environment."
+        )
+    else:
+        _root_note_text = (
+            "Each cluster job gets its own folder here, with one subfolder per "
+            "attempt. Leave blank for the default; on clusters with a small "
+            "home quota, use a scratch or project folder. Applies to new "
+            "submissions."
+        )
+    app.slurm_job_root_note = widgets.HTML(
+        f'<div style="font-size:11px;color:{_theme.css.TEXT_SUBTLE};margin:2px 0 0 0">'
+        f"{_root_note_text}</div>"
+    )
+    slurm_root_rows: list[Any] = []
+    if slurm_available:
+        slurm_root_rows = [
+            widgets.HTML(
+                f'<div style="font-size:12px;color:{_theme.css.TEXT_SLATE_DARK};'
+                'margin-top:12px;margin-bottom:0px">SLURM job folder '
+                f'<span style="color:{_theme.css.TEXT_SUBTLE};font-size:11px">'
+                "(persists across launches)</span></div>"
+            ),
+            app.slurm_job_root_txt,
+            app.slurm_job_root_note,
+        ]
+
     fp_settings_rows: list[Any] = [
         fp_toggle_label,
         app.freq_parallel_enabled_cb,
@@ -431,6 +472,7 @@ def _render_status(gpu_state: Any) -> str:
             exec_backend_label,
             app.execution_backend_dd,
             exec_backend_note,
+            *slurm_root_rows,
         ],
         layout=layout_fn(margin="0 0 8px 0"),
     )
diff --git a/quantui/app_slurm.py b/quantui/app_slurm.py
index 52df04f..d221689 100644
--- a/quantui/app_slurm.py
+++ b/quantui/app_slurm.py
@@ -756,6 +756,46 @@ def on_slurm_reconnect_clicked(app: Any, _btn: Any = None) -> None:
         attach_slurm_job(app, request_id)
 
 
+def on_slurm_job_root_changed(app: Any, value: str) -> None:
+    """Validate and persist the SLURM job folder root (M-JOBDIRS JD.1).
+
+    Blank restores the default. A new root applies to new submissions;
+    existing jobs keep their absolute folder paths in the registry.
+    """
+    note = getattr(app, "slurm_job_root_note", None)
+
+    def _say(message: str, *, error: bool = False) -> None:
+        if note is None:
+            return
+        color = _theme.css.ACCENT_ERROR if error else _theme.css.TEXT_SUBTLE
+        note.value = (
+            f'<div style="font-size:11px;color:{color};margin:2px 0 0 0">'
+            f"{html.escape(message)}</div>"
+        )
+
+    if _cluster_cfg.staging_root_env_configured():
+        return
+    raw = (value or "").strip()
+    if raw:
+        root = Path(raw).expanduser()
+        if not root.is_absolute():
+            _say("Use a full path (for example /work/<you>/quantui-jobs).", error=True)
+            return
+        try:
+            root.mkdir(parents=True, exist_ok=True)
+        except OSError as exc:
+            _say(f"Cannot use that folder: {exc}", error=True)
+            return
+    if raw == app._user_settings.compute.slurm_job_root:
+        return
+    app._user_settings.compute.slurm_job_root = raw
+    app._user_settings.save()
+    # Rebuild the registry on next use so new jobs land under the new root.
+    app._job_registry = None
+    where = raw or str(_cluster_cfg.DEFAULT_STAGING_ROOT)
+    _say(f"New cluster jobs will be created in {where}.")
+
+
 def use_slurm_execution(app: Any) -> bool:
     pref = getattr(app._user_settings.compute, "execution_backend", "local")
     return pref == "slurm" and is_slurm_available()
diff --git a/quantui/backends/cluster_config.py b/quantui/backends/cluster_config.py
index 75777c9..e615e51 100644
--- a/quantui/backends/cluster_config.py
+++ b/quantui/backends/cluster_config.py
@@ -118,11 +118,32 @@ def default_jobs_root() -> Path:
     return Path.home() / ".quantui" / "jobs"
 
 
+DEFAULT_STAGING_ROOT = Path("~") / ".quantui" / "staging"
+
+
+def staging_root_env_configured() -> bool:
+    """True when ``QUANTUI_STAGING_DIR`` pins the job root (UI shows it locked)."""
+    return bool(os.environ.get("QUANTUI_STAGING_DIR"))
+
+
 def default_staging_root() -> Path:
+    """Root for SLURM job folders.
+
+    Precedence: ``QUANTUI_STAGING_DIR`` > the ``compute.slurm_job_root``
+    user setting (M-JOBDIRS JD.1) > ``~/.quantui/staging``.
+    """
     override = os.environ.get("QUANTUI_STAGING_DIR")
     if override:
         return Path(override).expanduser()
-    return Path.home() / ".quantui" / "staging"
+    try:
+        from quantui.user_settings import UserSettings
+
+        configured = UserSettings.load().compute.slurm_job_root
+    except Exception:  # noqa: BLE001 — a broken settings file must not block jobs
+        configured = ""
+    if configured:
+        return Path(configured).expanduser()
+    return DEFAULT_STAGING_ROOT.expanduser()
 
 
 # Apptainer image for batch workers (NCShare-oriented default path).
diff --git a/quantui/user_settings.py b/quantui/user_settings.py
index dea06b2..cf6dbad 100644
--- a/quantui/user_settings.py
+++ b/quantui/user_settings.py
@@ -115,6 +115,12 @@ class ComputeSettings:
     # (cluster). SLURM is only offered in the UI when ``sbatch`` is available.
     execution_backend: str = "local"
 
+    # Where SLURM job folders are created (M-JOBDIRS JD.1). Empty means the
+    # default ``~/.quantui/staging``; clusters with small home quotas should
+    # point this at scratch or a project directory. ``QUANTUI_STAGING_DIR``
+    # in the environment overrides it when set.
+    slurm_job_root: str = ""
+
     # Which quantum chemistry library evaluates integrals and runs SCF.
     # ``auto`` keeps PySCF when both engines are installed; native Windows with
     # only PyFock selects PyFock. UI wiring lands in PYF.4 — PYF.1 persists only.
@@ -289,6 +295,16 @@ def _from_dict(cls, data: object) -> UserSettings:
                     candidate_backend,
                     compute.execution_backend,
                 )
+        if "slurm_job_root" in compute_section:
+            candidate_root = compute_section["slurm_job_root"]
+            if isinstance(candidate_root, str):
+                compute.slurm_job_root = candidate_root.strip()
+            else:
+                _LOG.warning(
+                    "Invalid compute.slurm_job_root %r; using %r",
+                    candidate_root,
+                    compute.slurm_job_root,
+                )
         if "quantum_engine" in compute_section:
             candidate_engine = compute_section["quantum_engine"]
             if (
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index 8521923..c10495c 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -705,3 +705,78 @@ def test_resubmit_legacy_record_explains(self, app, backend):
         ok, message = resubmit_slurm_job(app_ns, "legacy")
         assert not ok
         assert "before per-job folders" in message
+
+
+# ---------------------------------------------------------------------------
+# Configurable job root (JD.1)
+# ---------------------------------------------------------------------------
+
+
+class TestJobRoot:
+    @pytest.fixture(autouse=True)
+    def _private_settings(self, tmp_path, monkeypatch):
+        # Keep writes out of the suite-wide settings file other tests read.
+        monkeypatch.setenv("QUANTUI_SETTINGS_PATH", str(tmp_path / "settings.json"))
+
+    def test_precedence_env_then_setting_then_default(self, tmp_path, monkeypatch):
+        from quantui.user_settings import UserSettings
+
+        monkeypatch.delenv("QUANTUI_STAGING_DIR", raising=False)
+        settings = UserSettings.load()
+        settings.compute.slurm_job_root = ""
+        settings.save()
+        assert cfg.default_staging_root() == cfg.DEFAULT_STAGING_ROOT.expanduser()
+
+        settings.compute.slurm_job_root = str(tmp_path / "scratch")
+        settings.save()
+        assert cfg.default_staging_root() == tmp_path / "scratch"
+
+        monkeypatch.setenv("QUANTUI_STAGING_DIR", str(tmp_path / "env"))
+        assert cfg.default_staging_root() == tmp_path / "env"
+        assert cfg.staging_root_env_configured()
+
+    def test_setting_round_trips(self):
+        from quantui.user_settings import UserSettings
+
+        data = UserSettings().to_dict()
+        data["compute"]["slurm_job_root"] = "  /work/me/jobs "
+        assert UserSettings._from_dict(data).compute.slurm_job_root == "/work/me/jobs"
+        data["compute"]["slurm_job_root"] = 5
+        assert UserSettings._from_dict(data).compute.slurm_job_root == ""
+
+    def _app(self):
+        from types import SimpleNamespace
+
+        from quantui.user_settings import UserSettings
+
+        return SimpleNamespace(
+            _user_settings=UserSettings.load(),
+            slurm_job_root_note=SimpleNamespace(value=""),
+            _job_registry=object(),
+        )
+
+    def test_handler_saves_absolute_path_and_resets_registry(
+        self, tmp_path, monkeypatch
+    ):
+        from quantui.app_slurm import on_slurm_job_root_changed
+        from quantui.user_settings import UserSettings
+
+        monkeypatch.delenv("QUANTUI_STAGING_DIR", raising=False)
+        app = self._app()
+        target = tmp_path / "new root"
+        on_slurm_job_root_changed(app, str(target))
+        assert target.is_dir()
+        assert UserSettings.load().compute.slurm_job_root == str(target)
+        assert app._job_registry is None
+        assert "New cluster jobs will be created" in app.slurm_job_root_note.value
+
+    def test_handler_rejects_relative_path(self, monkeypatch):
+        from quantui.app_slurm import on_slurm_job_root_changed
+        from quantui.user_settings import UserSettings
+
+        monkeypatch.delenv("QUANTUI_STAGING_DIR", raising=False)
+        before = UserSettings.load().compute.slurm_job_root
+        app = self._app()
+        on_slurm_job_root_changed(app, "relative/jobs")
+        assert "full path" in app.slurm_job_root_note.value
+        assert UserSettings.load().compute.slurm_job_root == before

From 66457b73ad1121cd7c0584782f48bc8e8d04381e Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:08:24 +0000
Subject: [PATCH 7/9] feat(slurm): optional job name field on the Calculate tab

M-JOBDIRS JD.2 (UI).

In SLURM mode the run panel shows an optional "Job name" field. It names
the job folder and the SLURM --job-name (sanitized: letters, digits, - and
_ kept, other characters become _). Blank keeps the default
<formula>_<calc>_<method>_<basis>. A name already in use gets _2, _3...,
and the field clears after a successful submit so the next job does not
reuse it by accident. The row is hidden in Local mode.

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/app.py          |  5 +++++
 quantui/app_builders.py | 26 ++++++++++++++++++++++++++
 quantui/app_slurm.py    |  9 ++++++++-
 tests/test_app_slurm.py |  2 +-
 tests/test_jobdirs.py   | 36 ++++++++++++++++++++++++++++++++++++
 5 files changed, 76 insertions(+), 2 deletions(-)

diff --git a/quantui/app.py b/quantui/app.py
index 2ed7c14..67a8757 100644
--- a/quantui/app.py
+++ b/quantui/app.py
@@ -1448,6 +1448,8 @@ class QuantUIApp:
         freq_parallel_enabled_cb: Any
         execution_backend_dd: Any
         slurm_job_root_txt: Any
+        _slurm_job_name_txt: Any
+        _slurm_job_name_row: Any
         slurm_job_root_note: Any
         quantum_engine_dd: Any
         quantum_engine_note: Any
@@ -2477,6 +2479,9 @@ def _sync_root_tab_layout(self) -> None:
         self._slurm_jobs_tab_index = (
             order.index("slurm_jobs") if "slurm_jobs" in order else None
         )
+        self._slurm_job_name_row.layout.display = (
+            "flex" if "slurm_jobs" in order else "none"
+        )
         self._root_tab_order_cache = list(order)
 
         if "slurm_jobs" in order:
diff --git a/quantui/app_builders.py b/quantui/app_builders.py
index 270f317..183b2ec 100644
--- a/quantui/app_builders.py
+++ b/quantui/app_builders.py
@@ -1527,6 +1527,31 @@ def build_shared_widgets(
     # native phase (first gradient / Hessian) visibly advances.
     app._run_elapsed_lbl = widgets.HTML(value="")
 
+    # Optional job folder / SLURM job name (M-JOBDIRS JD.2). Shown only in
+    # SLURM mode; blank uses <formula>_<calc>_<method>_<basis>.
+    app._slurm_job_name_txt = widgets.Text(
+        value="",
+        placeholder="default: <formula>_<calc>_<method>_<basis>",
+        description="Job name:",
+        style={"description_width": "80px"},
+        layout=layout_fn(width="460px"),
+        tooltip=(
+            "Names this job's folder and its SLURM job. A name already in use "
+            "gets _2, _3, ... appended."
+        ),
+    )
+    app._slurm_job_name_row = widgets.VBox(
+        [
+            app._slurm_job_name_txt,
+            widgets.HTML(
+                f'<div style="font-size:11px;color:{_theme.css.TEXT_SUBTLE};'
+                'margin:0 0 6px 84px">Optional. Letters, digits, - and _ are kept; '
+                "other characters become _.</div>"
+            ),
+        ],
+        layout=layout_fn(display="none"),
+    )
+
     app._slurm_job_banner = widgets.HTML(value="", layout=layout_fn(display="none"))
     app._slurm_reconnect_btn = widgets.Button(
         description="View SLURM progress",
@@ -2036,6 +2061,7 @@ def build_run_section(app: Any, *, layout_fn: Any) -> None:
             app.perf_estimate_html,
             app._resume_notice_html,
             app._resume_cb,
+            app._slurm_job_name_row,
             app._slurm_job_banner,
             widgets.HBox(
                 [app._slurm_reconnect_btn],
diff --git a/quantui/app_slurm.py b/quantui/app_slurm.py
index d221689..5fdfa36 100644
--- a/quantui/app_slurm.py
+++ b/quantui/app_slurm.py
@@ -170,8 +170,11 @@ def submit_slurm_run(app: Any) -> None:
     app.run_btn.disabled = True
     app.run_status.value = "Submitting to SLURM…"
 
+    name_widget = getattr(app, "_slurm_job_name_txt", None)
+    job_name = (str(name_widget.value).strip() or None) if name_widget else None
+
     try:
-        request_id = backend.dispatch(request)
+        request_id = backend.dispatch(request, job_name=job_name)
     except SecurityError as exc:
         _reset_run_ui_after_submit_failure(app)
         app.run_status.value = str(exc)
@@ -185,6 +188,10 @@ def submit_slurm_run(app: Any) -> None:
         _append_run_html(app, format_error_html(str(exc)))
         return
 
+    if name_widget is not None:
+        # A name belongs to one job; clear it so the next submit does not
+        # silently reuse it (it would get a _2 suffix).
+        name_widget.value = ""
     refresh_slurm_jobs_tab(app)
     _update_slurm_jobs_tab_title(app)
 
diff --git a/tests/test_app_slurm.py b/tests/test_app_slurm.py
index fbec24a..30bd2ae 100644
--- a/tests/test_app_slurm.py
+++ b/tests/test_app_slurm.py
@@ -252,7 +252,7 @@ def test_security_error_resets_run_ui(
         self, mock_backend_for_app, mock_build_request, _mock_slurm, tmp_path
     ):
         backend = SimpleNamespace(
-            dispatch=lambda _req: (_ for _ in ()).throw(
+            dispatch=lambda _req, **_kw: (_ for _ in ()).throw(
                 SecurityError("Concurrent job limit reached (2/2).")
             )
         )
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index c10495c..6f5ed97 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -780,3 +780,39 @@ def test_handler_rejects_relative_path(self, monkeypatch):
         on_slurm_job_root_changed(app, "relative/jobs")
         assert "full path" in app.slurm_job_root_note.value
         assert UserSettings.load().compute.slurm_job_root == before
+
+
+# ---------------------------------------------------------------------------
+# Job name field on the Calculate tab (JD.2 UI)
+# ---------------------------------------------------------------------------
+
+
+@patch("quantui.app_slurm.threading.Thread")
+@patch("quantui.app_slurm.is_slurm_available", return_value=True)
+def test_submit_passes_job_name_and_clears_field(_avail, _thread, backend, monkeypatch):
+    from types import SimpleNamespace
+
+    from quantui.app_slurm import submit_slurm_run
+
+    monkeypatch.setattr("quantui.app_slurm.slurm_backend_for_app", lambda _app: backend)
+    monkeypatch.setattr(
+        "quantui.app_slurm.build_calculation_request", lambda _app: _request("ui1")
+    )
+    monkeypatch.setattr("quantui.app_slurm.calc_type_key_from_app", lambda _app: "sp")
+    monkeypatch.setattr("quantui.app_slurm._SUPPORTED_SLURM_CALC_TYPES", {"sp"})
+    monkeypatch.setattr("quantui.app_slurm.slurm_submit_block_reason", lambda _a: None)
+    app = _jobs_tab_app(
+        backend.registry,
+        _slurm_job_name_txt=SimpleNamespace(value="  water scan #1 "),
+        _calc_running=False,
+        run_status=_Value(),
+        run_output=_Sink(),
+        run_btn=SimpleNamespace(disabled=False),
+        cancel_btn=SimpleNamespace(disabled=True),
+        log_clear_btn=SimpleNamespace(disabled=False),
+    )
+    with _sbatch_returning("321"):
+        submit_slurm_run(app)
+    rec = backend.registry.load("ui1")
+    assert Path(rec.job_dir).name == "water_scan_1"
+    assert app._slurm_job_name_txt.value == ""

From 15d822772e7b3332faf8184bba96a6fe00d326c2 Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:09:24 +0000
Subject: [PATCH 8/9] feat(slurm): descriptive names for shareable files in
 attempt folders

M-JOBDIRS JD.7.

- In an attempt folder the worker renames the files people copy out,
  result.molden and trajectory.xyz, to <job name>.molden and
  <job name>_trajectory.xyz, and records the mapping in
  result.json["artifact_names"]. Files the app reads back (result.json,
  orbitals.npz, progress.json, live.log) keep fixed names.
- Ingest maps the descriptive names back, so History entries keep the
  fixed names their loaders expect.
- write_worker_result is now atomic (tmp file + os.replace): a finished
  result.json marks a successful attempt, and a Cluster Jobs refresh may
  scan for it while the worker is still writing.
- Legacy staging runs (no --attempt-dir) are unchanged.

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 quantui/backends/slurm_ingest.py   | 14 +++++---
 quantui/backends/worker.py         | 30 +++++++++++++++++
 quantui/backends/worker_payload.py |  7 +++-
 tests/test_jobdirs.py              | 54 ++++++++++++++++++++++++++++++
 4 files changed, 100 insertions(+), 5 deletions(-)

diff --git a/quantui/backends/slurm_ingest.py b/quantui/backends/slurm_ingest.py
index c38199c..409e97f 100644
--- a/quantui/backends/slurm_ingest.py
+++ b/quantui/backends/slurm_ingest.py
@@ -48,9 +48,15 @@ def slurm_provenance(record: JobRecord, staging: Path) -> dict[str, Any]:
     }
 
 
-def _copy_staging_sidecars(staging: Path, saved_dir: Path) -> None:
+def _copy_staging_sidecars(
+    staging: Path, saved_dir: Path, payload: dict[str, Any] | None = None
+) -> None:
+    # Attempt dirs may hold shareable files under descriptive names
+    # (result.json["artifact_names"], M-JOBDIRS JD.7); History keeps the
+    # fixed names its loaders expect.
+    renamed = (payload or {}).get("artifact_names") or {}
     for name in _STAGING_SIDECAR_FILES:
-        src = staging / name
+        src = staging / renamed.get(name, name)
         if src.exists():
             shutil.copy2(src, saved_dir / name)
 
@@ -141,7 +147,7 @@ def _ingest_frequency(
     ir = spectra.get("ir") or {}
     freqs = ir.get("frequencies_cm1")
     displacements = ir.get("displacements")
-    _copy_staging_sidecars(staging, saved_dir)
+    _copy_staging_sidecars(staging, saved_dir, payload)
     if not (saved_dir / "result.molden").exists() and freqs and displacements:
         mol_block = spectra.get("molecule") or {}
         atoms = mol_block.get("atoms") or []
@@ -225,7 +231,7 @@ def _ingest_with_sidecars(
     if calc_type := payload.get("calc_type"):
         if calc_type in ("geometry_opt", "pes_scan"):
             _copy_trajectory(staging, saved_dir, payload)
-    _copy_staging_sidecars(staging, saved_dir)
+    _copy_staging_sidecars(staging, saved_dir, payload)
     _finalize_history_entry(saved_dir)
     return saved_dir
 
diff --git a/quantui/backends/worker.py b/quantui/backends/worker.py
index bf8c6a0..75c00a0 100644
--- a/quantui/backends/worker.py
+++ b/quantui/backends/worker.py
@@ -283,6 +283,31 @@ def _maybe_run_preopt(
         return molecule, None
 
 
+# Files people copy out of a job folder get the job name as their stem
+# (M-JOBDIRS JD.7): fixed name -> suffix appended to the job name. Files the
+# app reads back (result.json, orbitals.npz, progress.json, live.log) keep
+# fixed names. Ingest maps these back via result.json["artifact_names"].
+_DESCRIPTIVE_ARTIFACTS = {
+    "result.molden": ".molden",
+    "trajectory.xyz": "_trajectory.xyz",
+}
+
+
+def _apply_descriptive_names(staging_dir: Path, stem: str) -> dict[str, str]:
+    """Rename shareable artifacts to ``<job name><suffix>``; return the map."""
+    renamed: dict[str, str] = {}
+    for fixed, suffix in _DESCRIPTIVE_ARTIFACTS.items():
+        src = staging_dir / fixed
+        if not src.exists():
+            continue
+        dest = staging_dir / f"{stem}{suffix}"
+        if dest.exists():
+            continue
+        src.rename(dest)
+        renamed[fixed] = dest.name
+    return renamed
+
+
 def _update_tracked_record(request: CalculationRequest, status: str, **kwargs) -> None:
     """Update the registry record, but only for the SLURM job it tracks.
 
@@ -781,6 +806,11 @@ def run_worker_request(
 
     elapsed = time.perf_counter() - t0
     payload = _build_payload(calc_type, outcome, staging_dir, request)
+    if attempt_dir is not None:
+        # Before result.json: its presence is what marks the attempt done.
+        names = _apply_descriptive_names(staging_dir, request_path.parent.name)
+        if names:
+            payload["artifact_names"] = names
     write_worker_result(staging_dir, payload)
 
     _update_tracked_record(request, "success", result_dir=str(staging_dir))
diff --git a/quantui/backends/worker_payload.py b/quantui/backends/worker_payload.py
index 3ded223..69ba23e 100644
--- a/quantui/backends/worker_payload.py
+++ b/quantui/backends/worker_payload.py
@@ -5,6 +5,7 @@
 from __future__ import annotations
 
 import json
+import os
 from pathlib import Path
 from typing import Any, Dict
 
@@ -306,8 +307,12 @@ def reorg_result_payload(result) -> Dict[str, Any]:
 
 
 def write_worker_result(staging_dir: Path, payload: Dict[str, Any]) -> Path:
+    # Atomic: a finished result.json marks a successful attempt, and the app
+    # may scan for it while the worker is writing (M-JOBDIRS JD.11).
     result_path = staging_dir / "result.json"
-    result_path.write_text(json.dumps(payload, indent=2))
+    tmp_path = staging_dir / "result.json.tmp"
+    tmp_path.write_text(json.dumps(payload, indent=2))
+    os.replace(tmp_path, result_path)
     return result_path
 
 
diff --git a/tests/test_jobdirs.py b/tests/test_jobdirs.py
index 6f5ed97..50dd6c1 100644
--- a/tests/test_jobdirs.py
+++ b/tests/test_jobdirs.py
@@ -816,3 +816,57 @@ def test_submit_passes_job_name_and_clears_field(_avail, _thread, backend, monke
     rec = backend.registry.load("ui1")
     assert Path(rec.job_dir).name == "water_scan_1"
     assert app._slurm_job_name_txt.value == ""
+
+
+# ---------------------------------------------------------------------------
+# Descriptive names for shareable files (JD.7)
+# ---------------------------------------------------------------------------
+
+
+class TestDescriptiveNames:
+    def test_worker_renames_shareable_files(self, tmp_path):
+        from quantui.backends.worker import _apply_descriptive_names
+
+        (tmp_path / "result.molden").write_text("m")
+        (tmp_path / "trajectory.xyz").write_text("x")
+        (tmp_path / "orbitals.npz").write_text("o")
+        names = _apply_descriptive_names(tmp_path, "H2_opt_B3LYP_def2-SVP")
+        assert names == {
+            "result.molden": "H2_opt_B3LYP_def2-SVP.molden",
+            "trajectory.xyz": "H2_opt_B3LYP_def2-SVP_trajectory.xyz",
+        }
+        assert (tmp_path / "H2_opt_B3LYP_def2-SVP.molden").read_text() == "m"
+        assert (tmp_path / "orbitals.npz").exists()  # read back by the app
+
+    def test_ingest_restores_fixed_names_in_history(
+        self, tmp_path, monkeypatch, registry
+    ):
+        from quantui.backends.slurm_ingest import ingest_staging_success
+        from tests.slurm_ingest_helpers import patch_results_root, sample_payload
+
+        patch_results_root(tmp_path, monkeypatch)
+        rec = registry.create(_request(), "cluster_slurm", job_name="run")
+        attempt = Path(rec.job_dir) / "attempt-01_job1"
+        attempt.mkdir()
+        (attempt / "run.molden").write_text("[Molden Format]\n")
+        payload = sample_payload("single_point")
+        payload["artifact_names"] = {"result.molden": "run.molden"}
+        (attempt / "result.json").write_text(json.dumps(payload))
+        saved = ingest_staging_success(rec, attempt_dir=attempt)
+        assert (saved / "result.molden").read_text() == "[Molden Format]\n"
+        assert (attempt / "run.molden").exists()  # the job folder copy stays
+
+    def test_legacy_worker_run_keeps_fixed_names(self, tmp_path, monkeypatch):
+        """No --attempt-dir (legacy staging) → no renaming, as before."""
+        from quantui.backends.worker import run_worker_request
+
+        monkeypatch.setenv("QUANTUI_JOBS_DIR", str(tmp_path / "jobs"))
+        monkeypatch.setenv("QUANTUI_STAGING_DIR", str(tmp_path / "staging"))
+        staging = tmp_path / "staging" / "leg"
+        staging.mkdir(parents=True)
+        (staging / "request.json").write_text(
+            json.dumps(_request("leg", calc_type="nope").to_dict())
+        )
+        (staging / "result.molden").write_text("m")
+        run_worker_request(staging / "request.json")
+        assert (staging / "result.molden").exists()

From eee00c284a735af95ed5a9ae4be6421749378804 Mon Sep 17 00:00:00 2001
From: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Date: Sat, 26 Sep 2026 05:10:19 +0000
Subject: [PATCH 9/9] docs(slurm): document job folders, attempts, Resubmit and
 the job root

M-JOBDIRS docs: job folder layout and rerun behavior in
apptainer/slurm/README.md (plus the QUANTUI_STAGING_DIR row and the
worker's --attempt-dir flag), the `quantui submit --job-name` help text,
and CHANGELOG entries under Unreleased.

Contributions:
- Claude (Opus 5.5): code edits, review, and conceptual discussion
- Jonathan Schultz: overall vision, planning, review, and orchestration

Co-authored-by: Jonathan Schultz <nccu-schultz-lab@users.noreply.github.com>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
---
 CHANGELOG.md              | 19 +++++++++++++++++++
 apptainer/slurm/README.md | 39 +++++++++++++++++++++++++++++++++++++--
 quantui/cli.py            |  5 ++++-
 3 files changed, 60 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 84fb7db..3e0a915 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -9,6 +9,18 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
 
 ### Added
 
+- **SLURM job folders with per-attempt subfolders (M-JOBDIRS)** — each cluster
+  job gets a readable folder (`<formula>_<calc>_<method>_<basis>`, or the
+  optional **Job name** on the Calculate tab) that doubles as the SLURM job
+  name. Every run of its `submit.slurm` creates a new `attempt-NN_job<id>/`
+  subfolder, so **Resubmit** (new, Cluster Jobs tab) and a hand-run
+  `sbatch submit.slurm` never overwrite an earlier attempt. The job root is
+  configurable in **System Settings** (or `QUANTUI_STAGING_DIR`), for
+  clusters with small home quotas.
+- **SLURM results marked in History** — entries show
+  `🖥 SLURM <job id>·a<attempt>` and the result card has a "Ran on" row with
+  the job folder. Hand-run attempts are added to History on the next Cluster
+  Jobs refresh; failed attempts stay in the job folder only.
 - **PyFock geometry and analysis phase** — native-Windows PBE/def2 geometry
   optimization now uses PyFock's ASE calculator with analytical density-fitted
   gradients. Single-point and final optimized-geometry results retain orbital
@@ -21,6 +33,13 @@ and this project follows [Semantic Versioning](https://semver.org/spec/v2.0.0.ht
   parity gate against PySCF. Unsupported hybrids, ions, open-shell systems,
   solvent, checkpoints, GPU, and orbital analysis are rejected before compute.
 
+### Fixed
+
+- Reconnecting to a finished SLURM job no longer saves a second copy of its
+  result to History.
+- Removed the unused `QUANTUI_RESULTS_DIR` export from generated SLURM
+  scripts (it only created an empty `results/` folder per job).
+
 ## [0.8.2] - 2026-08-30
 
 ### Added
diff --git a/apptainer/slurm/README.md b/apptainer/slurm/README.md
index 8975805..1e2b446 100644
--- a/apptainer/slurm/README.md
+++ b/apptainer/slurm/README.md
@@ -9,7 +9,37 @@ Reference scripts for operator-driven NCShare / Apptainer batch jobs.
 | `quantui-batch.sbatch` | Reference `sbatch` template (partition must be set) |
 | `quantui-gpu-test.sbatch` | GPU smoke test (operator use) |
 
-QuantUI generates per-job scripts automatically under `~/.quantui/staging/<request_id>/submit.slurm` when students submit from the UI.
+QuantUI generates per-job scripts automatically when students submit from the UI.
+
+## Job folders
+
+Each submission gets its own folder under the job root (default
+`~/.quantui/staging`; set it in **System Settings → SLURM job folder** or with
+`QUANTUI_STAGING_DIR`):
+
+```
+<job root>/H2O_opt_B3LYP_def2-SVP/        # job name = SLURM --job-name
+    request.json   submit.slurm
+    slurm-812345.out  slurm-812345.err   # SLURM's own output, one pair per job
+    attempt-01_job812345/                # first run
+        live.log  progress.json  ...
+    attempt-02_job812399/                # a rerun, never overwrites attempt-01
+        result.json  orbitals.npz  H2O_opt_B3LYP_def2-SVP.molden  ...
+    latest -> attempt-02_job812399
+```
+
+- The job name defaults to `<formula>_<calc>_<method>_<basis>`; the optional
+  **Job name** field on the Calculate tab (or `quantui submit --job-name`)
+  overrides it. A name already in use gets `_2`, `_3`, ....
+- `submit.slurm` creates a new `attempt-NN_job<SLURM id>/` folder every time it
+  runs, so both **Resubmit** on the Cluster Jobs tab and a hand-run
+  `sbatch submit.slurm` from the job folder start a fresh attempt. Earlier
+  attempts are never overwritten, and checkpoints let a rerun resume.
+- Every successful attempt appears in **History**, marked `🖥 SLURM <job id>·a<attempt>`,
+  including hand-run ones (picked up on the next Cluster Jobs refresh).
+  Failed attempts stay in the job folder only.
+- Jobs submitted before per-job folders existed keep their
+  `~/.quantui/staging/<request_id>/` layout and cannot be resubmitted in place.
 
 Status polling uses **`squeue`** for active jobs and **`sacct`** for terminal state, exit code, and cancel confirmation. Cluster Jobs **Remove** clears terminal registry rows without deleting staging logs.
 
@@ -26,6 +56,7 @@ See the [NCShare SLURM batch runbook](https://github.com/The-Schultz-Lab/QuantUI
 | `QUANTUI_SLURM_CANCEL_CONFIRM_S` | `30` | Seconds to wait for `scancel` confirmation via `sacct` |
 | `QUANTUI_SLURM_PARTITION` | `common` | Default `#SBATCH` partition |
 | `QUANTUI_BATCH_IMAGE` | `~/quantui-gpu.sif` | Apptainer image for batch worker |
+| `QUANTUI_STAGING_DIR` | *(unset — System Settings value, else `~/.quantui/staging`)* | Job folder root. Overrides the System Settings value and locks that field. A root outside `$HOME` is bound into Apptainer automatically. |
 
 ## Operator runbook
 
@@ -36,7 +67,11 @@ See the planning repo runbook:
 ## Worker entrypoint
 
 ```bash
-python -m quantui.backends.worker --request /path/to/staging/request.json
+python -m quantui.backends.worker --request /path/to/job/request.json \
+    --attempt-dir /path/to/job/attempt-01_job812345
 ```
 
+`submit.slurm` passes `--attempt-dir` itself. Without it, outputs go next to
+`request.json` (the pre-job-folder layout).
+
 Supported calc types: `single_point`, `geometry_opt`, `frequency`, `tddft`, `nmr`, `pes_scan`, `reorganization_energy`.
diff --git a/quantui/cli.py b/quantui/cli.py
index 747f6ca..10702f4 100644
--- a/quantui/cli.py
+++ b/quantui/cli.py
@@ -552,7 +552,10 @@ def _build_parser() -> argparse.ArgumentParser:
         type=str,
         default=None,
         metavar="NAME",
-        help="SLURM job name (default: derived from the molecule + method).",
+        help=(
+            "Name for the job folder and SLURM job (default: "
+            "<formula>_<calc>_<method>_<basis>)."
+        ),
     )
     submit_parser.add_argument(
         "--depends-on",