Fix #2316: skip synthetic-session checkpoints + serialize cross-source-repo pushes

gateway/checkpoint_handler.py

@@ -885,7 +885,7 @@ def store_checkpoint_v2(
 
         try:
             with (
-                _get_repo_lock(repo_path),
+                _get_store_lock(checkpoint_repo or repo_path),
                 tempfile.TemporaryDirectory(
                     prefix="checkpoint_", ignore_cleanup_errors=True
                 ) as temp_dir,
@@ -1024,10 +1024,19 @@ def store_checkpoint_v2(
                         ["commit", "--no-verify", "-m", commit_msg],
                     )
 
-                    # Retry push with pull-rebase on non-fast-forward rejection.
-                    # Multiple checkpoint stores can race: they fetch the same base,
-                    # commit locally, then the second push fails because the first
-                    # already advanced the remote branch.
+                    # Retry push with regenerate-on-non-fast-forward.
+                    # When two writers race on the shared checkpoint branch,
+                    # the second sees a non-FF rejection because the first
+                    # advanced the remote tip — typically with its own edits
+                    # to ``index.json``. Rebasing this writer's commit cannot
+                    # auto-merge textual diffs to ``index.json`` even though
+                    # the update is structurally an append. Instead we discard
+                    # our local commit, fetch the latest tip, reset the temp
+                    # worktree onto it, and replay this checkpoint's delta:
+                    # the unique-named checkpoint file is re-staged and
+                    # ``add_checkpoint_to_index_v2`` re-derives ``index.json``
+                    # from the freshly fetched index plus this checkpoint.
+                    # See #2316.
                     max_push_attempts = 3
                     for push_attempt in range(1, max_push_attempts + 1):
                         try:
@@ -1044,22 +1053,56 @@ def store_checkpoint_v2(
                             if push_attempt >= max_push_attempts:
                                 raise
                             logger.warning(
-                                "Checkpoint push rejected (non-fast-forward), rebasing and retrying",
+                                "Checkpoint push rejected (non-fast-forward), regenerating against fresh remote tip",
                                 attempt=push_attempt,
                                 max_attempts=max_push_attempts,
                                 checkpoint_id=checkpoint.id,
                             )
                             time.sleep(1)
-                            # Pull remote changes and rebase our commit on top
+                            # Detach the temp worktree from CHECKPOINT_BRANCH
+                            # so the next ``fetch +CHECKPOINT_BRANCH:CHECKPOINT_BRANCH``
+                            # can update that ref without git refusing because
+                            # the branch is checked out in this worktree. The
+                            # orphan path leaves the worktree on the branch
+                            # (``checkout --orphan`` switches to it), so this
+                            # step matters when a concurrent writer created
+                            # the branch on origin between ``_branch_exists``
+                            # and our push.
                             self._run_git(
                                 str(temp_path),
+                                ["checkout", "--detach"],
+                            )
+                            # Fetch the latest remote state into the local
+                            # branch, then reset the temp worktree to it —
+                            # discards our prior commit and pulls in any
+                            # concurrent writer's index.json updates.
+                            self._run_git(
+                                repo_path,
                                 ["fetch", target, f"+{CHECKPOINT_BRANCH}:{CHECKPOINT_BRANCH}"],
                                 timeout=60,
                                 github_token=github_token,
                             )
                             self._run_git(
                                 str(temp_path),
-                                ["rebase", CHECKPOINT_BRANCH],
+                                ["reset", "--hard", CHECKPOINT_BRANCH],
+                            )
+                            # Replay this checkpoint's delta against the
+                            # fresh index. Checkpoint path is unique-by-id
+                            # so it never conflicts; the index is rebuilt
+                            # from the new remote state plus this summary.
+                            save_checkpoint_v2(checkpoint, checkpoint_path)
+                            add_checkpoint_to_index_v2(checkpoint, index_path)
+                            self._run_git(
+                                str(temp_path),
+                                [
+                                    "add",
+                                    str(checkpoint_path.relative_to(temp_path)),
+                                    INDEX_FILE,
+                                ],
+                            )
+                            self._run_git(
+                                str(temp_path),
+                                ["commit", "--no-verify", "-m", commit_msg],
                             )
 
                     logger.info(
@@ -1342,40 +1385,51 @@ def _run_git(
 _checkpoint_handler: CheckpointHandler | None = None
 _handler_lock = threading.Lock()
 
-# Per-repo_path locks serializing concurrent checkpoint stores.
-# Concurrent threads operating on the same source repo race on
-# .git/worktrees state and ref locks under repo_path/.git/, producing
-# 'worktree add' failures and stale worktree directories. The lock is
-# keyed by repo_path only — the target (origin vs. external checkpoint
-# repo) does not affect the local .git state being contended.
-# Not trimmed: the set of distinct repo_paths in a gateway process is
-# small and stable in practice (one entry per source repo seen).
-# See #2069.
-_repo_locks: dict[str, threading.Lock] = {}
-_repo_locks_guard = threading.Lock()
+# Per-destination locks serializing concurrent checkpoint stores.
+# When ``checkpoint_repo`` is set, the lock key is that destination
+# ("owner/repo") so cross-source-repo writers contending on the same
+# shared ``egg/checkpoints/v2`` branch (e.g. multiple agents from
+# different repos all targeting ``jwbron/egg-checkpoints``) serialize
+# their fetch + commit + push sequence. When ``checkpoint_repo`` is
+# unset, fall back to the source ``repo_path`` so same-source-repo
+# writers still serialize on local ``.git/worktrees`` state.
+# See #2069 (per-source-repo origin) and #2316 (target-keyed).
+_store_locks: dict[str, threading.Lock] = {}
+_store_locks_guard = threading.Lock()
 
 
 @contextlib.contextmanager
-def _get_repo_lock(repo_path: str) -> Generator[None]:
-    """Hold the per-process per-repo lock for the whole checkpoint op.
-
-    In-process serialization only (#2069). Cross-process serialization
-    against the orchestrator's state-store on ``.git/config.lock``
-    (#2311) is handled with narrower ``bare_repo_lock`` windows around
-    the specific git operations that touch the bare repo's ``.git/``
-    (worktree add, worktree remove/prune) — see #2332.
-
-    Holding the cross-process flock for the entire op would block every
-    state-store commit and other gateway worktree op against the same
-    bare repo for up to ~135s under fetch-retry pathology, which is
-    much wider than necessary: fetch, the in-temp-worktree commit, and
-    the push do not race the state-store on ``.git/config.lock``.
+def _get_store_lock(key: str) -> Generator[None]:
+    """Hold the per-destination in-process lock for the whole checkpoint op.
+
+    Per-destination ``threading.Lock`` keyed on ``key`` for in-process
+    serialization.  When ``checkpoint_repo`` is set, callers pass it as
+    ``key`` so cross-source-repo writers contending on the same shared
+    ``egg/checkpoints/v2`` branch serialize their fetch + commit + push
+    sequence (#2316).  When unset, callers fall back to ``repo_path``
+    so same-source-repo writers still serialize on local
+    ``.git/worktrees`` state (#2069).  The destination key only
+    synchronizes within a *single gateway process* — multiple gateway
+    pods writing to the same ``checkpoint_repo`` race past this lock,
+    and the regenerate-on-non-FF retry in ``store_checkpoint_v2`` is
+    the actual cross-pod protection.
+
+    Cross-process serialization against the orchestrator's state-store
+    on ``.git/config.lock`` (#2311) is handled with narrower
+    ``bare_repo_lock`` windows around the specific git operations that
+    touch the bare repo's ``.git/`` (worktree add, worktree remove/
+    prune) — see #2332.  Holding the cross-process flock for the
+    entire op would block every state-store commit and other gateway
+    worktree op against the same bare repo for up to ~135s under
+    fetch-retry pathology, which is much wider than necessary: fetch,
+    the in-temp-worktree commit, and the push do not race the state-
+    store on ``.git/config.lock``.
     """
-    with _repo_locks_guard:
-        thread_lock = _repo_locks.get(repo_path)
+    with _store_locks_guard:
+        thread_lock = _store_locks.get(key)
         if thread_lock is None:
             thread_lock = threading.Lock()
-            _repo_locks[repo_path] = thread_lock
+            _store_locks[key] = thread_lock
     with thread_lock:
         yield
 

gateway/gateway.py

@@ -7981,6 +7981,7 @@ def session_create() -> tuple[Response, int] | Response:
     claude_code_version = data.get("claude_code_version")  # Optional Claude Code version
     branch = data.get("branch")  # Optional git branch for non-pushing sessions
     jira_ticket = data.get("jira_ticket")  # Optional Atlassian ticket key — advisory only
+    synthetic = data.get("synthetic", False)  # Orchestrator-internal temp session
 
     # Validate required fields
     if not container_id:
@@ -8057,6 +8058,10 @@ def session_create() -> tuple[Response, int] | Response:
         if len(branch) > 256:
             return make_error("Invalid branch: must be 256 characters or fewer")
 
+    # Validate synthetic if provided
+    if not isinstance(synthetic, bool):
+        return make_error("Invalid synthetic: must be a boolean")
+
     # Validate worktree_container_id if provided
     if worktree_container_id is not None:
         if not isinstance(worktree_container_id, str):
@@ -8260,6 +8265,7 @@ def session_create() -> tuple[Response, int] | Response:
         claude_code_version=claude_code_version,
         branch=branch,
         jira_ticket=jira_ticket if isinstance(jira_ticket, str) and jira_ticket else None,
+        synthetic=synthetic,
     )
 
     # Pre-populate checkpoint context so non-pushing sessions (reviewers,

gateway/session_manager.py

@@ -103,6 +103,13 @@ def _capture_and_cleanup_session(
         was skipped/failed. Callers can wait on this event to coordinate
         cleanup (e.g., worktree removal) with checkpoint storage.
     """
+    # Synthetic sessions are orchestrator-internal helpers (ls-remote,
+    # failsafe-fetch). They run no agent, have no proxy buffer, and would
+    # only produce metadata-only checkpoints whose push fails noisily when
+    # the source repo is read-only (#2316). Skip them entirely.
+    if session.synthetic:
+        return None
+
     # Deduplicate: only capture once per container
     with _captured_containers_lock:
         if session.container_id in _captured_containers:
@@ -317,6 +324,7 @@ class Session:
     assigned_branch: str | None = None  # Worktree branch locked to this session
     auto_commit_sha: str | None = None  # SHA from post-agent auto-commit
     jira_ticket: str | None = None  # Advisory Jira ticket key (issue #1556)
+    synthetic: bool = False  # Orchestrator-internal temp session — skip checkpoint capture
 
     def is_expired(self) -> bool:
         """Check if session has expired."""
@@ -364,6 +372,8 @@ def to_dict_for_persistence(self) -> dict[str, Any]:
             result["auto_commit_sha"] = self.auto_commit_sha
         if self.jira_ticket is not None:
             result["jira_ticket"] = self.jira_ticket
+        if self.synthetic:
+            result["synthetic"] = True
         return result
 
     @classmethod
@@ -391,6 +401,7 @@ def from_persistence(cls, data: dict[str, Any]) -> Session:
             assigned_branch=data.get("assigned_branch"),
             auto_commit_sha=data.get("auto_commit_sha"),
             jira_ticket=data.get("jira_ticket"),
+            synthetic=bool(data.get("synthetic", False)),
         )
 
 
@@ -548,6 +559,7 @@ def register_session(
         claude_code_version: str | None = None,
         branch: str | None = None,
         jira_ticket: str | None = None,
+        synthetic: bool = False,
     ) -> tuple[str, Session]:
         """
         Register a new session for a container.
@@ -590,6 +602,7 @@ def register_session(
             agent_anchor_id=agent_anchor_id,
             claude_code_version=claude_code_version,
             jira_ticket=jira_ticket,
+            synthetic=synthetic,
         )
 
         if branch: