diff --git a/pr_split/cli.py b/pr_split/cli.py
index 5a8656a..709f2c9 100644
--- a/pr_split/cli.py
+++ b/pr_split/cli.py
@@ -64,6 +64,7 @@
     fetch_fork_pr,
     is_worktree_clean,
     merge_base,
+    prune_remote_tracking_refs,
     push_branch,
     remove_worktree,
 )
@@ -461,7 +462,15 @@ def _create_branches_and_commits(
     *,
     author: str | None = None,
     stacked: bool = False,
+    keep: dict[str, BranchRecord] | None = None,
 ) -> list[BranchRecord]:
+    """Create a branch and commit per group; groups in ``keep`` reuse their record.
+
+    ``keep`` holds groups whose PRs already exist from an earlier, partly
+    failed run: their branches are left as they are (recreating them would
+    rewrite the open PR's head).
+    """
+    kept = keep or {}
     worktree_base = Path(tempfile.mkdtemp(prefix="pr-split-worktrees-"))
     step = load_per_group_step()
     order = {gid: i for i, gid in enumerate(PlanDAG(groups).topological_order(), start=1)}
@@ -482,9 +491,10 @@ def _create_branches_and_commits(
         batches = iter([[(group, base_branch, merge_base_ref) for group in groups]])
 
     try:
-        results: dict[str, BranchRecord] = {}
+        results: dict[str, BranchRecord] = dict(kept)
         errors: list[tuple[str, Exception]] = []
         for batch_args in batches:
+            batch_args = [args for args in batch_args if args[0].id not in kept]
             with ThreadPoolExecutor(max_workers=_WORKTREE_MAX_WORKERS) as executor:
                 future_to_group_id = {
                     executor.submit(
@@ -514,7 +524,9 @@ def _create_branches_and_commits(
         if errors:
             # Failed groups already removed their own branch inside the
             # worker; only the successful ones remain to roll back.
-            for record in results.values():
+            for gid, record in results.items():
+                if gid in kept:
+                    continue
                 try:
                     delete_branch(record.branch_name)
                 except PRSplitError as exc:
@@ -625,17 +637,26 @@ def _push_and_create_prs(
     branch_records: list[BranchRecord],
     *,
     draft: bool = False,
+    existing_prs: dict[str, PRRecord] | None = None,
 ) -> list[PRRecord]:
+    """Push each group's branch and open its PR; groups in ``existing_prs`` are done."""
+    done = existing_prs or {}
     record_map = {r.group_id: r for r in branch_records}
     errors: list[tuple[str, Exception]] = []
 
+    # Branch names are reused across runs; drop tracking refs left by an
+    # earlier split whose remote branch was merged or deleted since, or
+    # force-with-lease rejects every push as stale.
+    prune_remote_tracking_refs()
+
     # Children target parent branches, so every branch is pushed before any PR opens.
     with ThreadPoolExecutor(max_workers=_PUSH_MAX_WORKERS) as executor:
         push_futures = {
             executor.submit(push_branch, record_map[group.id].branch_name): group.id
             for group in groups
+            if group.id not in done
         }
-        pushed: set[str] = set()
+        pushed: set[str] = set(done)
         for future in as_completed(push_futures):
             group_id = push_futures[future]
             try:
@@ -670,9 +691,9 @@ def _base_pushed(group: Group) -> bool:
                 _create_single_pr, group, record_map[group.id], groups, draft=draft
             ): group.id
             for group in groups
-            if group.id in pushed and _base_pushed(group)
+            if group.id not in done and group.id in pushed and _base_pushed(group)
         }
-        results: dict[str, PRRecord] = {}
+        results: dict[str, PRRecord] = dict(done)
         for future in as_completed(future_to_group_id):
             group_id = future_to_group_id[future]
             try:
@@ -1880,10 +1901,25 @@ def execute(
     if draft and not plan.draft:
         plan = plan.model_copy(update={"draft": True})
 
-    if plan_file.git_state.prs:
+    existing_prs = {r.group_id: r for r in plan_file.git_state.prs}
+    recorded = {r.group_id: r for r in plan_file.git_state.branches}
+    kept_branches = {gid: recorded[gid] for gid in existing_prs if gid in recorded}
+    group_ids = {g.id for g in plan.groups}
+    if existing_prs and (
+        set(existing_prs) >= group_ids
+        or not set(existing_prs) <= group_ids
+        or set(kept_branches) != set(existing_prs)
+    ):
+        # Every group already has a PR, or the recorded PRs no longer match
+        # the plan's groups: there is nothing safe to resume.
         console.print("[red]This plan already has PRs. Use 'pr-split clean' first.[/red]")
         raise typer.Exit(1)
-    if plan_file.git_state.branches:
+    if existing_prs:
+        console.print(
+            f"[yellow]A previous run opened {len(existing_prs)} of {len(plan.groups)} PR(s)."
+            " Keeping those and creating the rest.[/yellow]"
+        )
+    elif plan_file.git_state.branches:
         # A previous execute created branches but no PRs - a failed push or PR
         # creation. Branch creation is idempotent (add_worktree recreates an
         # existing branch), so retry instead of forcing 'clean' + a full
@@ -1942,6 +1978,16 @@ def execute(
         console.print(f"[red]{exc}[/red]")
         raise typer.Exit(1) from exc
 
+    # Any plan with dependency edges is laid out along its DAG (stacked or not),
+    # so a kept layer's branch sits on its ancestors' current commits; rebuilding
+    # an ancestor would rewrite the base under the kept layer's open PR, which
+    # would then show the ancestor's changes again. Their PRs are still opened.
+    if plan.stacked or any(g.depends_on for g in plan.groups):
+        for gid in list(kept_branches):
+            for ancestor in dag.ancestors(gid):
+                if ancestor in recorded:
+                    kept_branches.setdefault(ancestor, recorded[ancestor])
+
     _present_plan(plan.groups)
     _report_oversized_groups(
         plan.groups, plan.max_loc, {pf.path: len(pf) for pf in parsed_diff.patch_set}
@@ -1960,12 +2006,15 @@ def execute(
             namespace,
             author=plan.author,
             stacked=plan.stacked,
+            keep=kept_branches,
         )
     except PRSplitError as exc:
         console.print(f"[red]{exc}[/red]")
         raise typer.Exit(1) from exc
     try:
-        pr_records = _push_and_create_prs(plan.groups, branch_records, draft=plan.draft)
+        pr_records = _push_and_create_prs(
+            plan.groups, branch_records, draft=plan.draft, existing_prs=existing_prs
+        )
     except PRCreationError as exc:
         save_plan(
             PlanFile(
diff --git a/pr_split/exceptions.py b/pr_split/exceptions.py
index fb208aa..55ac79c 100644
--- a/pr_split/exceptions.py
+++ b/pr_split/exceptions.py
@@ -102,6 +102,11 @@ class ErrorMsg(StrEnum):
         "Cannot tell which base branch the stack under '{prefix}' targets ({bases}); pass --base"
     )
     RECOVER_PR_LIST_FAILED = "Cannot list PRs to rebuild the plan ({detail})"
+    LLM_TOKEN_COUNT_FAILED = "Could not count prompt tokens with the LLM API: {detail}"
+    LLM_KEY_REJECTED = (
+        "{variable} was rejected by the LLM API; set a valid key, or plan without one"
+        " using --partition-strategy graph"
+    )
     BRANCH_CREATE_FAILED = "Failed to create branch '{branch}': {detail}"
     PR_CREATE_FAILED = "Failed to create PR for group '{group}': {detail}"
     MERGE_FAILED = "Merge of '{source}' into '{target}' failed: {detail}"
diff --git a/pr_split/git_ops/__init__.py b/pr_split/git_ops/__init__.py
index 2aae56d..0e076c0 100644
--- a/pr_split/git_ops/__init__.py
+++ b/pr_split/git_ops/__init__.py
@@ -25,6 +25,9 @@
 from .branches import (
     merge_base as merge_base,
 )
+from .branches import (
+    prune_remote_tracking_refs as prune_remote_tracking_refs,
+)
 from .branches import (
     push_branch as push_branch,
 )
diff --git a/pr_split/git_ops/branches.py b/pr_split/git_ops/branches.py
index f78fa00..41b4f3e 100644
--- a/pr_split/git_ops/branches.py
+++ b/pr_split/git_ops/branches.py
@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import contextlib
 import os
 import re
 import subprocess
+import time
 
 from loguru import logger
 
@@ -93,9 +95,50 @@ def is_worktree_clean() -> bool:
     return all(line.startswith("??") for line in output.splitlines())
 
 
+# Server-side failures GitHub reports for a push that can succeed on retry.
+_TRANSIENT_PUSH_ERRORS = (
+    "fatal error in commit_refs",
+    "the remote end hung up unexpectedly",
+    "internal server error",
+    "http 500",
+    "http 502",
+    "http 503",
+    "connection reset",
+    "operation timed out",
+)
+_PUSH_ATTEMPTS = 3
+_PUSH_RETRY_DELAY = 2.0
+
+
+def _remote_has_local_head(branch: str) -> bool:
+    try:
+        remote = run_git("ls-remote", "origin", f"refs/heads/{branch}").split()
+        return bool(remote) and remote[0] == run_git("rev-parse", branch)
+    except GitOperationError:
+        return False
+
+
 def push_branch(branch: str) -> None:
     logger.info(logs.PUSHING_BRANCH.format(branch=branch))
-    run_git("push", "--force-with-lease", "-u", "origin", branch)
+    retried = False
+    for attempt in range(1, _PUSH_ATTEMPTS + 1):
+        try:
+            run_git("push", "--force-with-lease", "-u", "origin", branch)
+            return
+        except GitOperationError as exc:
+            transient = any(marker in str(exc).lower() for marker in _TRANSIENT_PUSH_ERRORS)
+            # A push reported as failed (hung-up remote, timeout) may still have
+            # landed; the retry is then rejected as a stale lease although the
+            # branch is already on the remote.
+            if retried and not transient and _remote_has_local_head(branch):
+                return
+            if not transient or attempt == _PUSH_ATTEMPTS:
+                raise
+            retried = True
+            logger.warning(
+                logs.PUSH_RETRY.format(branch=branch, attempt=attempt, error=str(exc).strip())
+            )
+            time.sleep(_PUSH_RETRY_DELAY * attempt)
 
 
 _LOCAL_BRANCH_MISSING = "not found"
@@ -129,10 +172,36 @@ def delete_branch(branch: str, *, remote: bool = False) -> None:
             if _REMOTE_REF_MISSING not in str(exc):
                 raise
             logger.info(logs.BRANCH_ALREADY_GONE.format(branch=branch, where=" on origin"))
+        # Either way the remote branch is gone; drop the local tracking ref
+        # too, or the next `push --force-with-lease` of a reused branch name
+        # is rejected as "stale info" against a ref origin no longer has.
+        forget_remote_tracking_ref(branch)
     if local_error is not None:
         raise local_error
 
 
+def forget_remote_tracking_ref(branch: str) -> None:
+    with contextlib.suppress(GitOperationError):
+        run_git("update-ref", "-d", f"refs/remotes/origin/{branch}")
+
+
+def prune_remote_tracking_refs() -> None:
+    """Drop tracking refs whose remote branch no longer exists.
+
+    Split branch names are reused across runs; after a merge or `clean` the
+    remote branch is gone but `refs/remotes/origin/pr-split/...` may still
+    point at the old head, and every `push --force-with-lease` would be
+    rejected as stale. Only *gone* refs are dropped (`git remote prune`, no
+    fetch): refreshing the refs that still exist would make the lease
+    compare against origin's current tip and silently force over commits
+    someone else pushed to a reused branch.
+    """
+    try:
+        run_git("remote", "prune", "origin")
+    except GitOperationError as exc:
+        logger.warning(logs.PRUNE_FAILED.format(error=exc))
+
+
 def merge_base(ref_a: str, ref_b: str) -> str:
     return run_git("merge-base", ref_a, ref_b)
 
diff --git a/pr_split/logs.py b/pr_split/logs.py
index 2fee79b..07f16cb 100644
--- a/pr_split/logs.py
+++ b/pr_split/logs.py
@@ -148,6 +148,7 @@
     "CP-SAT stopped at the {timeout:g}s limit with a feasible but unproven-optimal plan "
     "({units} units, {groups} groups); raise --cp-sat-timeout for a better partition"
 )
+PUSH_RETRY = "Push of {branch} failed (attempt {attempt}): {error}; retrying"
 BASE_ALREADY_MERGED = (
     "The split's base branch {base} has already merged; run 'pr-split retarget' to move"
     " the split onto the new base before merging"
@@ -161,3 +162,7 @@
     " its sub-PR will not compile or import on its own"
 )
 ADOPTED_REMOTE_BRANCH = "Created local branch {branch} from {remote_ref}"
+PRUNE_FAILED = (
+    "Could not prune origin's tracking refs before pushing ({error}); "
+    "a reused branch name may be rejected as stale"
+)
diff --git a/pr_split/planner/client.py b/pr_split/planner/client.py
index 2a8b44d..0288b4d 100644
--- a/pr_split/planner/client.py
+++ b/pr_split/planner/client.py
@@ -105,12 +105,20 @@ def _extract_raw_output(block_input: dict[str, object]) -> list[_RawGroup]:
 
 def _count_tokens_anthropic(system: str, user: str, *, settings: Settings) -> int:
     client = anthropic.Anthropic(api_key=settings.api_key)
-    response = client.messages.count_tokens(
-        model=settings.model,
-        system=system,
-        messages=[{"role": "user", "content": user}],
-        tools=[_ANTHROPIC_TOOL_DEF],
-    )
+    # This is the first API call of a run, so a bad key, a network blip or a
+    # rate limit surfaces here; it must be an LLMError like every other
+    # provider failure rather than an SDK traceback.
+    try:
+        response = client.messages.count_tokens(
+            model=settings.model,
+            system=system,
+            messages=[{"role": "user", "content": user}],
+            tools=[_ANTHROPIC_TOOL_DEF],
+        )
+    except anthropic.AuthenticationError as exc:
+        raise LLMError(ErrorMsg.LLM_KEY_REJECTED(variable="ANTHROPIC_API_KEY")) from exc
+    except anthropic.APIError as exc:
+        raise LLMError(ErrorMsg.LLM_TOKEN_COUNT_FAILED(detail=str(exc))) from exc
     return response.input_tokens
 
 
diff --git a/tests/network_guards.py b/tests/network_guards.py
index 15df9ca..54b7fc7 100644
--- a/tests/network_guards.py
+++ b/tests/network_guards.py
@@ -13,6 +13,14 @@ def _no_base_fetch() -> Iterator[None]:
         yield
 
 
+@pytest.fixture(autouse=True)
+def _no_remote_prune() -> Iterator[None]:
+    """Pushing prunes origin's tracking refs first; keep that off the network
+    (and off this checkout's refs). Tests that exercise it patch it again."""
+    with patch("pr_split.cli.prune_remote_tracking_refs"):
+        yield
+
+
 @pytest.fixture(autouse=True)
 def _no_selected_plan() -> Iterator[None]:
     """A command's plan selection is process-wide; start each test without one."""
diff --git a/tests/test_cli_coverage.py b/tests/test_cli_coverage.py
index aa1d080..cb5bfe0 100644
--- a/tests/test_cli_coverage.py
+++ b/tests/test_cli_coverage.py
@@ -1035,12 +1035,139 @@ def test_branches_without_prs_is_retried(
         with patch("pr_split.cli.validate_coverage"):
             result = runner.invoke(app, ["execute"])
 
-        assert result.exit_code == 0
+        assert result.exit_code == 0, result.output
         assert "Recreating them and retrying" in result.output.replace("\n", " ")
         mock_create.assert_called_once()
         mock_push.assert_called_once()
         mock_save.assert_called_once()
 
+    @staticmethod
+    def _plan_with_one_of_two_prs(pr_groups: list[str]) -> object:
+        from pr_split.constants import Priority
+        from pr_split.schemas import BranchRecord, GitState, PlanFile, PRRecord, SplitPlan
+
+        return PlanFile(
+            plan=SplitPlan(
+                dev_branch="feature-branch",
+                base_branch="main",
+                max_loc=400,
+                priority=Priority.ORTHOGONAL,
+                merge_base_sha="0123456789abcdef",
+                raw_diff="some diff",
+                groups=[
+                    _group("pr-1", "feat: a", files=["a.py"]),
+                    _group("pr-2", "feat: b", files=["b.py"]),
+                ],
+            ),
+            git_state=GitState(
+                branches=[
+                    BranchRecord(
+                        group_id=gid,
+                        branch_name=f"pr-split/feature-branch/{gid}",
+                        base_branch="main",
+                        commit_sha="abc123",
+                    )
+                    for gid in ("pr-1", "pr-2")
+                ],
+                prs=[PRRecord(group_id=gid, pr_number=10, pr_url="u") for gid in pr_groups],
+            ),
+        )
+
+    @patch("pr_split.cli.typer.confirm", return_value=True)
+    @patch("pr_split.cli.save_plan")
+    @patch("pr_split.cli._push_and_create_prs", return_value=[])
+    @patch("pr_split.cli._create_branches_and_commits", return_value=[])
+    @patch("pr_split.cli.commit_exists", return_value=True)
+    @patch("pr_split.cli.parse_diff")
+    @patch("pr_split.cli.validate_coverage")
+    @patch("pr_split.cli.is_worktree_clean", return_value=True)
+    @patch("pr_split.cli.check_gh_auth", return_value=True)
+    @patch("pr_split.cli.branch_exists", return_value=True)
+    @patch("pr_split.cli.load_plan")
+    @patch("pr_split.cli.plan_exists", return_value=True)
+    def test_partial_prs_are_kept_and_the_rest_created(
+        self,
+        mock_pe: MagicMock,
+        mock_load: MagicMock,
+        mock_be: MagicMock,
+        mock_auth: MagicMock,
+        mock_clean: MagicMock,
+        mock_validate: MagicMock,
+        mock_parse: MagicMock,
+        mock_commit: MagicMock,
+        mock_create: MagicMock,
+        mock_push: MagicMock,
+        mock_save: MagicMock,
+        mock_confirm: MagicMock,
+    ) -> None:
+        mock_load.return_value = self._plan_with_one_of_two_prs(["pr-1"])
+
+        result = runner.invoke(app, ["execute"])
+
+        assert result.exit_code == 0, result.output
+        assert "opened 1 of 2 PR(s)" in result.output.replace("\n", " ")
+        assert set(mock_create.call_args.kwargs["keep"]) == {"pr-1"}
+        assert set(mock_push.call_args.kwargs["existing_prs"]) == {"pr-1"}
+
+    @pytest.mark.parametrize("stacked", [True, False])
+    @patch("pr_split.cli._link_stacks")
+    @patch("pr_split.cli._require_gh_stack")
+    @patch("pr_split.cli.typer.confirm", return_value=True)
+    @patch("pr_split.cli.save_plan")
+    @patch("pr_split.cli._push_and_create_prs", return_value=[])
+    @patch("pr_split.cli._create_branches_and_commits", return_value=[])
+    @patch("pr_split.cli.commit_exists", return_value=True)
+    @patch("pr_split.cli.parse_diff")
+    @patch("pr_split.cli.validate_coverage")
+    @patch("pr_split.cli.is_worktree_clean", return_value=True)
+    @patch("pr_split.cli.check_gh_auth", return_value=True)
+    @patch("pr_split.cli.branch_exists", return_value=True)
+    @patch("pr_split.cli.load_plan")
+    @patch("pr_split.cli.plan_exists", return_value=True)
+    def test_stacked_resume_keeps_the_parents_of_a_layer_with_a_pr(
+        self,
+        mock_pe: MagicMock,
+        mock_load: MagicMock,
+        mock_be: MagicMock,
+        mock_auth: MagicMock,
+        mock_clean: MagicMock,
+        mock_validate: MagicMock,
+        mock_parse: MagicMock,
+        mock_commit: MagicMock,
+        mock_create: MagicMock,
+        mock_push: MagicMock,
+        mock_save: MagicMock,
+        mock_confirm: MagicMock,
+        mock_stack: MagicMock,
+        mock_link: MagicMock,
+        stacked: bool,
+    ) -> None:
+        """pr-2's PR opened but its parent pr-1's did not: rebuilding pr-1 would
+        rewrite the base under pr-2's open PR, so the resume keeps it. A plan
+        with dependencies is laid out along its DAG with or without --stack."""
+        plan_file = self._plan_with_one_of_two_prs(["pr-2"])
+        plan_file.plan.groups[1].depends_on = ["pr-1"]
+        plan_file.plan.stacked = stacked
+        mock_load.return_value = plan_file
+
+        result = runner.invoke(app, ["execute"])
+
+        assert result.exit_code == 0, result.output
+        kept = set(mock_create.call_args.kwargs["keep"])
+        assert kept == {"pr-1", "pr-2"}
+        # pr-1 still gets its PR opened.
+        assert set(mock_push.call_args.kwargs["existing_prs"]) == {"pr-2"}
+
+    @patch("pr_split.cli.load_plan")
+    @patch("pr_split.cli.plan_exists", return_value=True)
+    def test_every_group_with_a_pr_is_refused(
+        self, mock_pe: MagicMock, mock_load: MagicMock
+    ) -> None:
+        mock_load.return_value = self._plan_with_one_of_two_prs(["pr-1", "pr-2"])
+        result = runner.invoke(app, ["execute"])
+        assert result.exit_code == 1
+        assert "already has PRs" in result.output
+
     @staticmethod
     def _plan_with_stale_pr2() -> object:
         from pr_split.constants import Priority
diff --git a/tests/test_cli_helpers.py b/tests/test_cli_helpers.py
index a1b6137..b099603 100644
--- a/tests/test_cli_helpers.py
+++ b/tests/test_cli_helpers.py
@@ -179,6 +179,30 @@ def test_returns_records_for_all_groups(
         assert result[0].group_id == "pr-1"
         assert result[1].group_id == "pr-2"
 
+    @patch("pr_split.cli.create_pr", return_value=(2, "https://github.com/pr/2"))
+    @patch("pr_split.cli.push_branch")
+    def test_groups_with_existing_prs_are_not_pushed_or_reopened(
+        self, mock_push: MagicMock, mock_create: MagicMock
+    ) -> None:
+        groups = [_group("pr-1", "feat: a"), _group("pr-2", "feat: b", ["pr-1"])]
+        records = [
+            _branch_record("pr-1", "pr-split/ns/pr-1"),
+            BranchRecord(
+                group_id="pr-2",
+                branch_name="pr-split/ns/pr-2",
+                base_branch="pr-split/ns/pr-1",
+                commit_sha="abc123",
+            ),
+        ]
+        existing = PRRecord(group_id="pr-1", pr_number=1, pr_url="https://github.com/pr/1")
+
+        result = _push_and_create_prs(groups, records, existing_prs={"pr-1": existing})
+
+        assert [call.args[0] for call in mock_push.call_args_list] == ["pr-split/ns/pr-2"]
+        mock_create.assert_called_once()
+        assert result[0] is existing
+        assert result[1].pr_number == 2
+
     @patch("pr_split.cli.create_pr", return_value=(1, "https://github.com/pr/1"))
     @patch("pr_split.cli.push_branch")
     def test_pushes_all_branches(self, mock_push: MagicMock, mock_create: MagicMock) -> None:
@@ -738,6 +762,22 @@ def push(branch: str) -> None:
         assert mock_create.call_count == 0
 
 
+class TestPushPrunesFirst:
+    @patch("pr_split.cli.create_pr", return_value=(1, "https://github.com/pr/1"))
+    @patch("pr_split.cli.push_branch")
+    @patch("pr_split.cli.prune_remote_tracking_refs")
+    def test_tracking_refs_are_refreshed_before_any_push(
+        self, mock_prune: MagicMock, mock_push: MagicMock, mock_create: MagicMock
+    ) -> None:
+        order: list[str] = []
+        mock_prune.side_effect = lambda: order.append("prune")
+        mock_push.side_effect = lambda branch: order.append("push")
+        _push_and_create_prs(
+            [_group("pr-1", "feat: a")], [_branch_record("pr-1", "pr-split/ns/pr-1")]
+        )
+        assert order == ["prune", "push"]
+
+
 class TestWorktreeWritesPreserveCrlf:
     @patch("pr_split.cli.commit_files_in_dir", return_value="sha1")
     @patch("pr_split.cli.materialize_group_files", return_value={"c.txt": "a\r\nb\r\n"})
diff --git a/tests/test_cli_new_features.py b/tests/test_cli_new_features.py
index 4987110..1e578b4 100644
--- a/tests/test_cli_new_features.py
+++ b/tests/test_cli_new_features.py
@@ -681,6 +681,7 @@ def test_close_failure_warning_includes_the_reason(
 
 
 class TestCleanupAfterMergeDeletedBranches:
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.cli.get_pr_state", return_value={"state": "MERGED"})
     @patch("pr_split.cli.shutil.rmtree")
     @patch("pr_split.cli.plan_path")
@@ -693,6 +694,7 @@ def test_fully_merged_split_cleans_up_completely(
         mock_path: MagicMock,
         mock_rmtree: MagicMock,
         mock_state: MagicMock,
+        mock_forget: MagicMock,
     ) -> None:
         mock_path.return_value.exists.return_value = True
         # merge --delete-branch already removed both local and remote branches
diff --git a/tests/test_client.py b/tests/test_client.py
index 72ee2a0..c705628 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -4,6 +4,7 @@
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
+import anthropic
 import pytest
 
 from pr_split.config import Settings
@@ -762,6 +763,26 @@ def test_returns_input_tokens(self, mock_cls: MagicMock) -> None:
         assert result == 42
         mock_client.messages.count_tokens.assert_called_once()
 
+    @patch("pr_split.planner.client.anthropic.Anthropic")
+    def test_api_failure_is_an_llm_error(self, mock_cls: MagicMock) -> None:
+        mock_client = mock_cls.return_value
+        mock_client.messages.count_tokens.side_effect = anthropic.APIConnectionError(
+            request=MagicMock()
+        )
+        settings = _make_settings(Provider.ANTHROPIC)
+        with pytest.raises(LLMError, match="Could not count prompt tokens"):
+            _count_tokens_anthropic("sys", "usr", settings=settings)
+
+    @patch("pr_split.planner.client.anthropic.Anthropic")
+    def test_rejected_key_names_the_variable(self, mock_cls: MagicMock) -> None:
+        response = MagicMock(status_code=401, headers={})
+        mock_cls.return_value.messages.count_tokens.side_effect = anthropic.AuthenticationError(
+            "API key is invalid.", response=response, body=None
+        )
+        settings = _make_settings(Provider.ANTHROPIC)
+        with pytest.raises(LLMError, match="ANTHROPIC_API_KEY was rejected"):
+            _count_tokens_anthropic("sys", "usr", settings=settings)
+
 
 # ---------------------------------------------------------------------------
 # _count_tokens_openai
diff --git a/tests/test_git_branches.py b/tests/test_git_branches.py
index 8823ccd..318f647 100644
--- a/tests/test_git_branches.py
+++ b/tests/test_git_branches.py
@@ -90,6 +90,90 @@ def test_calls_push(self, mock_git: MagicMock) -> None:
         )
 
 
+class TestPushBranchRetries:
+    @patch("pr_split.git_ops.branches.time.sleep")
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_transient_server_error_is_retried(
+        self, mock_git: MagicMock, mock_sleep: MagicMock
+    ) -> None:
+        mock_git.side_effect = [
+            GitOperationError("remote: fatal error in commit_refs"),
+            "",
+        ]
+        push_branch("pr-split/pr-5")
+        assert mock_git.call_count == 2
+        mock_sleep.assert_called_once()
+
+    @patch("pr_split.git_ops.branches.time.sleep")
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_rejection_is_not_retried(self, mock_git: MagicMock, mock_sleep: MagicMock) -> None:
+        mock_git.side_effect = GitOperationError("! [rejected] (stale info)")
+        with pytest.raises(GitOperationError, match="stale info"):
+            push_branch("pr-split/pr-5")
+        assert mock_git.call_count == 1
+        mock_sleep.assert_not_called()
+
+    @patch("pr_split.git_ops.branches.time.sleep")
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_gives_up_after_the_last_attempt(
+        self, mock_git: MagicMock, mock_sleep: MagicMock
+    ) -> None:
+        mock_git.side_effect = GitOperationError("fatal: the remote end hung up unexpectedly")
+        with pytest.raises(GitOperationError, match="hung up"):
+            push_branch("pr-split/pr-5")
+        assert mock_git.call_count == 3
+
+
+class TestPushThatLandedDespiteAnError:
+    @patch("pr_split.git_ops.branches.time.sleep")
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_stale_lease_after_a_landed_push_is_success(
+        self, mock_git: MagicMock, mock_sleep: MagicMock
+    ) -> None:
+        def git(*args: str) -> str:
+            if args[0] == "push":
+                if git.pushes == 0:
+                    git.pushes += 1
+                    raise GitOperationError("fatal: the remote end hung up unexpectedly")
+                raise GitOperationError("! [rejected] pr-split/pr-5 (stale info)")
+            if args[0] == "ls-remote":
+                return "abc123\trefs/heads/pr-split/pr-5"
+            return "abc123"  # rev-parse
+
+        git.pushes = 0  # type: ignore[attr-defined]
+        mock_git.side_effect = git
+        push_branch("pr-split/pr-5")
+
+    @patch("pr_split.git_ops.branches.time.sleep")
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_stale_lease_is_still_an_error_when_the_remote_differs(
+        self, mock_git: MagicMock, mock_sleep: MagicMock
+    ) -> None:
+        def git(*args: str) -> str:
+            if args[0] == "push":
+                if git.pushes == 0:
+                    git.pushes += 1
+                    raise GitOperationError("fatal: the remote end hung up unexpectedly")
+                raise GitOperationError("! [rejected] pr-split/pr-5 (stale info)")
+            if args[0] == "ls-remote":
+                return "fff999\trefs/heads/pr-split/pr-5"
+            return "abc123"
+
+        git.pushes = 0  # type: ignore[attr-defined]
+        mock_git.side_effect = git
+        with pytest.raises(GitOperationError, match="stale info"):
+            push_branch("pr-split/pr-5")
+
+    @patch("pr_split.git_ops.branches.run_git")
+    def test_a_first_attempt_rejection_does_not_consult_the_remote(
+        self, mock_git: MagicMock
+    ) -> None:
+        mock_git.side_effect = GitOperationError("! [rejected] (stale info)")
+        with pytest.raises(GitOperationError):
+            push_branch("pr-split/pr-5")
+        assert [c.args[0] for c in mock_git.call_args_list] == ["push"]
+
+
 class TestDeleteBranch:
     @patch("pr_split.git_ops.branches.run_git")
     def test_local_only(self, mock_git: MagicMock) -> None:
@@ -97,14 +181,18 @@ def test_local_only(self, mock_git: MagicMock) -> None:
         delete_branch("pr-split/pr-1")
         mock_git.assert_called_once_with("branch", "-D", "pr-split/pr-1")
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_with_remote(self, mock_git: MagicMock) -> None:
+    def test_with_remote(self, mock_git: MagicMock, mock_forget: MagicMock) -> None:
         mock_git.return_value = ""
         delete_branch("pr-split/pr-1", remote=True)
         assert mock_git.call_count == 2
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_local_failure_still_deletes_remote(self, mock_git: MagicMock) -> None:
+    def test_local_failure_still_deletes_remote(
+        self, mock_git: MagicMock, mock_forget: MagicMock
+    ) -> None:
         mock_git.side_effect = [GitOperationError("checked out"), ""]
         with pytest.raises(GitOperationError, match="checked out"):
             delete_branch("pr-split/pr-1", remote=True)
@@ -117,15 +205,21 @@ def test_local_failure_without_remote_raises_immediately(self, mock_git: MagicMo
             delete_branch("pr-split/pr-1")
         mock_git.assert_called_once()
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_branch_already_deleted_locally_counts_as_deleted(self, mock_git: MagicMock) -> None:
+    def test_branch_already_deleted_locally_counts_as_deleted(
+        self, mock_git: MagicMock, mock_forget: MagicMock
+    ) -> None:
         # `pr-split merge` deletes the local branch; cleanup must not fail on it.
         mock_git.side_effect = [GitOperationError("error: branch 'pr-split/pr-1' not found."), ""]
         delete_branch("pr-split/pr-1", remote=True)
         mock_git.assert_any_call("push", "origin", "--delete", "pr-split/pr-1")
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_branch_already_deleted_on_origin_counts_as_deleted(self, mock_git: MagicMock) -> None:
+    def test_branch_already_deleted_on_origin_counts_as_deleted(
+        self, mock_git: MagicMock, mock_forget: MagicMock
+    ) -> None:
         mock_git.side_effect = [
             "",
             GitOperationError(
@@ -134,8 +228,11 @@ def test_branch_already_deleted_on_origin_counts_as_deleted(self, mock_git: Magi
         ]
         delete_branch("pr-split/pr-1", remote=True)
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_branch_gone_everywhere_counts_as_deleted(self, mock_git: MagicMock) -> None:
+    def test_branch_gone_everywhere_counts_as_deleted(
+        self, mock_git: MagicMock, mock_forget: MagicMock
+    ) -> None:
         mock_git.side_effect = [
             GitOperationError("error: branch 'pr-split/pr-1' not found."),
             GitOperationError(
@@ -149,8 +246,11 @@ def test_missing_local_branch_without_remote_is_fine(self, mock_git: MagicMock)
         mock_git.side_effect = GitOperationError("error: branch 'pr-split/pr-1' not found.")
         delete_branch("pr-split/pr-1")
 
+    @patch("pr_split.git_ops.branches.forget_remote_tracking_ref")
     @patch("pr_split.git_ops.branches.run_git")
-    def test_other_remote_failure_still_raises(self, mock_git: MagicMock) -> None:
+    def test_other_remote_failure_still_raises(
+        self, mock_git: MagicMock, mock_forget: MagicMock
+    ) -> None:
         mock_git.side_effect = ["", GitOperationError("fatal: could not read from remote")]
         with pytest.raises(GitOperationError, match="could not read from remote"):
             delete_branch("pr-split/pr-1", remote=True)
@@ -333,6 +433,162 @@ def test_localised_git_still_reports_already_deleted(
         delete_branch("pr-split/ns/never")
 
 
+class TestStaleRemoteTrackingRefs:
+    def _repo_with_origin(self, tmp_path: Path) -> Path:
+        origin = tmp_path / "origin.git"
+        subprocess.run(["git", "init", "-q", "--bare", str(origin)], check=True)
+        repo = tmp_path / "repo"
+        subprocess.run(["git", "clone", "-q", str(origin), str(repo)], check=True)
+        subprocess.run(
+            [
+                "git",
+                "-c",
+                "user.name=t",
+                "-c",
+                "user.email=t@x",
+                "commit",
+                "-q",
+                "--allow-empty",
+                "-m",
+                "base",
+            ],
+            cwd=repo,
+            check=True,
+        )
+        subprocess.run(["git", "push", "-q", "-u", "origin", "HEAD:main"], cwd=repo, check=True)
+        return repo
+
+    def test_reused_branch_name_pushes_again_after_remote_deletion(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from pr_split.git_ops.branches import prune_remote_tracking_refs, push_branch
+
+        repo = self._repo_with_origin(tmp_path)
+        monkeypatch.chdir(repo)
+        subprocess.run(["git", "branch", "pr-split/ns/pr-1"], cwd=repo, check=True)
+        push_branch("pr-split/ns/pr-1")
+        # Simulate `gh pr merge --delete-branch` / GitHub deleting the head
+        # branch behind our back: the remote branch goes away but the local
+        # tracking ref still points at the old head.
+        subprocess.run(
+            ["git", "update-ref", "-d", "refs/heads/pr-split/ns/pr-1"],
+            cwd=tmp_path / "origin.git",
+            check=True,
+        )
+        subprocess.run(
+            [
+                "git",
+                "-c",
+                "user.name=t",
+                "-c",
+                "user.email=t@x",
+                "commit",
+                "-q",
+                "--allow-empty",
+                "-m",
+                "v2",
+            ],
+            cwd=repo,
+            check=True,
+        )
+        subprocess.run(["git", "branch", "-f", "pr-split/ns/pr-1", "HEAD"], cwd=repo, check=True)
+        with pytest.raises(GitOperationError, match="stale info"):
+            push_branch("pr-split/ns/pr-1")
+
+        prune_remote_tracking_refs()
+        push_branch("pr-split/ns/pr-1")
+
+    def test_prune_keeps_the_lease_against_third_party_pushes(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from pr_split.git_ops.branches import prune_remote_tracking_refs, push_branch
+
+        repo = self._repo_with_origin(tmp_path)
+        monkeypatch.chdir(repo)
+        subprocess.run(["git", "branch", "pr-split/ns/pr-9"], cwd=repo, check=True)
+        push_branch("pr-split/ns/pr-9")
+        # A reviewer pushes a fixup to the split branch from another clone.
+        other = tmp_path / "other"
+        subprocess.run(
+            ["git", "clone", "-q", str(tmp_path / "origin.git"), str(other)], check=True
+        )
+        subprocess.run(["git", "checkout", "-q", "pr-split/ns/pr-9"], cwd=other, check=True)
+        subprocess.run(
+            [
+                "git",
+                "-c",
+                "user.name=r",
+                "-c",
+                "user.email=r@x",
+                "commit",
+                "-q",
+                "--allow-empty",
+                "-m",
+                "fixup",
+            ],
+            cwd=other,
+            check=True,
+        )
+        subprocess.run(["git", "push", "-q", "origin", "pr-split/ns/pr-9"], cwd=other, check=True)
+        subprocess.run(
+            [
+                "git",
+                "-c",
+                "user.name=t",
+                "-c",
+                "user.email=t@x",
+                "commit",
+                "-q",
+                "--allow-empty",
+                "-m",
+                "resplit",
+            ],
+            cwd=repo,
+            check=True,
+        )
+        subprocess.run(["git", "branch", "-f", "pr-split/ns/pr-9", "HEAD"], cwd=repo, check=True)
+
+        prune_remote_tracking_refs()
+
+        # The lease must still protect the reviewer's commit.
+        with pytest.raises(GitOperationError, match="stale info"):
+            push_branch("pr-split/ns/pr-9")
+
+    def test_remote_delete_drops_the_tracking_ref(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from pr_split.git_ops.branches import push_branch
+
+        repo = self._repo_with_origin(tmp_path)
+        monkeypatch.chdir(repo)
+        subprocess.run(["git", "branch", "pr-split/ns/pr-2"], cwd=repo, check=True)
+        push_branch("pr-split/ns/pr-2")
+        assert branch_exists("refs/remotes/origin/pr-split/ns/pr-2")
+
+        delete_branch("pr-split/ns/pr-2", remote=True)
+
+        assert not branch_exists("refs/remotes/origin/pr-split/ns/pr-2")
+
+    def test_already_deleted_remote_still_drops_the_tracking_ref(
+        self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+    ) -> None:
+        from pr_split.git_ops.branches import push_branch
+
+        repo = self._repo_with_origin(tmp_path)
+        monkeypatch.chdir(repo)
+        subprocess.run(["git", "branch", "pr-split/ns/pr-3"], cwd=repo, check=True)
+        push_branch("pr-split/ns/pr-3")
+        subprocess.run(
+            ["git", "update-ref", "-d", "refs/heads/pr-split/ns/pr-3"],
+            cwd=tmp_path / "origin.git",
+            check=True,
+        )
+
+        delete_branch("pr-split/ns/pr-3", remote=True)
+
+        assert not branch_exists("refs/remotes/origin/pr-split/ns/pr-3")
+
+
 class TestIsWorktreeCleanIgnoresPlanDir:
     """The check must not trip on pr-split's own `.pr-split/plan.json`."""