From 3596fde77c924975b1093e1860f603ef33a3d3c1 Mon Sep 17 00:00:00 2001
From: Aljes <aleksandr.binkevich@gmail.com>
Date: Thu, 15 Oct 2026 14:18:11 +0100
Subject: [PATCH 1/2] subset, recast, split: write beside the output, rename
 into place

Fixes #35 and the other non-atomic writes found alongside it.

- subset: resolve the selection before the output exists, refuse an
  output that names the input (on Zarr it emptied the store), and write
  through staged_outputs, so a failed query leaves nothing and an
  existing output survives a failure.
- subset/recast --inplace: the source was deleted before the new store
  was moved in; it is now renamed aside and deleted only afterwards.
  recast -o is staged too.
- split: stage every output, write the manifest last, refuse an output
  that lands on the input.
- staged_outputs: refuse a leftover *.adata-old.* rather than deleting
  what may be the only copy, put back already-swapped outputs if a
  rename fails, and keep a suffixless Zarr store's temporary a Zarr store.
- same_path moves to util/path.py; convert's copy answered "different"
  when it could not tell. Drop recast's unused ALL_PREFIXES.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 docs/COMMANDS.md              |  16 ++-
 src/adata/commands/convert.py |  10 +-
 src/adata/commands/recast.py  | 125 ++++++++-----------
 src/adata/commands/split.py   |  43 ++++---
 src/adata/core/subset.py      |  61 +++++-----
 src/adata/util/path.py        |  10 ++
 src/adata/util/staging.py     |  76 +++++++++---
 tests/test_convert.py         |  31 -----
 tests/test_recast.py          |   2 +-
 tests/test_staging.py         | 113 +++++++++++++++++
 tests/test_write_safety.py    | 222 ++++++++++++++++++++++++++++++++++
 11 files changed, 526 insertions(+), 183 deletions(-)
 create mode 100644 tests/test_staging.py
 create mode 100644 tests/test_write_safety.py

diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md
index 03255df..8c1df1a 100644
--- a/docs/COMMANDS.md
+++ b/docs/COMMANDS.md
@@ -83,6 +83,13 @@ adata subset data.h5ad --inplace --obs barcodes.txt
 `raw/` is carried over and matched against its **own** var axis, which usually
 holds more genes than the main object.
 
+The selection is resolved before anything is written, and the output is
+written beside its destination (`out.adata-tmp.h5ad`) and moved into place
+only once complete. A query that fails therefore leaves nothing behind, and an
+existing output (or, with `--inplace`, the source) is replaced only once its
+successor is finished. The output path may not name the input; use
+`--inplace`.
+
 ## `convert`
 
 Change a store's format. The output's extension picks the format; `--to`
@@ -225,8 +232,9 @@ bigger than it started. `indices` and `indptr` keep their own widths, which
 can differ: a narrow matrix with more than 2^31 nonzeros needs int64 offsets
 over int32 coordinates, and both are range-checked before writing.
 
-The output path may not name the input; use `--inplace`, which writes to a
-temporary file and swaps it in only once the conversion has finished.
+The output path may not name the input; use `--inplace`. Either way the new
+store is written beside its destination and moved into place only once the
+conversion has finished, so a failure leaves the original untouched.
 
 Transposing streams by default and works on matrices too large to load, at
 the cost of two extra passes over the nonzeros. `--in-memory` is faster when
@@ -256,6 +264,10 @@ adata split data.h5ad --by cluster -o out/ --min-size 50
 | `--manifest / --no-manifest` | Write a CSV listing the outputs (default on) |
 
 Labels are sanitised for use as filenames; collisions get a numeric suffix.
+Every store is written beside its destination and moved into place only once
+all of them are complete, and the manifest is written last, so a split that
+fails leaves no partial stores. An output that would land on the input is
+refused.
 
 ## `concat`
 
diff --git a/src/adata/commands/convert.py b/src/adata/commands/convert.py
index 785453d..2e859ae 100644
--- a/src/adata/commands/convert.py
+++ b/src/adata/commands/convert.py
@@ -20,16 +20,10 @@
     detect_format,
     output_format,
 )
+from adata.util.path import same_path
 from adata.util.staging import staged_outputs
 
 
-def _same_path(left: Path, right: Path) -> bool:
-    try:
-        return left.resolve() == right.resolve()
-    except OSError:
-        return False
-
-
 def convert_store(
     file: Path,
     output: Path,
@@ -56,7 +50,7 @@ def convert_store(
 
     plan = converter.plan(file, output, src_format, dst_format, options)
     for out in plan.outputs:
-        if _same_path(file, out):
+        if same_path(file, out):
             # Writing a Zarr store over its own source empties it before a
             # single element has been read.
             raise ValueError(f"Output path is the input: {out}")
diff --git a/src/adata/commands/recast.py b/src/adata/commands/recast.py
index f07d431..3666b26 100644
--- a/src/adata/commands/recast.py
+++ b/src/adata/commands/recast.py
@@ -2,13 +2,14 @@
 
 Writes a whole new store rather than editing one in place, for the same
 reason `subset` does: a conversion that fails half way leaves the original
-untouched, and `--inplace` becomes an atomic swap of a finished file rather
-than a partial edit of a live one.
+untouched. The new store is written beside its destination and renamed into
+place only once complete (see `adata.util.staging`), so `--inplace` replaces a
+finished file rather than editing a live one, and the original is deleted only
+after its successor is in place.
 """
 
 from __future__ import annotations
 
-import shutil
 from pathlib import Path
 from typing import Any, List, Optional
 
@@ -25,15 +26,16 @@
 from adata.elements import spec
 from adata.elements.write import ensure_anndata_skeleton
 from adata.storage import copy_tree, detect_backend, is_group, open_store
-
-#: What `--all` means. The matrices an AnnData object is built from, not
-#: every 2-D array in the file: obsm/varm hold embeddings, whose dtype is
-#: rarely what anyone is trying to shrink. Those still convert by path.
-ALL_PREFIXES = ("X", "layers", "raw/X")
-
+from adata.util.path import same_path
+from adata.util.staging import staged_outputs
 
 def discover_matrices(root: Any) -> List[str]:
-    """Entry paths `--all` resolves to, in a stable order."""
+    """Entry paths `--all` resolves to, in a stable order.
+
+    The matrices an AnnData object is built from, not every 2-D array in the
+    file: obsm/varm hold embeddings, whose dtype is rarely what anyone is
+    trying to shrink. Those still convert by path.
+    """
     found: List[str] = []
     if "X" in root:
         found.append("X")
@@ -44,14 +46,6 @@ def discover_matrices(root: Any) -> List[str]:
     return found
 
 
-def _same_path(left: Path, right: Path) -> bool:
-    """Do these name the same store, through symlinks and `..` alike?"""
-    try:
-        return left.resolve() == right.resolve()
-    except OSError:  # pragma: no cover - unresolvable path
-        return left.absolute() == right.absolute()
-
-
 def _resolve(root: Any, path: str) -> Any:
     obj = root
     for part in path.split("/"):
@@ -96,7 +90,7 @@ def recast_store(
         else None
     )
 
-    if output is not None and not inplace and _same_path(file, output):
+    if output is not None and not inplace and same_path(file, output):
         # Opening the destination "w" clears it while the source is still
         # being read from it. HDF5 refuses; Zarr does not, and quietly
         # produced an empty store where the data used to be.
@@ -105,69 +99,54 @@ def recast_store(
             "it, which writes to a temporary file first."
         )
 
-    if inplace:
-        backend = detect_backend(file)
-        if backend == "zarr":
-            base = file.stem if file.suffix else file.name
-            dst_path = file.with_name(f"{base}.recast-tmp.zarr")
-        else:
-            dst_path = file.with_name(f"{file.name}.recast-tmp")
-        if dst_path.exists():
-            raise FileExistsError(f"Temporary path already exists: {dst_path}")
-    else:
-        dst_path = output
+    dst_path = file if inplace else Path(output)
 
     if zarr_format is None and detect_backend(file) == "zarr":
         with open_store(file, "r") as probe:
             zarr_format = probe.zarr_format
 
-    with open_store(file, "r") as src_store:
-        src = src_store.root
-        targets = discover_matrices(src) if convert_all else list(entries or [])
-        if not targets:
-            raise ValueError(
-                "No matrices selected. Name one (e.g. `X`) or pass --all."
-            )
-        # Every check first, before the destination exists. A refusal must
-        # not leave behind a store holding a copy of obs and var and nothing
-        # else -- the caller cannot tell that from a finished conversion.
-        plans = {
-            path: plan_conversion(
-                _resolve(src, path),
-                path,
-                dtype=data_dtype,
-                index_dtype=index_dtype,
-                layout=layout,
-                chunk=chunk,
-                force=force,
-                console=console,
+    # The source is opened inside the staging block, so it is closed before
+    # `--inplace` renames it aside.
+    with staged_outputs([dst_path]) as (temp,):
+        with open_store(file, "r") as src_store:
+            src = src_store.root
+            targets = discover_matrices(src) if convert_all else list(entries or [])
+            if not targets:
+                raise ValueError(
+                    "No matrices selected. Name one (e.g. `X`) or pass --all."
+                )
+            # Every check first, before the destination exists. A refusal must
+            # not leave behind a store holding a copy of obs and var and nothing
+            # else -- the caller cannot tell that from a finished conversion.
+            plans = {
+                path: plan_conversion(
+                    _resolve(src, path),
+                    path,
+                    dtype=data_dtype,
+                    index_dtype=index_dtype,
+                    layout=layout,
+                    chunk=chunk,
+                    force=force,
+                    console=console,
+                )
+                for path in targets
+            }
+
+            console.print(
+                f"[cyan]Converting {len(targets)} matrix/matrices:[/] "
+                + ", ".join(targets)
             )
-            for path in targets
-        }
-
-        console.print(
-            f"[cyan]Converting {len(targets)} matrix/matrices:[/] "
-            + ", ".join(targets)
-        )
 
-        with open_store(dst_path, "w", zarr_format=zarr_format) as dst_store:
-            dst = dst_store.root
-            _write(
-                src, dst, plans,
-                chunk=chunk, chunk_rows=chunk_rows, in_memory=in_memory,
-                console=console,
-            )
-            ensure_anndata_skeleton(dst)
+            with open_store(temp, "w", zarr_format=zarr_format) as dst_store:
+                dst = dst_store.root
+                _write(
+                    src, dst, plans,
+                    chunk=chunk, chunk_rows=chunk_rows, in_memory=in_memory,
+                    console=console,
+                )
+                ensure_anndata_skeleton(dst)
 
     if inplace:
-        if file.is_dir():
-            shutil.rmtree(file)
-        elif file.exists():
-            file.unlink()
-        if dst_path.is_dir():
-            shutil.move(str(dst_path), str(file))
-        else:
-            dst_path.replace(file)
         console.print(f"[green]Converted[/] {file}")
     else:
         console.print(f"[green]Wrote[/] {dst_path}")
diff --git a/src/adata/commands/split.py b/src/adata/commands/split.py
index 8ce4403..c172769 100644
--- a/src/adata/commands/split.py
+++ b/src/adata/commands/split.py
@@ -4,6 +4,11 @@
 streaming: the source is never loaded into memory, only the column being split
 on is read, and each output is produced by the same subset machinery used by
 `adata subset`, fanned out so every matrix is read once for all groups.
+
+Every output is written beside its destination and renamed into place only
+once all of them are complete, so a split that fails leaves no partial stores
+(and existing outputs as they were). The manifest is written last, so it
+exists only when everything it lists does.
 """
 
 from __future__ import annotations
@@ -19,6 +24,8 @@
 from adata.core.subset import split_h5ad
 from adata.storage import detect_backend, open_store
 from adata.util.names import unique_names
+from adata.util.path import same_path
+from adata.util.staging import staged_outputs
 
 
 def split_store(
@@ -78,6 +85,13 @@ def split_store(
     for label, out_path, count in planned:
         console.print(f"  {label!r} -> {out_path} ({count} {axis})")
 
+    for _, out_path, _ in planned:
+        if same_path(file, out_path):
+            raise ValueError(
+                f"Output path is the input: {out_path}. Choose another "
+                "--output-dir."
+            )
+
     if dry_run:
         console.print("[yellow]Dry run: nothing written.[/]")
         return planned
@@ -86,20 +100,21 @@ def split_store(
 
     # One call for every group, so each matrix is read once rather than once
     # per group.
-    split_h5ad(
-        file,
-        [
-            (
-                out_path,
-                np.sort(groups[label]) if axis == "obs" else None,
-                np.sort(groups[label]) if axis == "var" else None,
-            )
-            for label, out_path, _ in planned
-        ],
-        chunk_rows=chunk_rows,
-        console=console,
-        zarr_format=zarr_format,
-    )
+    with staged_outputs([out_path for _, out_path, _ in planned]) as temps:
+        split_h5ad(
+            file,
+            [
+                (
+                    temp,
+                    np.sort(groups[label]) if axis == "obs" else None,
+                    np.sort(groups[label]) if axis == "var" else None,
+                )
+                for (label, _, _), temp in zip(planned, temps)
+            ],
+            chunk_rows=chunk_rows,
+            console=console,
+            zarr_format=zarr_format,
+        )
 
     if manifest:
         _write_manifest(file, output_dir, column, planned, console)
diff --git a/src/adata/core/subset.py b/src/adata/core/subset.py
index e6081ab..65ae03e 100644
--- a/src/adata/core/subset.py
+++ b/src/adata/core/subset.py
@@ -5,7 +5,6 @@
 from contextlib import ExitStack
 from dataclasses import dataclass
 from pathlib import Path
-import shutil
 from typing import Optional, Set, Tuple, List, Dict, Any
 
 import numpy as np
@@ -43,6 +42,8 @@
     with_compression,
     zarr_format_of,
 )
+from adata.util.path import same_path
+from adata.util.staging import staged_outputs
 
 
 def _target_backend(dst_group: Any) -> str:
@@ -952,18 +953,15 @@ def subset_h5ad(
     if not inplace and output is None:
         raise ValueError("Output file is required unless --inplace is specified.")
 
-    if inplace:
-        src_backend = detect_backend(file)
-        if src_backend == "zarr":
-            base_name = file.stem if file.suffix else file.name
-            tmp_path = file.with_name(f"{base_name}.subset-tmp.zarr")
-        else:
-            tmp_path = file.with_name(f"{file.name}.subset-tmp")
-        if tmp_path.exists():
-            raise FileExistsError(f"Temporary path already exists: {tmp_path}")
-        dst_path = tmp_path
-    else:
-        dst_path = output
+    if not inplace and same_path(file, output):
+        # Opening the output "w" would empty a Zarr source before a single
+        # element had been read from it.
+        raise ValueError(
+            f"Output path is the input: {output}. Use --inplace to replace "
+            "it, which writes to a temporary file first."
+        )
+
+    dst_path = file if inplace else Path(output)
 
     if zarr_format is None and detect_backend(file) == "zarr":
         with open_store(file, "r") as probe:
@@ -975,12 +973,17 @@ def subset_h5ad(
     status = console.status("[magenta]Opening files...[/]")
     status.start()
     try:
-        with open_store(file, "r") as src_store, open_store(
-            dst_path, "w", zarr_format=zarr_format
-        ) as dst_store:
+        # The output is written beside its destination and renamed into place
+        # only once complete, so a failure leaves any existing output (or,
+        # with --inplace, the source) as it was. The source is opened inside
+        # the staging block, so it is closed before it is renamed aside.
+        with staged_outputs([dst_path]) as (temp,), open_store(
+            file, "r"
+        ) as src_store:
             src = src_store.root
-            dst = dst_store.root
 
+            # The selection is resolved before the output exists: a query
+            # that fails must not leave an empty store behind.
             obs_idx = (
                 obs_indices
                 if obs_indices is not None
@@ -1000,26 +1003,16 @@ def subset_h5ad(
 
             status.stop()
 
-            _write_targets(
-                src,
-                [Target(dst, obs_idx, var_idx, var_keep)],
-                chunk_rows=chunk_rows,
-                console=console,
-            )
+            with open_store(temp, "w", zarr_format=zarr_format) as dst_store:
+                _write_targets(
+                    src,
+                    [Target(dst_store.root, obs_idx, var_idx, var_keep)],
+                    chunk_rows=chunk_rows,
+                    console=console,
+                )
     finally:
         status.stop()
 
-    if inplace:
-        if file.exists():
-            if file.is_dir():
-                shutil.rmtree(file)
-            else:
-                file.unlink()
-        if dst_path.is_dir():
-            shutil.move(str(dst_path), str(file))
-        else:
-            dst_path.replace(file)
-
 
 def _open_output_limit() -> int:
     """How many output stores `split_h5ad` may hold open at once.
diff --git a/src/adata/util/path.py b/src/adata/util/path.py
index c5c7102..24d3832 100644
--- a/src/adata/util/path.py
+++ b/src/adata/util/path.py
@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+from pathlib import Path
+
 
 def norm_path(path: str) -> str:
     """Normalize object paths used inside h5ad/zarr stores."""
@@ -7,3 +9,11 @@ def norm_path(path: str) -> str:
     if not value:
         raise ValueError("Object path must be non-empty.")
     return value.lstrip("/")
+
+
+def same_path(left: Path, right: Path) -> bool:
+    """Do these name the same store, through symlinks and `..` alike?"""
+    try:
+        return Path(left).resolve() == Path(right).resolve()
+    except OSError:  # pragma: no cover - unresolvable path
+        return Path(left).absolute() == Path(right).absolute()
diff --git a/src/adata/util/staging.py b/src/adata/util/staging.py
index 2874a90..df2f7c7 100644
--- a/src/adata/util/staging.py
+++ b/src/adata/util/staging.py
@@ -1,10 +1,15 @@
 """Write outputs beside their destination, and move them into place at the end.
 
-A conversion that fails half way must not leave a half-written store where
-the output should be: the next command to open it would read whatever got
-written as if it were the whole thing. Writing to a temporary sibling and
-renaming it only on success means the output either appears whole or not at
-all, and an existing output is replaced only once its successor is complete.
+A command that fails half way must not leave a half-written store where the
+output should be: the next command to open it would read whatever got written
+as if it were the whole thing. Writing to a temporary sibling and renaming it
+only on success means the output either appears whole or not at all.
+
+An existing output is renamed aside (`out.adata-old.zarr`) before its
+successor is renamed in, and deleted only once every output is in place. If a
+rename fails part way, the outputs already swapped are put back. A run killed
+between the two renames leaves the old data under the `adata-old` name, never
+nowhere, and the next run refuses to start until that has been dealt with.
 """
 
 from __future__ import annotations
@@ -12,7 +17,7 @@
 import shutil
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Iterator, List, Sequence
+from typing import Iterator, List, Sequence, Tuple
 
 
 def _sibling(path: Path, tag: str) -> Path:
@@ -20,6 +25,15 @@ def _sibling(path: Path, tag: str) -> Path:
     return path.with_name(f"{path.stem}.{tag}{path.suffix}")
 
 
+def _temp_for(path: Path) -> Path:
+    temp = _sibling(path, "adata-tmp")
+    if path.is_dir() and path.suffix != ".zarr":
+        # A Zarr store named without the suffix. A temporary named the same
+        # way would not exist yet, and so be opened as HDF5.
+        temp = temp.with_name(f"{temp.name}.zarr")
+    return temp
+
+
 def _remove(path: Path) -> None:
     if path.is_dir() and not path.is_symlink():
         shutil.rmtree(path)
@@ -27,6 +41,10 @@ def _remove(path: Path) -> None:
         path.unlink()
 
 
+def _exists(path: Path) -> bool:
+    return path.exists() or path.is_symlink()
+
+
 @contextmanager
 def staged_outputs(outputs: Sequence[Path]) -> Iterator[List[Path]]:
     """Yield temporary paths to write `outputs` to; move them into place after.
@@ -36,12 +54,13 @@ def staged_outputs(outputs: Sequence[Path]) -> Iterator[List[Path]]:
     temporary paths are deleted and the outputs are left as they were.
     """
     outputs = [Path(p) for p in outputs]
-    temps = [_sibling(p, "adata-tmp") for p in outputs]
-    for temp in temps:
-        if temp.exists():
+    temps = [_temp_for(p) for p in outputs]
+    backups = [_sibling(p, "adata-old") for p in outputs]
+    for leftover in temps + backups:
+        if _exists(leftover):
             raise FileExistsError(
-                f"'{temp}' already exists, probably left by an interrupted run. "
-                "Remove it and try again."
+                f"'{leftover}' already exists, probably left by an interrupted "
+                "run. Check what it holds, remove it, and try again."
             )
 
     try:
@@ -51,13 +70,30 @@ def staged_outputs(outputs: Sequence[Path]) -> Iterator[List[Path]]:
             _remove(temp)
         raise
 
-    backups = []
-    for output, temp in zip(outputs, temps):
-        if output.exists():
-            backup = _sibling(output, "adata-old")
+    swapped: List[Tuple[Path, Path, bool]] = []
+    try:
+        for output, temp, backup in zip(outputs, temps, backups):
+            had_output = _exists(output)
+            if had_output:
+                output.rename(backup)
+            try:
+                temp.rename(output)
+            except BaseException:
+                if had_output:
+                    backup.rename(output)
+                raise
+            swapped.append((output, backup, had_output))
+    except BaseException:
+        # Put back what was already swapped, newest first, and drop the
+        # temporaries that never made it.
+        for output, backup, had_output in reversed(swapped):
+            _remove(output)
+            if had_output:
+                backup.rename(output)
+        for temp in temps:
+            _remove(temp)
+        raise
+
+    for _, backup, had_output in swapped:
+        if had_output:
             _remove(backup)
-            output.rename(backup)
-            backups.append(backup)
-        temp.rename(output)
-    for backup in backups:
-        _remove(backup)
diff --git a/tests/test_convert.py b/tests/test_convert.py
index ec8046c..73a87db 100644
--- a/tests/test_convert.py
+++ b/tests/test_convert.py
@@ -389,34 +389,3 @@ def test_forwarding_still_validates_like_recast(h5ad, tmp_path):
     )
     assert result.exit_code == 1
     assert "--layout must be one of" in _out(result)
-
-
-# ---------------------------------------------------------------------------
-# staging, directly
-
-
-def test_staging_replaces_an_existing_directory_rather_than_nesting_in_it(tmp_path):
-    from adata.util.staging import staged_outputs
-
-    out = tmp_path / "out.zarr"
-    out.mkdir()
-    (out / "old").write_text("old")
-
-    with staged_outputs([out]) as (temp,):
-        temp.mkdir()
-        (temp / "new").write_text("new")
-
-    assert sorted(p.name for p in out.iterdir()) == ["new"]
-    assert sorted(p.name for p in tmp_path.iterdir()) == ["out.zarr"]
-
-
-def test_staging_moves_several_outputs_together(tmp_path):
-    from adata.util.staging import staged_outputs
-
-    outs = [tmp_path / "a.h5ad", tmp_path / "b.h5ad"]
-    with staged_outputs(outs) as temps:
-        assert [t.suffix for t in temps] == [".h5ad", ".h5ad"]
-        for temp in temps:
-            temp.write_text(temp.name)
-
-    assert [p.read_text() for p in outs] == ["a.adata-tmp.h5ad", "b.adata-tmp.h5ad"]
diff --git a/tests/test_recast.py b/tests/test_recast.py
index b49c41e..e6cb376 100644
--- a/tests/test_recast.py
+++ b/tests/test_recast.py
@@ -483,7 +483,7 @@ def test_inplace_replaces_the_source(tmp_path):
     got = ad.read_h5ad(store)
     assert got.X.dtype == np.dtype("float32")
     assert np.array_equal(got.X.toarray(), matrix.toarray().astype("float32"))
-    assert not list(tmp_path.glob("*recast-tmp*")), "temp file left behind"
+    assert not list(tmp_path.glob("*adata-*")), "temp file left behind"
 
 
 def test_converting_to_zarr_keeps_the_values(tmp_path):
diff --git a/tests/test_staging.py b/tests/test_staging.py
new file mode 100644
index 0000000..a781822
--- /dev/null
+++ b/tests/test_staging.py
@@ -0,0 +1,113 @@
+"""Tests for `adata.util.staging`, the write-beside-then-rename helper.
+
+Every command that writes a store goes through it, so its failure paths are
+tested here directly rather than through each command.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import pytest
+
+from adata.util.staging import staged_outputs
+
+
+def _names(directory: Path) -> list:
+    return sorted(p.name for p in directory.iterdir())
+
+
+def test_staging_replaces_an_existing_directory_rather_than_nesting_in_it(tmp_path):
+    out = tmp_path / "out.zarr"
+    out.mkdir()
+    (out / "old").write_text("old")
+
+    with staged_outputs([out]) as (temp,):
+        temp.mkdir()
+        (temp / "new").write_text("new")
+
+    assert _names(out) == ["new"]
+    assert _names(tmp_path) == ["out.zarr"]
+
+
+def test_staging_moves_several_outputs_together(tmp_path):
+    outs = [tmp_path / "a.h5ad", tmp_path / "b.h5ad"]
+    with staged_outputs(outs) as temps:
+        assert [t.suffix for t in temps] == [".h5ad", ".h5ad"]
+        for temp in temps:
+            temp.write_text(temp.name)
+
+    assert [p.read_text() for p in outs] == ["a.adata-tmp.h5ad", "b.adata-tmp.h5ad"]
+
+
+def test_a_failure_while_writing_keeps_the_existing_output(tmp_path):
+    out = tmp_path / "out.h5ad"
+    out.write_text("old")
+
+    with pytest.raises(RuntimeError):
+        with staged_outputs([out]) as (temp,):
+            temp.write_text("half")
+            raise RuntimeError("interrupted")
+
+    assert out.read_text() == "old"
+    assert _names(tmp_path) == ["out.h5ad"]
+
+
+def test_a_leftover_backup_is_refused_not_deleted(tmp_path):
+    """After a run killed mid-swap, the backup may be the only copy."""
+    out = tmp_path / "out.h5ad"
+    backup = tmp_path / "out.adata-old.h5ad"
+    backup.write_text("the only copy")
+
+    with pytest.raises(FileExistsError, match="adata-old"):
+        with staged_outputs([out]):
+            pytest.fail("the body must not run")
+
+    assert backup.read_text() == "the only copy"
+
+
+def test_a_leftover_temporary_is_refused(tmp_path):
+    out = tmp_path / "out.h5ad"
+    (tmp_path / "out.adata-tmp.h5ad").write_text("stale")
+
+    with pytest.raises(FileExistsError, match="adata-tmp"):
+        with staged_outputs([out]):
+            pytest.fail("the body must not run")
+
+
+def test_a_failed_swap_puts_back_the_outputs_already_swapped(tmp_path, monkeypatch):
+    outs = [tmp_path / "a.h5ad", tmp_path / "b.h5ad"]
+    for out in outs:
+        out.write_text(f"old {out.stem}")
+
+    real_rename = Path.rename
+
+    def rename(self, target):
+        if self.name == "b.adata-tmp.h5ad":
+            raise OSError("disk full")
+        return real_rename(self, target)
+
+    monkeypatch.setattr(Path, "rename", rename)
+
+    with pytest.raises(OSError, match="disk full"):
+        with staged_outputs(outs) as temps:
+            for temp in temps:
+                temp.write_text(f"new {temp.name}")
+
+    assert [p.read_text() for p in outs] == ["old a", "old b"]
+    assert _names(tmp_path) == ["a.h5ad", "b.h5ad"]
+
+
+def test_a_zarr_store_named_without_the_suffix_stays_zarr(tmp_path):
+    """`store.adata-tmp` would not exist yet, and so be opened as HDF5."""
+    out = tmp_path / "store"
+    out.mkdir()
+    (out / "zarr.json").write_text("{}")
+
+    with staged_outputs([out]) as (temp,):
+        assert temp.suffix == ".zarr"
+        temp.mkdir()
+        (temp / "zarr.json").write_text('{"new": true}')
+
+    assert (out / "zarr.json").read_text() == '{"new": true}'
+    assert _names(tmp_path) == ["store"]
diff --git a/tests/test_write_safety.py b/tests/test_write_safety.py
new file mode 100644
index 0000000..462b9c5
--- /dev/null
+++ b/tests/test_write_safety.py
@@ -0,0 +1,222 @@
+"""Failures in `subset`, `recast` and `split` must not cost the user data.
+
+Each test reproduces a way these commands used to destroy or corrupt a store
+while reporting success, or leave a half-written one behind (issue #35):
+
+- `subset x.zarr -o x.zarr` emptied the store: the output was opened "w"
+  before the selection was read from the source it had just cleared.
+- A selection that failed still left an empty output, because the output was
+  opened before the query was evaluated.
+- `--inplace` deleted the source and then moved the new store in, so a crash
+  between the two lost both.
+- `split` wrote straight into its output directory.
+"""
+
+from __future__ import annotations
+
+import re
+from pathlib import Path
+
+import numpy as np
+import pytest
+from typer.testing import CliRunner
+
+from adata.cli import app
+
+ad = pytest.importorskip("anndata", reason="anndata is required for these tests")
+pd = pytest.importorskip("pandas")
+
+runner = CliRunner()
+_ANSI = re.compile(r"\x1b\[[0-9;]*[a-zA-Z]")
+
+
+def _out(result) -> str:
+    text = result.stdout + (result.stderr or "")
+    return " ".join(_ANSI.sub("", text).split())
+
+
+def _store(path: Path) -> Path:
+    obj = ad.AnnData(
+        X=np.arange(16, dtype="float32").reshape(4, 4),
+        obs=pd.DataFrame(
+            {"grp": pd.Categorical(["a", "b", "a", "b"])},
+            index=[f"c{i}" for i in range(4)],
+        ),
+        var=pd.DataFrame(index=[f"g{i}" for i in range(4)]),
+    )
+    if path.suffix == ".zarr":
+        obj.write_zarr(path)
+    else:
+        obj.write_h5ad(path)
+    return path
+
+
+def _read(path: Path):
+    return ad.read_zarr(path) if path.suffix == ".zarr" else ad.read_h5ad(path)
+
+
+def _debris(directory: Path) -> list:
+    return sorted(p.name for p in directory.glob("*adata-*"))
+
+
+def _fail(*args, **kwargs):
+    raise RuntimeError("interrupted while writing")
+
+
+@pytest.fixture(params=[".h5ad", ".zarr"])
+def suffix(request) -> str:
+    return request.param
+
+
+# ---------------------------------------------------------------------------
+# subset
+
+
+def test_subset_onto_its_own_input_is_refused(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+
+    result = runner.invoke(
+        app, ["subset", str(store), "-o", str(store), "-q", "grp == a"]
+    )
+    assert result.exit_code == 1
+    assert "Output path is the input" in _out(result)
+    assert _read(store).shape == (4, 4)
+
+
+def test_subset_a_failed_selection_writes_nothing(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+    out = tmp_path / f"out{suffix}"
+
+    result = runner.invoke(
+        app, ["subset", str(store), "-o", str(out), "-q", "missing == 1"]
+    )
+    assert result.exit_code == 1
+    assert not out.exists()
+    assert _debris(tmp_path) == []
+
+
+def test_subset_a_failed_selection_keeps_an_existing_output(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+    out = _store(tmp_path / f"out{suffix}")
+
+    result = runner.invoke(
+        app, ["subset", str(store), "-o", str(out), "-q", "missing == 1"]
+    )
+    assert result.exit_code == 1
+    assert _read(out).shape == (4, 4)
+
+
+def test_subset_still_replaces_an_existing_output(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+    out = _store(tmp_path / f"out{suffix}")
+
+    result = runner.invoke(
+        app, ["subset", str(store), "-o", str(out), "-q", "grp == a"]
+    )
+    assert result.exit_code == 0, _out(result)
+    assert list(_read(out).obs_names) == ["c0", "c2"]
+    assert _debris(tmp_path) == []
+
+
+def test_subset_inplace_replaces_the_source(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+
+    result = runner.invoke(app, ["subset", str(store), "--inplace", "-q", "grp == a"])
+    assert result.exit_code == 0, _out(result)
+    assert list(_read(store).obs_names) == ["c0", "c2"]
+    assert _debris(tmp_path) == []
+
+
+def test_subset_inplace_keeps_the_source_if_writing_fails(
+    tmp_path, suffix, monkeypatch
+):
+    from adata.core import subset as subset_mod
+
+    store = _store(tmp_path / f"in{suffix}")
+    monkeypatch.setattr(subset_mod, "_write_targets", _fail)
+
+    result = runner.invoke(app, ["subset", str(store), "--inplace", "-q", "grp == a"])
+    assert result.exit_code == 1
+    assert "interrupted" in _out(result)
+    assert _read(store).shape == (4, 4)
+    assert _debris(tmp_path) == []
+
+
+# ---------------------------------------------------------------------------
+# recast
+
+
+def test_recast_inplace_replaces_the_source(tmp_path, suffix):
+    store = _store(tmp_path / f"in{suffix}")
+
+    result = runner.invoke(
+        app, ["recast", str(store), "X", "--inplace", "--dtype", "float64"]
+    )
+    assert result.exit_code == 0, _out(result)
+    assert _read(store).X.dtype == np.dtype("float64")
+    assert _debris(tmp_path) == []
+
+
+def test_recast_inplace_keeps_the_source_if_writing_fails(
+    tmp_path, suffix, monkeypatch
+):
+    from adata.commands import recast as recast_mod
+
+    store = _store(tmp_path / f"in{suffix}")
+    monkeypatch.setattr(recast_mod, "_write", _fail)
+
+    result = runner.invoke(
+        app, ["recast", str(store), "X", "--inplace", "--dtype", "float64"]
+    )
+    assert result.exit_code == 1
+    got = _read(store)
+    assert got.X.dtype == np.dtype("float32")
+    assert np.array_equal(got.X, np.arange(16, dtype="float32").reshape(4, 4))
+    assert _debris(tmp_path) == []
+
+
+def test_recast_keeps_an_existing_output_if_writing_fails(
+    tmp_path, suffix, monkeypatch
+):
+    from adata.commands import recast as recast_mod
+
+    store = _store(tmp_path / f"in{suffix}")
+    out = _store(tmp_path / f"out{suffix}")
+    monkeypatch.setattr(recast_mod, "_write", _fail)
+
+    result = runner.invoke(
+        app, ["recast", str(store), "X", "-o", str(out), "--dtype", "float64"]
+    )
+    assert result.exit_code == 1
+    assert _read(out).X.dtype == np.dtype("float32")
+    assert _debris(tmp_path) == []
+
+
+# ---------------------------------------------------------------------------
+# split
+
+
+def test_split_leaves_nothing_behind_if_writing_fails(tmp_path, monkeypatch):
+    from adata.core import subset as subset_mod
+
+    store = _store(tmp_path / "in.h5ad")
+    out_dir = tmp_path / "parts"
+    out_dir.mkdir()
+    (out_dir / "a.h5ad").write_text("an older split")
+    monkeypatch.setattr(subset_mod, "_write_targets", _fail)
+
+    result = runner.invoke(app, ["split", str(store), "--by", "grp", "-o", str(out_dir)])
+    assert result.exit_code == 1
+    assert sorted(p.name for p in out_dir.iterdir()) == ["a.h5ad"]
+    assert (out_dir / "a.h5ad").read_text() == "an older split"
+
+
+def test_split_onto_its_own_input_is_refused(tmp_path):
+    # Splitting a.h5ad on a column with a value "a", into its own directory.
+    store = _store(tmp_path / "a.h5ad")
+
+    result = runner.invoke(app, ["split", str(store), "--by", "grp", "-o", str(tmp_path)])
+    assert result.exit_code == 1
+    assert "Output path is the input" in _out(result)
+    assert _read(store).shape == (4, 4)
+    assert not (tmp_path / "b.h5ad").exists()

From 8a3550a54ed4a0068132a63a0108837106c45372 Mon Sep 17 00:00:00 2001
From: Aljes <aleksandr.binkevich@gmail.com>
Date: Thu, 15 Oct 2026 14:22:05 +0100
Subject: [PATCH 2/2] concat: write beside the output, rename into place

A failure part way through left a partial store at -o, which then read
as a finished one and blocked the rerun (concat refuses an existing
output).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 src/adata/core/concat.py   |  7 ++++++-
 tests/test_write_safety.py | 18 ++++++++++++++++++
 2 files changed, 24 insertions(+), 1 deletion(-)

diff --git a/src/adata/core/concat.py b/src/adata/core/concat.py
index 0cee9aa..10c6c81 100644
--- a/src/adata/core/concat.py
+++ b/src/adata/core/concat.py
@@ -46,6 +46,7 @@
     is_group,
     open_store,
 )
+from adata.util.staging import staged_outputs
 
 MERGE_STRATEGIES = ("same", "unique", "first", "only")
 #: What the CLI accepts. "drop" is the default and maps to no merge at all.
@@ -906,7 +907,11 @@ def concat_on_disk(
         # half-written store behind.
         check_matrix_encodings(roots, console)
 
-        with open_store(output, "w", zarr_format=zarr_format) as dst_store:
+        # Written beside the output and renamed into place once complete, so
+        # a failure part way leaves nothing that reads as a finished store.
+        with staged_outputs([output]) as (temp,), open_store(
+            temp, "w", zarr_format=zarr_format
+        ) as dst_store:
             dst = dst_store.root
             _write_obs(
                 dst, roots, target_obs, obs_counts, join, label, input_keys, console
diff --git a/tests/test_write_safety.py b/tests/test_write_safety.py
index 462b9c5..997e688 100644
--- a/tests/test_write_safety.py
+++ b/tests/test_write_safety.py
@@ -220,3 +220,21 @@ def test_split_onto_its_own_input_is_refused(tmp_path):
     assert "Output path is the input" in _out(result)
     assert _read(store).shape == (4, 4)
     assert not (tmp_path / "b.h5ad").exists()
+
+
+# ---------------------------------------------------------------------------
+# concat
+
+
+def test_concat_leaves_nothing_behind_if_writing_fails(tmp_path, suffix, monkeypatch):
+    from adata.core import concat as concat_mod
+
+    a = _store(tmp_path / f"a{suffix}")
+    b = _store(tmp_path / f"b{suffix}")
+    out = tmp_path / f"m{suffix}"
+    monkeypatch.setattr(concat_mod, "_write_matrices", _fail)
+
+    result = runner.invoke(app, ["concat", str(a), str(b), "-o", str(out)])
+    assert result.exit_code == 1
+    assert not out.exists()
+    assert _debris(tmp_path) == []