From e1049008b4de711c505a20c2fd87493ce11c57fc Mon Sep 17 00:00:00 2001
From: Jack McCarthy <37917934+Deicyde@users.noreply.github.com>
Date: Mon, 5 Oct 2026 05:25:30 -0400
Subject: [PATCH 1/3] Add all-or-nothing multi-key claims

ClaimBoard gains acquire_many, renew_many, and release_many. Each reads
every key with one ls-remote, applies the per-key checks of the
single-key method, and changes all refs in one git push --atomic with a
--force-with-lease per ref, so a batch never holds some claims and not
others. A lost race or a held claim returns a ClaimBatchResult naming
the blocking keys, read from the porcelain status lines or the remote's
"cannot lock ref" error. A board that does not support atomic pushes
raises ClaimTransportError instead of falling back to separate pushes.

autoform claim acquire|renew|release now takes one or more nodes. One
node keeps today's code path and output; several print one line per
target on success, one error line naming the blockers on failure, and
exit 2 on a duplicate target.
---
 autoform_cli/__main__.py |  38 +++++-
 autoform_cli/claims.py   | 185 +++++++++++++++++++++++----
 tests/test_claim_cli.py  |  97 ++++++++++++++-
 tests/test_claims.py     | 263 +++++++++++++++++++++++++++++++++++++++
 4 files changed, 554 insertions(+), 29 deletions(-)

diff --git a/autoform_cli/__main__.py b/autoform_cli/__main__.py
index 45b4510e..f4d251d1 100644
--- a/autoform_cli/__main__.py
+++ b/autoform_cli/__main__.py
@@ -136,7 +136,7 @@ def main(argv: Sequence[str] | None = None) -> int:
     claim_subparsers = claim.add_subparsers(dest="claim_command", required=True)
     for operation in ("acquire", "renew", "release"):
         command = claim_subparsers.add_parser(operation)
-        command.add_argument("node_id")
+        command.add_argument("node_id", nargs="+", help="claim target(s); several change all-or-nothing")
         _add_claim_board_arguments(command)
         if operation in {"acquire", "renew"}:
             command.add_argument("--ttl", type=int, default=CLAIM_TTL_S)
@@ -544,7 +544,36 @@ def _claim(args: argparse.Namespace) -> int:
             print(f"removed {board.cleanup()} expired claim(s)")
             return 0
 
-        key = author_claim_key(args.node_id)
+        past_tense = {"acquire": "acquired", "renew": "renewed", "release": "released"}
+        if len(args.node_id) > 1:
+            # Several targets change in one atomic push, so a failure holds none of them.
+            targets: dict[str, str] = {}
+            for node_id in args.node_id:
+                key = author_claim_key(node_id)
+                if key in targets:
+                    print(f"error: duplicate claim target: {node_id}", file=sys.stderr)
+                    return 2
+                targets[key] = node_id
+            if operation == "acquire":
+                result = board.acquire_many(list(targets), ttl=args.ttl, note=args.note)
+            elif operation == "renew":
+                result = board.renew_many(list(targets), ttl=args.ttl)
+            else:
+                result = board.release_many(list(targets))
+            if result:
+                for key, node_id in targets.items():
+                    print(f"{past_tense[operation]} {node_id} ({key})")
+                return 0
+            blocking = ", ".join(targets.get(key, key) for key in result.blocking)
+            reason = f"{result.reason}: {blocking}" if blocking else result.reason
+            print(
+                f"error: could not {operation} {', '.join(args.node_id)}; "
+                f"no claim was {past_tense[operation]}: {reason}"
+            )
+            return 1
+
+        node_id = args.node_id[0]
+        key = author_claim_key(node_id)
         if operation == "acquire":
             succeeded = board.acquire(key, ttl=args.ttl, note=args.note)
         elif operation == "renew":
@@ -552,10 +581,9 @@ def _claim(args: argparse.Namespace) -> int:
         else:
             succeeded = board.release(key)
         if succeeded:
-            past_tense = {"acquire": "acquired", "renew": "renewed", "release": "released"}
-            print(f"{past_tense[operation]} {args.node_id} ({key})")
+            print(f"{past_tense[operation]} {node_id} ({key})")
             return 0
-        print(f"error: could not {operation} {args.node_id}; ownership is held or unverifiable")
+        print(f"error: could not {operation} {node_id}; ownership is held or unverifiable")
         return 1
     except (ClaimTransportError, ValueError) as exc:
         print(f"error: {exc}")
diff --git a/autoform_cli/claims.py b/autoform_cli/claims.py
index 4b5c727c..12d90f53 100644
--- a/autoform_cli/claims.py
+++ b/autoform_cli/claims.py
@@ -17,8 +17,9 @@
 import subprocess
 import threading
 import time
+from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Mapping
+from typing import Any, Mapping, Sequence
 
 CLAIM_REF_PREFIX = "refs/autoform-claims/"
 CLAIM_SCHEMA = "autoform-claim/v1"
@@ -38,6 +39,9 @@
     "remote ref updated since checkout",
     "cannot lock ref",
 )
+# When the remote aborts an atomic push, every ref reports "atomic transaction
+# failed"; only this error line names the ref whose expected value was gone.
+_LOCK_FAILURE_RE = re.compile(r"cannot lock ref '([^']+)'")
 
 
 class ClaimTransportError(RuntimeError):
@@ -48,6 +52,22 @@ class MalformedLeaseError(ClaimTransportError):
     """A claim ref exists, but its lease cannot be verified safely."""
 
 
+@dataclass(frozen=True, slots=True)
+class ClaimBatchResult:
+    """The outcome of an all-or-nothing multi-key claim operation.
+
+    It is true exactly when every key changed. Otherwise no key changed,
+    ``reason`` says why, and ``blocking`` names the keys responsible when known.
+    """
+
+    ok: bool
+    blocking: tuple[str, ...] = ()
+    reason: str = ""
+
+    def __bool__(self) -> bool:
+        return self.ok
+
+
 def _validate_key(key: str) -> str:
     if not isinstance(key, str) or not CLAIM_KEY_RE.fullmatch(key) or ".." in key:
         raise ValueError(f"invalid claim key {key!r}")
@@ -57,6 +77,32 @@ def _validate_key(key: str) -> str:
     return key
 
 
+def _validate_keys(keys: Sequence[str]) -> tuple[str, ...]:
+    if isinstance(keys, str):
+        raise TypeError("claim keys must be a sequence of keys, not one string")
+    batch = tuple(_validate_key(key) for key in keys)
+    if not batch:
+        raise ValueError("at least one claim key is required")
+    seen: set[str] = set()
+    for key in batch:
+        if key in seen:
+            raise ValueError(f"duplicate claim key {key!r}")
+        seen.add(key)
+    return batch
+
+
+def _rejected_keys(detail: str, keys_by_ref: Mapping[str, str]) -> tuple[str, ...]:
+    """Return the keys a refused push reports as stale or contended, in push order."""
+    refs = set(_LOCK_FAILURE_RE.findall(detail))
+    for line in detail.splitlines():
+        # Porcelain status lines read "<flag>\t<from>:<to>\t<summary>".
+        flag, _, rest = line.partition("\t")
+        refspec, _, summary = rest.partition("\t")
+        if flag == "!" and any(marker in summary.lower() for marker in _CAS_REJECTIONS):
+            refs.add(refspec.rpartition(":")[2])
+    return tuple(key for ref, key in keys_by_ref.items() if ref in refs)
+
+
 def _is_finite_number(value: object) -> bool:
     if isinstance(value, bool):
         return False
@@ -145,6 +191,19 @@ def _remote_oid(self, key: str) -> str | None:
         line = proc.stdout.strip()
         return line.split("\t", 1)[0] if line else None
 
+    def _remote_oids(self, keys: Sequence[str]) -> dict[str, str]:
+        """Read every present key's object ID with one ``ls-remote``."""
+        keys_by_ref = {self._ref(key): key for key in keys}
+        proc = self._git(["ls-remote", self.repo_url, *keys_by_ref])
+        oids: dict[str, str] = {}
+        for line in proc.stdout.splitlines():
+            oid, separator, ref = line.partition("\t")
+            # ls-remote patterns also match longer names ending in the same
+            # path, such as refs/heads/<ref>; only the exact ref is the claim.
+            if separator and ref in keys_by_ref:
+                oids[keys_by_ref[ref]] = oid
+        return oids
+
     def _read_lease(self, key: str, oid: str) -> dict[str, Any]:
         ref = self._ref(key)
         if self._git(["cat-file", "-e", f"{oid}^{{commit}}"], check=False).returncode != 0:
@@ -182,6 +241,27 @@ def _lease_is_valid(lease: Mapping[str, Any], key: str | None = None) -> bool:
         )
         return bool(valid and (key is None or lease.get("resource") == key))
 
+    def _held_by_peer(self, key: str, old: str | None) -> bool:
+        """Return whether ``old`` is another worker's verified live lease, which blocks acquiring."""
+        if old is None:
+            return False
+        lease = self._read_lease(key, old)
+        return bool(
+            lease is not None
+            and self._lease_is_valid(lease, key)
+            and lease.get("owner") != self.worker_id
+            and not self.expired(lease)
+        )
+
+    def _owned_lease(self, key: str, old: str | None) -> dict[str, Any] | None:
+        """Return this worker's verified lease at ``old``, or ``None`` if absent or not owned."""
+        if old is None:
+            return None
+        lease = self._read_lease(key, old)
+        if lease is None or not self._lease_is_valid(lease, key) or lease.get("owner") != self.worker_id:
+            return None
+        return lease
+
     def _make_lease_commit(self, key: str, ttl: int | float, note: str = "") -> str:
         key = _validate_key(key)
         ttl = _validate_ttl(ttl)
@@ -210,24 +290,41 @@ def _make_lease_commit(self, key: str, ttl: int | float, note: str = "") -> str:
         return self._git(["commit-tree", tree, "-m", message]).stdout.strip()
 
     def _cas_push(self, key: str, old: str | None, new: str) -> bool:
-        ref = self._ref(key)
-        source = new if new else ""
+        return self._cas_push_refs([(key, old, new)]).ok
+
+    def _cas_push_refs(
+        self,
+        updates: Sequence[tuple[str, str | None, str]],
+        *,
+        atomic: bool = False,
+    ) -> ClaimBatchResult:
+        """Push ``(key, old, new)`` updates, each leased on its observed ``old``; empty ``new`` deletes.
+
+        With ``atomic`` the remote applies every update or none. A lost
+        compare-and-swap is a failed result; anything else raises.
+        """
+        keys_by_ref = {self._ref(key): key for key, _, _ in updates}
         proc = self._git(
             [
                 "push",
                 "--quiet",
                 "--porcelain",
-                f"--force-with-lease={ref}:{old or ''}",
+                *(["--atomic"] if atomic else []),
+                *(f"--force-with-lease={self._ref(key)}:{old or ''}" for key, old, _ in updates),
                 self.repo_url,
-                f"{source}:{ref}",
+                *(f"{new if new else ''}:{self._ref(key)}" for key, _, new in updates),
             ],
             check=False,
         )
         if proc.returncode == 0:
-            return True
+            return ClaimBatchResult(True)
         detail = f"{proc.stdout}\n{proc.stderr}".strip()
+        if atomic and "does not support --atomic" in detail:
+            # Falling back to separate pushes could leave a partial claim set.
+            raise ClaimTransportError("claim board does not support atomic pushes; refusing a multi-key update")
         if any(marker in detail.lower() for marker in _CAS_REJECTIONS):
-            return False
+            stale = _rejected_keys(detail, keys_by_ref)
+            return ClaimBatchResult(False, stale, "changed concurrently" if stale else "a claim changed concurrently")
         raise ClaimTransportError(f"claim CAS push failed: {detail[:300]}")
 
     def read(self, key: str) -> dict[str, Any] | None:
@@ -253,16 +350,8 @@ def acquire(self, key: str, ttl: int | float = CLAIM_TTL_S, steal: bool = False,
         _validate_ttl(ttl)
         self._ensure_scratch()
         old = self._remote_oid(key)
-        if old is not None:
-            lease = self._read_lease(key, old)
-            if (
-                lease is not None
-                and self._lease_is_valid(lease, key)
-                and lease.get("owner") != self.worker_id
-                and not self.expired(lease)
-                and not steal
-            ):
-                return False
+        if self._held_by_peer(key, old) and not steal:
+            return False
         new = self._make_lease_commit(key, ttl, note)
         return self._cas_push(key, old, new)
 
@@ -272,10 +361,8 @@ def renew(self, key: str, ttl: int | float = CLAIM_TTL_S) -> bool:
         _validate_ttl(ttl)
         self._ensure_scratch()
         old = self._remote_oid(key)
-        if old is None:
-            return False
-        lease = self._read_lease(key, old)
-        if lease is None or not self._lease_is_valid(lease, key) or lease.get("owner") != self.worker_id:
+        lease = self._owned_lease(key, old)
+        if lease is None:
             return False
         new = self._make_lease_commit(key, ttl, str(lease.get("note", "")))
         return self._cas_push(key, old, new)
@@ -287,11 +374,62 @@ def release(self, key: str) -> bool:
         old = self._remote_oid(key)
         if old is None:
             return True
-        lease = self._read_lease(key, old)
-        if lease is None or not self._lease_is_valid(lease, key) or lease.get("owner") != self.worker_id:
+        if self._owned_lease(key, old) is None:
             return False
         return self._cas_push(key, old, "")
 
+    # The *_many methods read every key with one ls-remote, apply the per-key
+    # checks of the single-key method, and change all keys in one atomic push,
+    # so a batch never leaves some of its claims taken and others not.
+
+    def acquire_many(
+        self,
+        keys: Sequence[str],
+        *,
+        ttl: int | float = CLAIM_TTL_S,
+        note: str = "",
+    ) -> ClaimBatchResult:
+        """Acquire every key or none; a live lease of another worker on any key blocks all."""
+        keys = _validate_keys(keys)
+        _validate_ttl(ttl)
+        self._ensure_scratch()
+        olds = self._remote_oids(keys)
+        held = tuple(key for key in keys if self._held_by_peer(key, olds.get(key)))
+        if held:
+            return ClaimBatchResult(False, held, "held by another worker")
+        updates = [(key, olds.get(key), self._make_lease_commit(key, ttl, note)) for key in keys]
+        return self._cas_push_refs(updates, atomic=True)
+
+    def renew_many(self, keys: Sequence[str], *, ttl: int | float = CLAIM_TTL_S) -> ClaimBatchResult:
+        """Renew every key or none; each must be this worker's lease, as :meth:`renew` requires."""
+        keys = _validate_keys(keys)
+        _validate_ttl(ttl)
+        self._ensure_scratch()
+        olds = self._remote_oids(keys)
+        notes: dict[str, str] = {}
+        for key in keys:
+            lease = self._owned_lease(key, olds.get(key))
+            if lease is not None:
+                notes[key] = str(lease.get("note", ""))
+        lost = tuple(key for key in keys if key not in notes)
+        if lost:
+            return ClaimBatchResult(False, lost, "not held by this worker")
+        updates = [(key, olds[key], self._make_lease_commit(key, ttl, notes[key])) for key in keys]
+        return self._cas_push_refs(updates, atomic=True)
+
+    def release_many(self, keys: Sequence[str]) -> ClaimBatchResult:
+        """Delete every present key or none; absent keys count as released, as in :meth:`release`."""
+        keys = _validate_keys(keys)
+        self._ensure_scratch()
+        olds = self._remote_oids(keys)
+        present = [key for key in keys if key in olds]
+        foreign = tuple(key for key in present if self._owned_lease(key, olds[key]) is None)
+        if foreign:
+            return ClaimBatchResult(False, foreign, "not held by this worker")
+        if not present:
+            return ClaimBatchResult(True)
+        return self._cas_push_refs([(key, olds[key], "") for key in present], atomic=True)
+
     def holds(self, key: str) -> bool:
         """Return whether this worker verifiably owns the current live lease."""
         lease = self.read(key)
@@ -419,6 +557,7 @@ def _run(self) -> None:
     "CLAIM_REF_PREFIX",
     "CLAIM_SCHEMA",
     "CLAIM_TTL_S",
+    "ClaimBatchResult",
     "ClaimBoard",
     "ClaimTransportError",
     "Heartbeat",
diff --git a/tests/test_claim_cli.py b/tests/test_claim_cli.py
index e0a48236..d108066d 100644
--- a/tests/test_claim_cli.py
+++ b/tests/test_claim_cli.py
@@ -5,7 +5,7 @@
 from pathlib import Path
 
 from autoform_cli.__main__ import main
-from autoform_cli.claims import CLAIM_REF_PREFIX, author_claim_key
+from autoform_cli.claims import CLAIM_REF_PREFIX, ClaimBoard, author_claim_key
 
 
 def _bare_repo(tmp_path: Path) -> Path:
@@ -102,3 +102,98 @@ def test_claim_cli_refuses_malformed_remote_lease(tmp_path: Path, capsys) -> Non
 
     assert main(_args(repo, tmp_path / "scratch", "acquire", node_id)) == 1
     assert "invalid lease JSON" in capsys.readouterr().out
+
+
+def _claim_refs(repo: Path) -> list[str]:
+    listing = subprocess.run(
+        ["git", "for-each-ref", "--format=%(refname)", CLAIM_REF_PREFIX],
+        cwd=repo,
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout
+    return sorted(listing.split())
+
+
+def _peer_args(repo: Path, scratch: Path, *command: str) -> list[str]:
+    return ["claim", *command, "--repo", str(repo), "--worker-id", "worker-b", "--scratch", str(scratch)]
+
+
+def test_claim_cli_single_node_output_is_unchanged(tmp_path: Path, capsys, monkeypatch) -> None:
+    def batch_path(*args: object, **kwargs: object) -> None:
+        raise AssertionError("one target must take the single-key path")
+
+    for name in ("acquire_many", "renew_many", "release_many"):
+        monkeypatch.setattr(ClaimBoard, name, batch_path)
+    repo = _bare_repo(tmp_path)
+    scratch = tmp_path / "scratch"
+    key = author_claim_key("node")
+
+    assert main(_args(repo, scratch, "acquire", "node", "--ttl", "600")) == 0
+    assert capsys.readouterr().out == f"acquired node ({key})\n"
+    assert main(_args(repo, scratch, "renew", "node", "--ttl", "600")) == 0
+    assert capsys.readouterr().out == f"renewed node ({key})\n"
+    assert main(_peer_args(repo, tmp_path / "peer", "acquire", "node")) == 1
+    assert capsys.readouterr().out == "error: could not acquire node; ownership is held or unverifiable\n"
+    assert main(_args(repo, scratch, "release", "node")) == 0
+    assert capsys.readouterr().out == f"released node ({key})\n"
+
+
+def test_claim_cli_multi_node_round_trip_prints_one_line_per_target(tmp_path: Path, capsys) -> None:
+    repo = _bare_repo(tmp_path)
+    scratch = tmp_path / "scratch"
+    nodes = ["chapter/main theorem", "chapter/helper"]
+
+    def expected(past_tense: str) -> str:
+        return "".join(f"{past_tense} {node} ({author_claim_key(node)})\n" for node in nodes)
+
+    assert main(_args(repo, scratch, "acquire", *nodes, "--ttl", "600", "--note", "revision")) == 0
+    assert capsys.readouterr().out == expected("acquired")
+    assert main(_args(repo, scratch, "renew", *nodes, "--ttl", "600")) == 0
+    assert capsys.readouterr().out == expected("renewed")
+    assert main(_args(repo, scratch, "list")) == 0
+    leases = json.loads(capsys.readouterr().out)
+    assert sorted(lease["_key"] for lease in leases) == sorted(author_claim_key(node) for node in nodes)
+    assert {(lease["owner"], lease["note"]) for lease in leases} == {("worker-a", "revision")}
+    assert main(_args(repo, scratch, "release", *nodes)) == 0
+    assert capsys.readouterr().out == expected("released")
+    assert _claim_refs(repo) == []
+
+
+def test_claim_cli_multi_node_failure_changes_no_claim(tmp_path: Path, capsys) -> None:
+    repo = _bare_repo(tmp_path)
+    scratch = tmp_path / "scratch"
+    assert main(_peer_args(repo, tmp_path / "peer", "acquire", "b")) == 0
+    capsys.readouterr()
+
+    assert main(_args(repo, scratch, "acquire", "a", "b", "c")) == 1
+    captured = capsys.readouterr()
+    assert captured.out == "error: could not acquire a, b, c; no claim was acquired: held by another worker: b\n"
+    assert captured.err == ""
+    assert _claim_refs(repo) == [CLAIM_REF_PREFIX + author_claim_key("b")]
+
+    assert main(_args(repo, scratch, "acquire", "a")) == 0
+    capsys.readouterr()
+    assert main(_args(repo, scratch, "release", "a", "b")) == 1
+    assert capsys.readouterr().out == (
+        "error: could not release a, b; no claim was released: not held by this worker: b\n"
+    )
+    assert _claim_refs(repo) == sorted(CLAIM_REF_PREFIX + author_claim_key(node) for node in ("a", "b"))
+
+
+def test_claim_cli_rejects_duplicate_targets_before_touching_the_board(tmp_path: Path, capsys) -> None:
+    repo = _bare_repo(tmp_path)
+    scratch = tmp_path / "scratch"
+
+    assert main(_args(repo, scratch, "acquire", "a", "b", "a")) == 2
+    captured = capsys.readouterr()
+    assert captured.err == "error: duplicate claim target: a\n"
+    assert captured.out == ""
+    assert _claim_refs(repo) == []
+    assert not scratch.exists()
+
+
+def test_claim_cli_multi_node_transport_failure_is_nonzero(tmp_path: Path, capsys) -> None:
+    missing = tmp_path / "missing" / "claims.git"
+    assert main(_args(missing, tmp_path / "scratch", "acquire", "a", "b")) == 1
+    assert capsys.readouterr().out.startswith("error: ")
diff --git a/tests/test_claims.py b/tests/test_claims.py
index 7f140c2d..99fd5fdf 100644
--- a/tests/test_claims.py
+++ b/tests/test_claims.py
@@ -6,6 +6,7 @@
 import os
 import subprocess
 import threading
+from collections.abc import Sequence
 from pathlib import Path
 
 import pytest
@@ -456,3 +457,265 @@ def test_acquire_rejects_nonfinite_expiry_before_commit_or_push(
         board.acquire("bad-expiry", ttl=1e308)
 
     assert _git("for-each-ref", "--format=%(refname)", claims.CLAIM_REF_PREFIX + "bad-expiry", cwd=board_repo) == ""
+
+
+def _refs(repo: Path) -> dict[str, str]:
+    listing = _git("for-each-ref", "--format=%(refname) %(objectname)", cwd=repo)
+    return dict(line.split(" ", 1) for line in listing.splitlines())
+
+
+def test_acquire_many_is_all_or_nothing_when_a_peer_holds_one_key(tmp_path: Path, board_repo: Path) -> None:
+    peer = _board(tmp_path, board_repo, "worker-b")
+    assert peer.acquire("held", ttl=600)
+    before = _refs(board_repo)
+    board = _board(tmp_path, board_repo, "worker-a")
+
+    result = board.acquire_many(["first", "held", "last"], ttl=600)
+
+    assert result == claims.ClaimBatchResult(False, ("held",), "held by another worker")
+    assert not result
+    assert _refs(board_repo) == before
+
+
+def test_acquire_many_takes_over_an_expired_key(tmp_path: Path, board_repo: Path) -> None:
+    _plant_lease(board_repo, "expired", owner="worker-b")
+    board = _board(tmp_path, board_repo, "worker-a")
+
+    result = board.acquire_many(["fresh", "expired"], ttl=600, note="revision")
+
+    assert result == claims.ClaimBatchResult(True)
+    for key in ("fresh", "expired"):
+        lease = board.read(key)
+        assert lease["owner"] == "worker-a"
+        assert lease["note"] == "revision"
+        assert board.holds(key)
+
+
+def test_acquire_many_changes_no_ref_when_one_lease_goes_stale_before_the_push(
+    tmp_path: Path, board_repo: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _plant_lease(board_repo, "contested", owner="worker-b")
+    board = _board(tmp_path, board_repo, "worker-a")
+    peer = _board(tmp_path, board_repo, "worker-b")
+    push = board._cas_push_refs
+    observed: dict[str, dict[str, str]] = {}
+
+    def peer_renews_first(updates: Sequence[tuple[str, str | None, str]], *, atomic: bool = False):
+        # The expired lease passed the ownership check; its owner renews it
+        # before the push, so exactly one of the two leases is stale.
+        assert peer.renew("contested", ttl=600)
+        observed["refs"] = _refs(board_repo)
+        return push(updates, atomic=atomic)
+
+    monkeypatch.setattr(board, "_cas_push_refs", peer_renews_first)
+    result = board.acquire_many(["free", "contested"], ttl=600)
+
+    assert result == claims.ClaimBatchResult(False, ("contested",), "changed concurrently")
+    assert _refs(board_repo) == observed["refs"]
+    assert claims.CLAIM_REF_PREFIX + "free" not in observed["refs"]
+    assert peer.holds("contested")
+
+
+def test_acquire_many_changes_no_ref_when_the_remote_aborts_the_transaction(
+    tmp_path: Path, board_repo: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    board._ensure_scratch()
+    rival = _git("commit-tree", _git("mktree", cwd=board_repo, input_text=""), "-m", "rival", cwd=board_repo)
+    contested = claims.CLAIM_REF_PREFIX + "contested"
+    # Git runs pre-push after checking every lease against the remote's
+    # advertisement, so this rival claim meets the remote's own check inside
+    # the ref transaction instead.
+    hooks = board.scratch / "hooks"
+    hooks.mkdir(exist_ok=True)
+    hook = hooks / "pre-push"
+    hook.write_text(f'#!/bin/sh\ncat >/dev/null\nexec git --git-dir="{board_repo}" update-ref {contested} {rival}\n')
+    hook.chmod(0o755)
+    monkeypatch.setenv("GIT_CONFIG_COUNT", "1")
+    monkeypatch.setenv("GIT_CONFIG_KEY_0", "core.hooksPath")
+    monkeypatch.setenv("GIT_CONFIG_VALUE_0", str(hooks))
+
+    result = board.acquire_many(["free", "contested"], ttl=600)
+
+    assert result == claims.ClaimBatchResult(False, ("contested",), "changed concurrently")
+    assert _refs(board_repo) == {contested: rival}
+
+
+def test_overlapping_batch_race_has_one_winner_and_no_partial_loser(tmp_path: Path, board_repo: Path) -> None:
+    boards = {owner: _board(tmp_path, board_repo, owner) for owner in ("worker-a", "worker-b")}
+    barrier = threading.Barrier(2)
+    original_remote_oids = claims.ClaimBoard._remote_oids
+
+    def synchronized_remote_oids(self: claims.ClaimBoard, keys: Sequence[str]) -> dict[str, str]:
+        oids = original_remote_oids(self, keys)
+        barrier.wait(timeout=60)
+        return oids
+
+    for board in boards.values():
+        board._remote_oids = synchronized_remote_oids.__get__(board, claims.ClaimBoard)  # type: ignore[method-assign]
+
+    results: dict[str, claims.ClaimBatchResult] = {}
+    errors: list[BaseException] = []
+
+    def acquire(owner: str) -> None:
+        try:
+            results[owner] = boards[owner].acquire_many([f"only-{owner}", "shared"], ttl=600)
+        except BaseException as exc:  # pragma: no cover - reported below
+            errors.append(exc)
+
+    threads = [threading.Thread(target=acquire, args=(owner,)) for owner in boards]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join(timeout=120)
+
+    assert not errors
+    assert not any(thread.is_alive() for thread in threads)
+    winners = [owner for owner, result in results.items() if result]
+    assert len(winners) == 1
+    winner = winners[0]
+    loser = next(owner for owner in boards if owner != winner)
+    assert results[loser] == claims.ClaimBatchResult(False, ("shared",), "changed concurrently")
+    assert sorted(_refs(board_repo)) == [claims.CLAIM_REF_PREFIX + key for key in (f"only-{winner}", "shared")]
+    assert boards[loser].read("shared")["owner"] == winner
+
+
+def test_renew_many_renews_every_owned_key_and_keeps_notes(
+    tmp_path: Path, board_repo: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    now = 1_000.0
+    monkeypatch.setattr(claims.time, "time", lambda: now)
+    board = _board(tmp_path, board_repo, "worker-a")
+    assert board.acquire_many(["first", "second"], ttl=60, note="revision")
+    monkeypatch.setattr(claims.time, "time", lambda: now + 30)
+
+    assert board.renew_many(["first", "second"], ttl=600) == claims.ClaimBatchResult(True)
+    for key in ("first", "second"):
+        lease = board.read(key)
+        assert lease["expires_at"] == now + 30 + 600
+        assert lease["note"] == "revision"
+
+
+def test_renew_many_renews_nothing_unless_every_key_is_owned(tmp_path: Path, board_repo: Path) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    peer = _board(tmp_path, board_repo, "worker-b")
+    assert board.acquire("mine", ttl=600)
+    assert peer.acquire("theirs", ttl=600)
+    before = _refs(board_repo)
+
+    result = board.renew_many(["mine", "theirs", "absent"], ttl=900)
+
+    assert result == claims.ClaimBatchResult(False, ("theirs", "absent"), "not held by this worker")
+    assert _refs(board_repo) == before
+
+
+def test_release_many_deletes_owned_keys_and_treats_absent_keys_as_released(tmp_path: Path, board_repo: Path) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    assert board.acquire_many(["first", "second"], ttl=600)
+
+    assert board.release_many(["first", "absent", "second"]) == claims.ClaimBatchResult(True)
+    assert _refs(board_repo) == {}
+    assert board.release_many(["first", "second"]) == claims.ClaimBatchResult(True)
+
+
+def test_release_many_releases_nothing_when_one_key_is_foreign(tmp_path: Path, board_repo: Path) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    peer = _board(tmp_path, board_repo, "worker-b")
+    assert board.acquire("mine", ttl=600)
+    assert peer.acquire("theirs", ttl=600)
+    before = _refs(board_repo)
+
+    result = board.release_many(["mine", "theirs"])
+
+    assert result == claims.ClaimBatchResult(False, ("theirs",), "not held by this worker")
+    assert _refs(board_repo) == before
+
+
+def test_release_many_deletes_no_ref_when_one_lease_goes_stale_before_the_push(
+    tmp_path: Path, board_repo: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    peer = _board(tmp_path, board_repo, "worker-b")
+    assert board.acquire_many(["kept", "stolen"], ttl=600)
+    push = board._cas_push_refs
+    observed: dict[str, dict[str, str]] = {}
+
+    def peer_steals_first(updates: Sequence[tuple[str, str | None, str]], *, atomic: bool = False):
+        assert peer.acquire("stolen", ttl=600, steal=True)
+        observed["refs"] = _refs(board_repo)
+        return push(updates, atomic=atomic)
+
+    monkeypatch.setattr(board, "_cas_push_refs", peer_steals_first)
+    result = board.release_many(["kept", "stolen"])
+
+    assert result == claims.ClaimBatchResult(False, ("stolen",), "changed concurrently")
+    assert _refs(board_repo) == observed["refs"]
+    assert board.holds("kept")
+
+
+@pytest.mark.parametrize("method", ["acquire_many", "renew_many", "release_many"])
+@pytest.mark.parametrize(
+    ("keys", "error", "match"),
+    [
+        (["first", "second", "first"], ValueError, "duplicate claim key 'first'"),
+        ([], ValueError, "at least one claim key is required"),
+        (["first", "../escape"], ValueError, "invalid claim key"),
+        ("first", TypeError, "not one string"),
+    ],
+)
+def test_batch_methods_reject_bad_keys_before_touching_the_board(
+    tmp_path: Path, board_repo: Path, method: str, keys: object, error: type[Exception], match: str
+) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+
+    with pytest.raises(error, match=match):
+        getattr(board, method)(keys)
+
+    assert not board.scratch.exists()
+    assert _refs(board_repo) == {}
+
+
+def test_batch_refuses_a_board_without_atomic_push_support(tmp_path: Path, board_repo: Path) -> None:
+    with (board_repo / "config").open("a", encoding="utf-8") as config:
+        config.write("[receive]\n\tadvertiseAtomic = false\n")
+    board = _board(tmp_path, board_repo, "worker-a")
+
+    with pytest.raises(claims.ClaimTransportError, match="does not support atomic pushes"):
+        board.acquire_many(["first", "second"], ttl=600)
+
+    assert _refs(board_repo) == {}
+    assert board.acquire("first", ttl=600)
+
+
+def test_single_key_push_is_unchanged_and_batches_push_atomically(
+    tmp_path: Path, board_repo: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    git = board._git
+    pushes: list[list[str]] = []
+
+    def recording_git(args: list[str], **kwargs: object) -> subprocess.CompletedProcess[str]:
+        if args[0] == "push":
+            pushes.append(args)
+        return git(args, **kwargs)
+
+    monkeypatch.setattr(board, "_git", recording_git)
+    assert board.acquire("single", ttl=600)
+    assert board.acquire_many(["one", "two"], ttl=600)
+
+    single, one, two = (claims.CLAIM_REF_PREFIX + key for key in ("single", "one", "two"))
+    refs = _refs(board_repo)
+    assert pushes == [
+        ["push", "--quiet", "--porcelain", f"--force-with-lease={single}:", board.repo_url, f"{refs[single]}:{single}"],
+        [
+            "push",
+            "--quiet",
+            "--porcelain",
+            "--atomic",
+            f"--force-with-lease={one}:",
+            f"--force-with-lease={two}:",
+            board.repo_url,
+            f"{refs[one]}:{one}",
+            f"{refs[two]}:{two}",
+        ],
+    ]

From b094a1f0a94c7d6f0900568aceba38e51dc8951b Mon Sep 17 00:00:00 2001
From: Jack McCarthy <37917934+Deicyde@users.noreply.github.com>
Date: Mon, 5 Oct 2026 06:00:51 -0400
Subject: [PATCH 2/3] Name malformed leases as blocking keys in claim batches

The B1 contract separates transport problems, which raise
ClaimTransportError, from a lost race or a held or unverifiable claim,
which return a failure that names the blocking keys. The batch methods
raised MalformedLeaseError for an unverifiable lease, so the CLI printed
a bare key-level error instead of the "no claim was <op>ed" line.

acquire_many, renew_many and release_many now report such a key as
blocking with the reason "malformed lease". When keys block for
different reasons, every key is named in batch order and the reasons
are joined with "or". The per-key checks and the single-key methods are
unchanged, and nothing is pushed.

Tests cover each batch method with a malformed key, a batch with mixed
reasons, the multi-node CLI line for a malformed claim, and the renew
failure line.
---
 autoform_cli/claims.py  | 54 ++++++++++++++++++++++++++++++-----------
 tests/test_claim_cli.py | 13 ++++++++++
 tests/test_claims.py    | 28 +++++++++++++++++++++
 3 files changed, 81 insertions(+), 14 deletions(-)

diff --git a/autoform_cli/claims.py b/autoform_cli/claims.py
index 12d90f53..51b6f843 100644
--- a/autoform_cli/claims.py
+++ b/autoform_cli/claims.py
@@ -56,8 +56,9 @@ class MalformedLeaseError(ClaimTransportError):
 class ClaimBatchResult:
     """The outcome of an all-or-nothing multi-key claim operation.
 
-    It is true exactly when every key changed. Otherwise no key changed,
-    ``reason`` says why, and ``blocking`` names the keys responsible when known.
+    It is true exactly when the operation succeeded for every key. Otherwise no
+    key changed, ``reason`` says why, and ``blocking`` names the keys
+    responsible when known.
     """
 
     ok: bool
@@ -103,6 +104,11 @@ def _rejected_keys(detail: str, keys_by_ref: Mapping[str, str]) -> tuple[str, ..
     return tuple(key for ref, key in keys_by_ref.items() if ref in refs)
 
 
+def _refusal(blocked: Mapping[str, str]) -> ClaimBatchResult:
+    """Refuse a batch before pushing, naming each blocking key in batch order and every reason."""
+    return ClaimBatchResult(False, tuple(blocked), " or ".join(dict.fromkeys(blocked.values())))
+
+
 def _is_finite_number(value: object) -> bool:
     if isinstance(value, bool):
         return False
@@ -380,7 +386,9 @@ def release(self, key: str) -> bool:
 
     # The *_many methods read every key with one ls-remote, apply the per-key
     # checks of the single-key method, and change all keys in one atomic push,
-    # so a batch never leaves some of its claims taken and others not.
+    # so a batch never leaves some of its claims taken and others not. Where
+    # the single-key method raises MalformedLeaseError, a batch names that key
+    # as blocking instead, so one refusal can explain every blocked key.
 
     def acquire_many(
         self,
@@ -394,9 +402,15 @@ def acquire_many(
         _validate_ttl(ttl)
         self._ensure_scratch()
         olds = self._remote_oids(keys)
-        held = tuple(key for key in keys if self._held_by_peer(key, olds.get(key)))
-        if held:
-            return ClaimBatchResult(False, held, "held by another worker")
+        blocked: dict[str, str] = {}
+        for key in keys:
+            try:
+                if self._held_by_peer(key, olds.get(key)):
+                    blocked[key] = "held by another worker"
+            except MalformedLeaseError:
+                blocked[key] = "malformed lease"
+        if blocked:
+            return _refusal(blocked)
         updates = [(key, olds.get(key), self._make_lease_commit(key, ttl, note)) for key in keys]
         return self._cas_push_refs(updates, atomic=True)
 
@@ -407,13 +421,19 @@ def renew_many(self, keys: Sequence[str], *, ttl: int | float = CLAIM_TTL_S) ->
         self._ensure_scratch()
         olds = self._remote_oids(keys)
         notes: dict[str, str] = {}
+        blocked: dict[str, str] = {}
         for key in keys:
-            lease = self._owned_lease(key, olds.get(key))
-            if lease is not None:
+            try:
+                lease = self._owned_lease(key, olds.get(key))
+            except MalformedLeaseError:
+                blocked[key] = "malformed lease"
+                continue
+            if lease is None:
+                blocked[key] = "not held by this worker"
+            else:
                 notes[key] = str(lease.get("note", ""))
-        lost = tuple(key for key in keys if key not in notes)
-        if lost:
-            return ClaimBatchResult(False, lost, "not held by this worker")
+        if blocked:
+            return _refusal(blocked)
         updates = [(key, olds[key], self._make_lease_commit(key, ttl, notes[key])) for key in keys]
         return self._cas_push_refs(updates, atomic=True)
 
@@ -423,9 +443,15 @@ def release_many(self, keys: Sequence[str]) -> ClaimBatchResult:
         self._ensure_scratch()
         olds = self._remote_oids(keys)
         present = [key for key in keys if key in olds]
-        foreign = tuple(key for key in present if self._owned_lease(key, olds[key]) is None)
-        if foreign:
-            return ClaimBatchResult(False, foreign, "not held by this worker")
+        blocked: dict[str, str] = {}
+        for key in present:
+            try:
+                if self._owned_lease(key, olds[key]) is None:
+                    blocked[key] = "not held by this worker"
+            except MalformedLeaseError:
+                blocked[key] = "malformed lease"
+        if blocked:
+            return _refusal(blocked)
         if not present:
             return ClaimBatchResult(True)
         return self._cas_push_refs([(key, olds[key], "") for key in present], atomic=True)
diff --git a/tests/test_claim_cli.py b/tests/test_claim_cli.py
index d108066d..a09d4803 100644
--- a/tests/test_claim_cli.py
+++ b/tests/test_claim_cli.py
@@ -174,6 +174,10 @@ def test_claim_cli_multi_node_failure_changes_no_claim(tmp_path: Path, capsys) -
 
     assert main(_args(repo, scratch, "acquire", "a")) == 0
     capsys.readouterr()
+    assert main(_args(repo, scratch, "renew", "a", "b")) == 1
+    assert capsys.readouterr().out == (
+        "error: could not renew a, b; no claim was renewed: not held by this worker: b\n"
+    )
     assert main(_args(repo, scratch, "release", "a", "b")) == 1
     assert capsys.readouterr().out == (
         "error: could not release a, b; no claim was released: not held by this worker: b\n"
@@ -181,6 +185,15 @@ def test_claim_cli_multi_node_failure_changes_no_claim(tmp_path: Path, capsys) -
     assert _claim_refs(repo) == sorted(CLAIM_REF_PREFIX + author_claim_key(node) for node in ("a", "b"))
 
 
+def test_claim_cli_multi_node_names_a_malformed_claim(tmp_path: Path, capsys) -> None:
+    repo = _bare_repo(tmp_path)
+    _plant_message(repo, author_claim_key("b"), "not json")
+
+    assert main(_args(repo, tmp_path / "scratch", "acquire", "a", "b")) == 1
+    assert capsys.readouterr().out == "error: could not acquire a, b; no claim was acquired: malformed lease: b\n"
+    assert _claim_refs(repo) == [CLAIM_REF_PREFIX + author_claim_key("b")]
+
+
 def test_claim_cli_rejects_duplicate_targets_before_touching_the_board(tmp_path: Path, capsys) -> None:
     repo = _bare_repo(tmp_path)
     scratch = tmp_path / "scratch"
diff --git a/tests/test_claims.py b/tests/test_claims.py
index 99fd5fdf..84b3c939 100644
--- a/tests/test_claims.py
+++ b/tests/test_claims.py
@@ -653,6 +653,34 @@ def peer_steals_first(updates: Sequence[tuple[str, str | None, str]], *, atomic:
     assert board.holds("kept")
 
 
+@pytest.mark.parametrize("method", ["acquire_many", "renew_many", "release_many"])
+def test_batch_names_a_malformed_lease_as_blocking_and_changes_nothing(
+    tmp_path: Path, board_repo: Path, method: str
+) -> None:
+    board = _board(tmp_path, board_repo, "worker-a")
+    assert board.acquire("mine", ttl=600)
+    _plant_message(board_repo, "broken", "not json")
+    before = _refs(board_repo)
+
+    result = getattr(board, method)(["mine", "broken"])
+
+    assert result == claims.ClaimBatchResult(False, ("broken",), "malformed lease")
+    assert _refs(board_repo) == before
+
+
+def test_acquire_many_names_every_blocking_key_in_batch_order(tmp_path: Path, board_repo: Path) -> None:
+    peer = _board(tmp_path, board_repo, "worker-b")
+    assert peer.acquire("held", ttl=600)
+    _plant_message(board_repo, "broken", "not json")
+    before = _refs(board_repo)
+    board = _board(tmp_path, board_repo, "worker-a")
+
+    result = board.acquire_many(["broken", "free", "held"], ttl=600)
+
+    assert result == claims.ClaimBatchResult(False, ("broken", "held"), "malformed lease or held by another worker")
+    assert _refs(board_repo) == before
+
+
 @pytest.mark.parametrize("method", ["acquire_many", "renew_many", "release_many"])
 @pytest.mark.parametrize(
     ("keys", "error", "match"),

From d73846537665e9b1d5a574a0ff648b95f78c91ac Mon Sep 17 00:00:00 2001
From: Jack McCarthy <Deicyde@users.noreply.github.com>
Date: Mon, 5 Oct 2026 15:14:22 -0400
Subject: [PATCH 3/3] Document atomic claim batches

---
 autoform_cli/README.md | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/autoform_cli/README.md b/autoform_cli/README.md
index d5db721b..32c1dc30 100644
--- a/autoform_cli/README.md
+++ b/autoform_cli/README.md
@@ -554,6 +554,13 @@ Leases expire after 1500 seconds unless `--ttl` sets another length. Renew well
 within that, and confirm a claim is still held with `renew`, not `acquire`,
 which also succeeds once a lease has expired or been released.
 
+Pass several targets to `acquire`, `renew`, or `release` when one change needs
+shared ownership. The board reads the set once and sends one atomic push with a
+lease for every ref, so either every target changes or none does. A remote that
+cannot push atomically is refused, and duplicate or malformed targets fail
+before any push. To avoid deadlock, acquire the complete set in one command;
+never hold a partial set while waiting for another target.
+
 Claims are fail-closed compare-and-swap leases under
 `refs/autoform-claims/` on the Git `origin`; pass `--repo` for another claim
 board. A failed acquire or renew means the caller cannot prove ownership and