From 417bdc2b66fe1e94736e7d7fd3a16254f4bc7675 Mon Sep 17 00:00:00 2001
From: James Wiesebron <jameswiesebron@khanacademy.org>
Date: Wed, 29 Apr 2026 22:04:19 -0700
Subject: [PATCH 1/5] Fix #2316: skip synthetic-session checkpoints + serialize
 cross-source-repo pushes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Two checkpoint storage failures were firing on every pipeline run.

Failure 1: orchestrator-internal helpers (ls-remote, failsafe-fetch)
register synthetic sessions on the gateway. On session deletion, the
gateway tries to capture a session-end checkpoint for them — but they
have no proxy buffer (no agent ran), so they only produce metadata-only
checkpoints. When the source repo is read-only (e.g. Khan/actions), the
push falls back to the source origin and fails with permission denied,
logging at ERROR.

Fix: add a `synthetic` flag to Session, plumb it through
register_session and the four temp-session helpers in gateway_client.py,
and short-circuit `_capture_and_cleanup_session` for synthetic sessions.
No data is lost — these checkpoints only carried synthetic container_id
metadata.

Failure 2: when two writers from different source repos targeted the
same shared `egg/checkpoints/v2` branch in jwbron/egg-checkpoints, the
per-source-repo lock didn't serialize them. The second writer hit a
non-FF rejection, attempted rebase, and failed with a content conflict
on `index.json` (which is structurally an append but textually a diff).
After 3 attempts the checkpoint was dropped — real session-end
transcripts and tool calls lost.

Fix:
- Re-key the store lock by `checkpoint_repo or repo_path` so writers
  contending on the shared destination serialize. Renamed the dict and
  helper to reflect the broader scope (`_store_locks`, `_get_store_lock`).
- Replace rebase-on-non-FF with regenerate-on-non-FF: discard the local
  commit, fetch the latest tip, reset the temp worktree, re-apply this
  checkpoint's delta against the freshly fetched index, recommit, push.
  This treats `index.json` as a structural append rather than a textual
  diff, so concurrent writers never conflict.
---
 gateway/checkpoint_handler.py            |  87 +++++++++++++------
 gateway/gateway.py                       |   2 +
 gateway/session_manager.py               |  13 +++
 gateway/tests/test_checkpoint_handler.py | 103 ++++++++++++++++++-----
 gateway/tests/test_session_manager.py    |  42 +++++++++
 orchestrator/gateway_client.py           |   7 ++
 6 files changed, 207 insertions(+), 47 deletions(-)

diff --git a/gateway/checkpoint_handler.py b/gateway/checkpoint_handler.py
index 289190c9cf..0dfb62d14e 100644
--- a/gateway/checkpoint_handler.py
+++ b/gateway/checkpoint_handler.py
@@ -879,11 +879,14 @@ def store_checkpoint_v2(
             return False
 
         target = _resolve_checkpoint_target(checkpoint_repo, remote, repo_path)
-        repo_lock = _get_repo_lock(repo_path)
+        # Lock key: prefer checkpoint_repo so cross-source-repo writers
+        # targeting the same shared checkpoint branch serialize (#2316).
+        # Fall back to repo_path for the same-source-repo case (#2069).
+        store_lock = _get_store_lock(checkpoint_repo or repo_path)
 
         try:
             with (
-                repo_lock,
+                store_lock,
                 tempfile.TemporaryDirectory(
                     prefix="checkpoint_", ignore_cleanup_errors=True
                 ) as temp_dir,
@@ -1010,10 +1013,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:
@@ -1030,22 +1042,43 @@ 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
+                            # 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(
-                                str(temp_path),
+                                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(
@@ -1324,25 +1357,25 @@ 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()
-
-
-def _get_repo_lock(repo_path: str) -> threading.Lock:
-    with _repo_locks_guard:
-        lock = _repo_locks.get(repo_path)
+# 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()
+
+
+def _get_store_lock(key: str) -> threading.Lock:
+    with _store_locks_guard:
+        lock = _store_locks.get(key)
         if lock is None:
             lock = threading.Lock()
-            _repo_locks[repo_path] = lock
+            _store_locks[key] = lock
         return lock
 
 
diff --git a/gateway/gateway.py b/gateway/gateway.py
index 3845465780..8418bd081d 100644
--- a/gateway/gateway.py
+++ b/gateway/gateway.py
@@ -8007,6 +8007,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 = bool(data.get("synthetic", False))  # Orchestrator-internal temp session
 
     # Validate required fields
     if not container_id:
@@ -8286,6 +8287,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,
diff --git a/gateway/session_manager.py b/gateway/session_manager.py
index fe22b70ab7..7a66776f56 100644
--- a/gateway/session_manager.py
+++ b/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:
diff --git a/gateway/tests/test_checkpoint_handler.py b/gateway/tests/test_checkpoint_handler.py
index fb5dcbed31..0c8f7dc695 100644
--- a/gateway/tests/test_checkpoint_handler.py
+++ b/gateway/tests/test_checkpoint_handler.py
@@ -668,7 +668,7 @@ def test_worktree_prune_called_in_cleanup(self):
         import checkpoint_handler
 
         # Reset the per-repo lock dict so this test is isolated.
-        checkpoint_handler._repo_locks.clear()
+        checkpoint_handler._store_locks.clear()
 
         handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
 
@@ -695,7 +695,7 @@ def test_worktree_prune_runs_when_worktree_add_fails(self):
         """If `worktree add` itself raises, cleanup still runs `worktree prune`."""
         import checkpoint_handler
 
-        checkpoint_handler._repo_locks.clear()
+        checkpoint_handler._store_locks.clear()
 
         handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
 
@@ -729,7 +729,7 @@ def test_concurrent_stores_on_same_repo_serialized(self):
         import checkpoint_handler
 
         # Reset the per-repo lock dict so this test is isolated.
-        checkpoint_handler._repo_locks.clear()
+        checkpoint_handler._store_locks.clear()
 
         handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
 
@@ -777,7 +777,7 @@ def test_concurrent_stores_on_different_repos_not_serialized(self):
 
         import checkpoint_handler
 
-        checkpoint_handler._repo_locks.clear()
+        checkpoint_handler._store_locks.clear()
 
         handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
 
@@ -823,6 +823,62 @@ def run_store(repo_path):
             f"saw max_in_flight={max_in_flight}"
         )
 
+    def test_concurrent_stores_with_shared_checkpoint_repo_serialized(self):
+        """Different source repos pushing to the same checkpoint_repo serialize.
+
+        Regression test for #2316: before the target-keyed lock, two source
+        repos targeting the shared ``egg/checkpoints/v2`` branch in
+        ``jwbron/egg-checkpoints`` could push concurrently, and the second
+        writer would see a non-FF rejection.
+        """
+        import threading
+        import time
+
+        import checkpoint_handler
+
+        checkpoint_handler._store_locks.clear()
+
+        handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
+
+        in_flight = 0
+        max_in_flight = 0
+        observe_lock = threading.Lock()
+
+        def track_run_git(cwd, args, **kwargs):
+            nonlocal in_flight, max_in_flight
+            with observe_lock:
+                in_flight += 1
+                max_in_flight = max(max_in_flight, in_flight)
+            time.sleep(0.05)
+            with observe_lock:
+                in_flight -= 1
+            return MagicMock(returncode=0, stdout="", stderr="")
+
+        handler._run_git = track_run_git
+        handler._branch_exists = MagicMock(return_value=True)
+
+        def run_store(repo_path):
+            try:
+                handler.store_checkpoint_v2(
+                    self._make_checkpoint(),
+                    repo_path,
+                    checkpoint_repo="jwbron/egg-checkpoints",
+                )
+            except Exception:
+                pass
+
+        t1 = threading.Thread(target=run_store, args=("/fake/repo-a",))
+        t2 = threading.Thread(target=run_store, args=("/fake/repo-b",))
+        t1.start()
+        t2.start()
+        t1.join()
+        t2.join()
+
+        assert max_in_flight == 1, (
+            "Expected serialization across source repos when checkpoint_repo "
+            f"is shared, saw max_in_flight={max_in_flight}"
+        )
+
 
 class TestStoreCheckpointV2RemoteTarget:
     """Tests for store_checkpoint_v2 remote URL resolution (issue #1767).
@@ -1797,7 +1853,7 @@ def track_run_git(cwd, args, **kwargs):
 
     @patch("time.sleep")
     def test_push_retries_on_non_fast_forward(self, mock_sleep):
-        """Push fails with non-fast-forward, fetch+rebase succeeds, second push succeeds."""
+        """Push fails with non-fast-forward, regenerate against fresh tip succeeds."""
         import checkpoint_handler
 
         handler, checkpoint = self._make_handler_and_checkpoint()
@@ -1822,15 +1878,19 @@ def track_run_git(cwd, args, **kwargs):
 
         push_calls = [c for c in git_calls if "push" in c[1]]
         assert len(push_calls) == 2, f"Expected 2 push attempts, got {len(push_calls)}"
-        # Verify fetch+rebase happened between pushes
-        fetch_after_push = [
-            c
-            for c in git_calls
-            if "fetch" in c[1] and git_calls.index(c) > git_calls.index(push_calls[0])
-        ]
-        rebase_calls = [c for c in git_calls if "rebase" in c[1]]
+        # Verify regenerate flow ran between pushes: fetch + reset --hard + re-add + re-commit
+        first_push_idx = git_calls.index(push_calls[0])
+        post_push = git_calls[first_push_idx + 1 :]
+        fetch_after_push = [c for c in post_push if "fetch" in c[1]]
+        reset_after_push = [c for c in post_push if "reset" in c[1]]
+        commit_after_push = [c for c in post_push if "commit" in c[1]]
         assert len(fetch_after_push) >= 1, "Expected fetch after failed push"
-        assert len(rebase_calls) >= 1, "Expected rebase after failed push"
+        assert len(reset_after_push) >= 1, "Expected reset --hard after failed push"
+        assert len(commit_after_push) >= 1, "Expected re-commit after regenerate"
+        # No rebase under the new strategy
+        assert not [c for c in git_calls if "rebase" in c[1]], (
+            "rebase should not be invoked under the regenerate strategy"
+        )
 
     @patch("time.sleep")
     def test_push_raises_after_max_attempts(self, mock_sleep):
@@ -1918,37 +1978,40 @@ def track_run_git(cwd, args, **kwargs):
         )
 
     @patch("time.sleep")
-    def test_push_fails_when_rebase_in_retry_fails(self, mock_sleep):
-        """Rebase within the retry loop fails — returns False."""
+    def test_push_fails_when_regenerate_commit_fails(self, mock_sleep):
+        """Regenerate-step commit failure during retry surfaces as False."""
         import checkpoint_handler
 
         handler, checkpoint = self._make_handler_and_checkpoint()
         git_calls = []
         push_count = 0
+        push_failed = False
 
         def track_run_git(cwd, args, **kwargs):
-            nonlocal push_count
+            nonlocal push_count, push_failed
             git_calls.append((cwd, args, kwargs))
             if "push" in args:
                 push_count += 1
                 if push_count == 1:
+                    push_failed = True
                     raise checkpoint_handler.CheckpointError(
                         "Git command failed: ! [rejected] non-fast-forward"
                     )
-            if "rebase" in args:
+            # Fail the re-commit during the regenerate step.
+            if push_failed and args[:1] == ["commit"]:
                 raise checkpoint_handler.CheckpointError(
-                    "Git command failed: CONFLICT (content): Merge conflict"
+                    "Git command failed: nothing to commit, working tree clean"
                 )
             return MagicMock(returncode=0, stdout="", stderr="")
 
         handler._run_git = track_run_git
 
         result = handler.store_checkpoint_v2(checkpoint, "/fake/repo")
-        assert result is False, "Expected store to return False on rebase failure"
+        assert result is False, "Expected store to return False on regenerate-commit failure"
 
         push_calls = [c for c in git_calls if "push" in c[1]]
         assert len(push_calls) == 1, (
-            f"Expected 1 push attempt before rebase failure, got {len(push_calls)}"
+            f"Expected 1 push attempt before regenerate failure, got {len(push_calls)}"
         )
 
 
diff --git a/gateway/tests/test_session_manager.py b/gateway/tests/test_session_manager.py
index a546ea1114..9e8031f140 100644
--- a/gateway/tests/test_session_manager.py
+++ b/gateway/tests/test_session_manager.py
@@ -187,6 +187,17 @@ def test_register_session(self, manager):
         assert len(token) > 32  # Should be a substantial token
         assert session.container_id == "test-container"
         assert session.mode == "private"
+        assert session.synthetic is False
+
+    def test_register_synthetic_session(self, manager):
+        """Synthetic flag plumbs through to the Session."""
+        _token, session = manager.register_session(
+            container_id="pipeline-1-state-ls-remote",
+            container_ip=None,
+            mode="public",
+            synthetic=True,
+        )
+        assert session.synthetic is True
 
     def test_validate_valid_session(self, manager):
         """Test validating a valid session."""
@@ -1697,6 +1708,37 @@ def test_prune_captures_expired_checkpoints(self, manager):
             for c in mock_capture.call_args_list:
                 assert c[0][1] == "expired"
 
+    def test_synthetic_session_skips_checkpoint_capture(self):
+        """Synthetic temp sessions skip the session-end checkpoint path entirely.
+
+        Regression test for #2316: orchestrator-internal helpers (ls-remote,
+        failsafe-fetch) register synthetic sessions that have no proxy buffer
+        and would only produce metadata-only checkpoints whose push fails
+        noisily on read-only source repos.
+        """
+        from unittest.mock import patch
+
+        from session_manager import _capture_and_cleanup_session
+
+        now = datetime.now(UTC)
+        session = Session(
+            session_token="t",
+            session_token_hash=_hash_token("t"),
+            container_id="pipeline-1-state-ls-remote",
+            container_ip=None,
+            mode="public",
+            created_at=now,
+            last_seen=now,
+            expires_at=now + timedelta(hours=1),
+            synthetic=True,
+        )
+
+        with patch.object(session_manager_module, "_cleanup_transcript_buffer") as cleanup:
+            event = _capture_and_cleanup_session(session, "completed")
+
+        assert event is None
+        cleanup.assert_not_called()
+
     def test_delete_by_token_not_found_skips_capture(self, manager):
         """delete_session with invalid token doesn't capture checkpoint."""
         from unittest.mock import patch
diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py
index 4d7ec1ac40..c2ac025977 100644
--- a/orchestrator/gateway_client.py
+++ b/orchestrator/gateway_client.py
@@ -356,6 +356,7 @@ def register_session(
         branch: str | None = None,
         worktree_container_id: str | None = None,
         jira_ticket: str | None = None,
+        synthetic: bool = False,
     ) -> SessionInfo:
         """Register a session for a container.
 
@@ -424,6 +425,8 @@ def register_session(
             # any Jira call on its value — the project allowlist is the only
             # hard boundary.
             request_data["jira_ticket"] = jira_ticket
+        if synthetic:
+            request_data["synthetic"] = True
         result = self._make_request(
             "/api/v1/sessions/create",
             method="POST",
@@ -1735,6 +1738,7 @@ def fetch_worktree_branch(
                 container_ip=self.self_ip,
                 mode=mode,
                 pipeline_id=pipeline_id,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1797,6 +1801,7 @@ def fetch_branch(
                 container_ip=self.self_ip,
                 mode=mode,
                 pipeline_id=pipeline_id,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1860,6 +1865,7 @@ def ls_remote_branch(
                 container_ip=self.self_ip,
                 mode=mode,
                 pipeline_id=pipeline_id,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1921,6 +1927,7 @@ def get_remote_branch_sha(
                 container_ip=self.self_ip,
                 mode=mode,
                 pipeline_id=pipeline_id,
+                synthetic=True,
             )
             session_token = session.session_token
 

From 0214e3acfffd471eb1fb0b883fe2680d790c3cef Mon Sep 17 00:00:00 2001
From: "egg-reviewer[bot]"
 <261018737+egg-reviewer[bot]@users.noreply.github.com>
Date: Thu, 30 Apr 2026 06:28:05 +0000
Subject: [PATCH 2/5] Address review feedback on #2316 fix
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

- Mark 5 additional orchestrator-internal temp-session helpers as
  synthetic=True (auto-pr, rebase_onto, slice-branch, stacked-pr-list,
  stacked-pr-ls-remote) so their session-end checkpoints skip the same
  noisy push path the four read-only helpers already skip.
- Validate ``synthetic`` is a bool at /api/v1/sessions/create (matches
  the surrounding pipeline_id / agent_role validation style; replaces
  the silent ``bool(data.get(...))`` coercion).
- Document in ``_get_store_lock`` that the destination key only
  serializes within a single gateway process — cross-pod writers race
  past it and the regenerate-on-non-FF retry is the actual cross-pod
  protection.
- Add ``threading.Barrier`` to the new shared-checkpoint_repo
  serialization test, mirroring its sibling, so a regression that drops
  the destination-keyed lock is caught even if the OS happens to
  schedule the threads sequentially.
- Detach the temp worktree from CHECKPOINT_BRANCH at the top of the
  regenerate retry block. The orphan path leaves the worktree on the
  branch (``checkout --orphan`` switches to it), so a concurrent writer
  creating the branch on origin between ``_branch_exists`` and our push
  would otherwise cause the regenerate fetch to fail (git refuses to
  fetch into a ref checked out in any worktree).
---
 gateway/checkpoint_handler.py            | 21 +++++++++++++++++-
 gateway/gateway.py                       |  6 ++++-
 gateway/tests/test_checkpoint_handler.py | 20 ++++++++++++++++-
 gateway/tests/test_gateway.py            | 28 ++++++++++++++++++++++++
 orchestrator/gateway_client.py           |  5 +++++
 5 files changed, 77 insertions(+), 3 deletions(-)

diff --git a/gateway/checkpoint_handler.py b/gateway/checkpoint_handler.py
index d4ca6e4f69..2cf58e8c72 100644
--- a/gateway/checkpoint_handler.py
+++ b/gateway/checkpoint_handler.py
@@ -1047,6 +1047,19 @@ def store_checkpoint_v2(
                                 checkpoint_id=checkpoint.id,
                             )
                             time.sleep(1)
+                            # 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
@@ -1381,7 +1394,13 @@ def _get_store_lock(key: str, repo_path: str) -> Generator[None]:
       ``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).
+      ``.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 (the ``bare_repo_lock`` flock
+      below is keyed by ``repo_path``, not by destination, so it does
+      not serialize cross-pod writers on the shared destination either).
     * ``bare_repo_lock(repo_path)`` for cross-process serialization
       against the orchestrator's state-store, which runs git from a
       different pod but shares the same hostPath-mounted bare repo
diff --git a/gateway/gateway.py b/gateway/gateway.py
index 8418bd081d..af7ff44971 100644
--- a/gateway/gateway.py
+++ b/gateway/gateway.py
@@ -8007,7 +8007,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 = bool(data.get("synthetic", False))  # Orchestrator-internal temp session
+    synthetic = data.get("synthetic", False)  # Orchestrator-internal temp session
 
     # Validate required fields
     if not container_id:
@@ -8084,6 +8084,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):
diff --git a/gateway/tests/test_checkpoint_handler.py b/gateway/tests/test_checkpoint_handler.py
index ce4f0fbd7a..3f83dfc566 100644
--- a/gateway/tests/test_checkpoint_handler.py
+++ b/gateway/tests/test_checkpoint_handler.py
@@ -875,12 +875,24 @@ def test_concurrent_stores_with_shared_checkpoint_repo_serialized(self):
         in_flight = 0
         max_in_flight = 0
         observe_lock = threading.Lock()
+        # Force overlap if the lock disappears: a regression that drops
+        # the destination-keyed lock would let both threads into
+        # ``_run_git`` concurrently, the barrier would release them
+        # together, and ``max_in_flight`` would jump to 2. With the
+        # lock in place only one thread reaches the barrier; the wait
+        # times out, ``BrokenBarrierError`` is caught, and serialization
+        # is still observed.
+        barrier = threading.Barrier(2)
 
         def track_run_git(cwd, args, **kwargs):
             nonlocal in_flight, max_in_flight
             with observe_lock:
                 in_flight += 1
                 max_in_flight = max(max_in_flight, in_flight)
+            try:
+                barrier.wait(timeout=2.0)
+            except threading.BrokenBarrierError:
+                pass
             time.sleep(0.05)
             with observe_lock:
                 in_flight -= 1
@@ -1910,12 +1922,18 @@ def track_run_git(cwd, args, **kwargs):
 
         push_calls = [c for c in git_calls if "push" in c[1]]
         assert len(push_calls) == 2, f"Expected 2 push attempts, got {len(push_calls)}"
-        # Verify regenerate flow ran between pushes: fetch + reset --hard + re-add + re-commit
+        # Verify regenerate flow ran between pushes:
+        # checkout --detach + fetch + reset --hard + re-add + re-commit
         first_push_idx = git_calls.index(push_calls[0])
         post_push = git_calls[first_push_idx + 1 :]
+        detach_after_push = [c for c in post_push if "checkout" in c[1] and "--detach" in c[1]]
         fetch_after_push = [c for c in post_push if "fetch" in c[1]]
         reset_after_push = [c for c in post_push if "reset" in c[1]]
         commit_after_push = [c for c in post_push if "commit" in c[1]]
+        assert len(detach_after_push) >= 1, (
+            "Expected checkout --detach before fetch so the local "
+            "CHECKPOINT_BRANCH ref is updatable from the orphan path"
+        )
         assert len(fetch_after_push) >= 1, "Expected fetch after failed push"
         assert len(reset_after_push) >= 1, "Expected reset --hard after failed push"
         assert len(commit_after_push) >= 1, "Expected re-commit after regenerate"
diff --git a/gateway/tests/test_gateway.py b/gateway/tests/test_gateway.py
index 7d5969d866..0347a2544e 100644
--- a/gateway/tests/test_gateway.py
+++ b/gateway/tests/test_gateway.py
@@ -4476,6 +4476,34 @@ def test_session_create_without_repos_rejected(self, client, launcher_auth_heade
             data = json.loads(response.data)
             assert "repos" in data["message"].lower()
 
+    def test_session_create_rejects_non_bool_synthetic(
+        self, client, launcher_auth_headers
+    ):
+        """``synthetic`` must be a boolean — non-bool truthy values are rejected.
+
+        Matches the validation style of the surrounding optional fields
+        (``pipeline_id``, ``agent_role``) so callers get a clear 400
+        instead of a silent ``bool("anything-truthy")`` coercion.
+        """
+        response = client.post(
+            "/api/v1/sessions/create",
+            headers=launcher_auth_headers,
+            data=json.dumps(
+                {
+                    "container_id": "test-container",
+                    "container_ip": "172.18.0.5",
+                    "mode": "private",
+                    "repos": ["owner/repo"],
+                    "synthetic": "yes",
+                }
+            ),
+            content_type="application/json",
+        )
+
+        assert response.status_code == 400
+        data = json.loads(response.data)
+        assert "synthetic" in data["message"].lower()
+
 
 class TestSessionCreateRepoVisibilityFiltering:
     """Tests for session creation repo filtering based on visibility and mode."""
diff --git a/orchestrator/gateway_client.py b/orchestrator/gateway_client.py
index c2ac025977..266850bb2f 100644
--- a/orchestrator/gateway_client.py
+++ b/orchestrator/gateway_client.py
@@ -1155,6 +1155,7 @@ def create_pr(
                 repos=[repo],
                 issue_number=issue_number,
                 agent_role=agent_role,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1378,6 +1379,7 @@ def rebase_onto(
                 pipeline_id=pipeline_id,
                 agent_role=agent_role,
                 branch=branch if retarget_requested else None,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1501,6 +1503,7 @@ def create_slice_integration_branch(
                 pipeline_id=pipeline_id,
                 agent_role=agent_role,
                 branch=integration_branch,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1578,6 +1581,7 @@ def list_open_prs(
                 mode=mode,
                 pipeline_id=pipeline_id,
                 agent_role=agent_role,
+                synthetic=True,
             )
             session_token = session.session_token
 
@@ -1673,6 +1677,7 @@ def list_remote_branches(
                 mode=mode,
                 pipeline_id=pipeline_id,
                 agent_role=agent_role,
+                synthetic=True,
             )
             session_token = session.session_token
 

From 662658ddeb6306e806ce2d621f30e15346ca0a88 Mon Sep 17 00:00:00 2001
From: egg <egg@localhost>
Date: Thu, 30 Apr 2026 06:29:40 +0000
Subject: [PATCH 3/5] Fix checks: apply automated formatting fixes

---
 gateway/tests/test_gateway.py | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/gateway/tests/test_gateway.py b/gateway/tests/test_gateway.py
index 0347a2544e..28673770b9 100644
--- a/gateway/tests/test_gateway.py
+++ b/gateway/tests/test_gateway.py
@@ -4476,9 +4476,7 @@ def test_session_create_without_repos_rejected(self, client, launcher_auth_heade
             data = json.loads(response.data)
             assert "repos" in data["message"].lower()
 
-    def test_session_create_rejects_non_bool_synthetic(
-        self, client, launcher_auth_headers
-    ):
+    def test_session_create_rejects_non_bool_synthetic(self, client, launcher_auth_headers):
         """``synthetic`` must be a boolean — non-bool truthy values are rejected.
 
         Matches the validation style of the surrounding optional fields

From 3e7e8ee71747da1c4f97dfd69b148ad6ee2977e4 Mon Sep 17 00:00:00 2001
From: "egg-reviewer[bot]"
 <261018737+egg-reviewer[bot]@users.noreply.github.com>
Date: Thu, 30 Apr 2026 07:30:50 +0000
Subject: [PATCH 4/5] Add flock-depth coverage for regenerate fetch +
 orphan-path branch -D

Two new TestStoreCheckpointV2Concurrency tests close coverage gaps
called out as non-blocking review feedback on PR #2328:

- test_bare_repo_lock_not_held_across_regenerate_fetch drives a non-FF
  push rejection so the regenerate-path fetch runs, and asserts every
  fetch (initial + regenerate) executes at flock depth 0. The existing
  test_bare_repo_lock_not_held_across_fetch_retry only exercised the
  initial fetch; a regression that wrapped the regenerate fetch in
  bare_repo_lock would have gone undetected.

- test_bare_repo_lock_wraps_branch_d_in_orphan_path mocks
  _branch_exists=False to drive the orphan path. Asserts `branch -D`
  and `worktree add` run under the flock (depth >=1), and
  `checkout --orphan` runs outside it (depth 0). The existing test
  mocked _branch_exists=True so the orphan path's `branch -D` was
  never executed.
---
 gateway/tests/test_checkpoint_handler.py | 135 +++++++++++++++++++++++
 1 file changed, 135 insertions(+)

diff --git a/gateway/tests/test_checkpoint_handler.py b/gateway/tests/test_checkpoint_handler.py
index 142814e501..8a19037260 100644
--- a/gateway/tests/test_checkpoint_handler.py
+++ b/gateway/tests/test_checkpoint_handler.py
@@ -998,6 +998,141 @@ def track_run_git(cwd, args, **kwargs):
             f"Worktree ops must run under bare_repo_lock, saw {worktree_observations}"
         )
 
+    def test_bare_repo_lock_not_held_across_regenerate_fetch(self, monkeypatch):
+        """Cross-process flock is also released around the regenerate-path fetch.
+
+        Companion to ``test_bare_repo_lock_not_held_across_fetch_retry``: the
+        non-FF push retry runs a second ``fetch +CHECKPOINT_BRANCH:CHECKPOINT_BRANCH``
+        from a separate code path (``checkpoint_handler.py:1079-1084``). It has
+        the same fetch-timeout pathology — holding the flock across it would
+        block every state-store commit and worktree op against the same bare
+        repo for up to ~60s. A regression that wrapped the regenerate fetch
+        in ``bare_repo_lock`` would not be caught by the initial-fetch test.
+        """
+        import checkpoint_handler
+
+        checkpoint_handler._store_locks.clear()
+
+        flock_depth = [0]
+        observations: list[tuple[list[str], int]] = []
+
+        @contextlib.contextmanager
+        def recording_flock(_repo_path):
+            flock_depth[0] += 1
+            try:
+                yield
+            finally:
+                flock_depth[0] -= 1
+
+        monkeypatch.setattr(checkpoint_handler, "bare_repo_lock", recording_flock)
+        monkeypatch.setattr(checkpoint_handler.time, "sleep", lambda _s: None)
+
+        handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
+
+        push_count = [0]
+
+        def track_run_git(cwd, args, **kwargs):
+            observations.append((list(args), flock_depth[0]))
+            if "push" in args:
+                push_count[0] += 1
+                if push_count[0] == 1:
+                    raise checkpoint_handler.CheckpointError(
+                        "Git command failed: ! [rejected] non-fast-forward"
+                    )
+            return MagicMock(returncode=0, stdout="", stderr="")
+
+        handler._run_git = track_run_git
+        handler._branch_exists = MagicMock(return_value=True)
+
+        result = handler.store_checkpoint_v2(self._make_checkpoint(), "/fake/repo")
+        assert result is True
+
+        push_calls = [args for args, _ in observations if "push" in args]
+        assert len(push_calls) == 2, (
+            f"Expected 2 push attempts (initial + regenerate retry), got {len(push_calls)}"
+        )
+
+        # Both the initial fetch and the regenerate-path fetch must run
+        # outside the flock. Assert there are at least 2 fetches and all
+        # of them are at depth 0.
+        fetch_observations = [depth for args, depth in observations if args[:1] == ["fetch"]]
+        assert len(fetch_observations) >= 2, (
+            f"Expected initial + regenerate fetches, observed {len(fetch_observations)}"
+        )
+        assert all(d == 0 for d in fetch_observations), (
+            "bare_repo_lock must not be held during any fetch (including the "
+            f"regenerate-path retry), saw depths {fetch_observations}"
+        )
+
+    def test_bare_repo_lock_wraps_branch_d_in_orphan_path(self, monkeypatch):
+        """``branch -D`` and the orphan-path ``worktree add`` run under the flock.
+
+        Companion to ``test_bare_repo_lock_not_held_across_fetch_retry``: that
+        test mocks ``_branch_exists=True`` so the orphan path's ``branch -D``
+        is never executed. A regression that moved ``branch -D`` outside the
+        flock window (``checkpoint_handler.py:978-987``) — or moved
+        ``checkout --orphan`` *inside* it — would not be caught. This test
+        drives the orphan path explicitly.
+        """
+        import checkpoint_handler
+
+        checkpoint_handler._store_locks.clear()
+
+        flock_depth = [0]
+        observations: list[tuple[list[str], int]] = []
+
+        @contextlib.contextmanager
+        def recording_flock(_repo_path):
+            flock_depth[0] += 1
+            try:
+                yield
+            finally:
+                flock_depth[0] -= 1
+
+        monkeypatch.setattr(checkpoint_handler, "bare_repo_lock", recording_flock)
+        monkeypatch.setattr(checkpoint_handler.time, "sleep", lambda _s: None)
+
+        handler = checkpoint_handler.CheckpointHandler(github_token="test-token")
+
+        def track_run_git(cwd, args, **kwargs):
+            observations.append((list(args), flock_depth[0]))
+            return MagicMock(returncode=0, stdout="", stderr="")
+
+        handler._run_git = track_run_git
+        handler._branch_exists = MagicMock(return_value=False)
+
+        result = handler.store_checkpoint_v2(self._make_checkpoint(), "/fake/repo")
+        assert result is True
+
+        # Both bare-repo writers in the orphan path must run under the flock.
+        branch_d_observations = [
+            depth for args, depth in observations if args[:2] == ["branch", "-D"]
+        ]
+        assert branch_d_observations, "Expected `branch -D` to run in the orphan path"
+        assert all(d >= 1 for d in branch_d_observations), (
+            f"`branch -D` must run under bare_repo_lock, saw depths {branch_d_observations}"
+        )
+
+        worktree_add_observations = [
+            depth for args, depth in observations if args[:2] == ["worktree", "add"]
+        ]
+        assert worktree_add_observations, "Expected `worktree add` to run in the orphan path"
+        assert all(d >= 1 for d in worktree_add_observations), (
+            f"`worktree add` must run under bare_repo_lock, saw depths {worktree_add_observations}"
+        )
+
+        # ``checkout --orphan`` runs inside the temp worktree, not the bare
+        # repo, so it must run *outside* the flock. A regression that nested
+        # it inside the flock window would inflate the cross-process critical
+        # section unnecessarily.
+        orphan_observations = [
+            depth for args, depth in observations if args[:2] == ["checkout", "--orphan"]
+        ]
+        assert orphan_observations, "Expected `checkout --orphan` to run in the orphan path"
+        assert all(d == 0 for d in orphan_observations), (
+            f"`checkout --orphan` must run outside bare_repo_lock, saw depths {orphan_observations}"
+        )
+
 
 class TestStoreCheckpointV2RemoteTarget:
     """Tests for store_checkpoint_v2 remote URL resolution (issue #1767).

From 42dc5f85cfe75bcd50701770c9645c863d392175 Mon Sep 17 00:00:00 2001
From: "egg-reviewer[bot]"
 <261018737+egg-reviewer[bot]@users.noreply.github.com>
Date: Thu, 30 Apr 2026 07:53:39 +0000
Subject: [PATCH 5/5] Tighten regenerate-fetch count and assert branch -D
 ordering in orphan-path test
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Addresses non-blocking review feedback on PR #2328:

- test_bare_repo_lock_not_held_across_regenerate_fetch: pin
  fetch count to exactly 2 (initial + regenerate-path) so a
  regression that adds a redundant fetch *inside* the flock —
  masked by other at-depth-0 fetches — is caught. Production
  code path runs exactly one initial fetch (non-failing) plus
  one regenerate fetch when the first push is rejected non-FF.

- test_bare_repo_lock_wraps_branch_d_in_orphan_path: assert
  branch -D runs before worktree add in the orphan path. They
  share one flock window so the depth assertion alone would
  pass even if they were swapped, which would then fail at
  the git layer in production (worktree add against a still-
  existing branch). The new index ordering check guards against
  that swap regression.
---
 gateway/tests/test_checkpoint_handler.py | 27 ++++++++++++++++++++----
 1 file changed, 23 insertions(+), 4 deletions(-)

diff --git a/gateway/tests/test_checkpoint_handler.py b/gateway/tests/test_checkpoint_handler.py
index 8a19037260..d0dbdc86a2 100644
--- a/gateway/tests/test_checkpoint_handler.py
+++ b/gateway/tests/test_checkpoint_handler.py
@@ -1053,11 +1053,12 @@ def track_run_git(cwd, args, **kwargs):
         )
 
         # Both the initial fetch and the regenerate-path fetch must run
-        # outside the flock. Assert there are at least 2 fetches and all
-        # of them are at depth 0.
+        # outside the flock. Pin the count to exactly 2 (initial + regenerate)
+        # so a regression that adds a redundant fetch *inside* the flock
+        # — masked by an at-depth-0 fetch elsewhere — is caught.
         fetch_observations = [depth for args, depth in observations if args[:1] == ["fetch"]]
-        assert len(fetch_observations) >= 2, (
-            f"Expected initial + regenerate fetches, observed {len(fetch_observations)}"
+        assert len(fetch_observations) == 2, (
+            f"Expected exactly 2 fetches (initial + regenerate), observed {len(fetch_observations)}"
         )
         assert all(d == 0 for d in fetch_observations), (
             "bare_repo_lock must not be held during any fetch (including the "
@@ -1121,6 +1122,24 @@ def track_run_git(cwd, args, **kwargs):
             f"`worktree add` must run under bare_repo_lock, saw depths {worktree_add_observations}"
         )
 
+        # ``branch -D`` must run *before* ``worktree add`` in the orphan path.
+        # Running ``worktree add`` against a still-existing branch would fail
+        # at the git layer in production. The two calls share the same flock
+        # window, so the only way to keep them safe is the explicit ordering
+        # at ``checkpoint_handler.py:978-987``. A regression that swapped them
+        # would still pass the depth check above; this assertion catches that.
+        observed_args = [args for args, _ in observations]
+        first_branch_d = next(
+            i for i, args in enumerate(observed_args) if args[:2] == ["branch", "-D"]
+        )
+        first_worktree_add = next(
+            i for i, args in enumerate(observed_args) if args[:2] == ["worktree", "add"]
+        )
+        assert first_branch_d < first_worktree_add, (
+            "`branch -D` must run before `worktree add` in the orphan path "
+            f"(saw branch -D at index {first_branch_d}, worktree add at index {first_worktree_add})"
+        )
+
         # ``checkout --orphan`` runs inside the temp worktree, not the bare
         # repo, so it must run *outside* the flock. A regression that nested
         # it inside the flock window would inflate the cross-process critical