diff --git a/.github/scripts/github-actions-queue/DATA_BRANCH_README.md b/.github/scripts/github-actions-queue/DATA_BRANCH_README.md
index b9bca210965c..44052a23afcf 100644
--- a/.github/scripts/github-actions-queue/DATA_BRANCH_README.md
+++ b/.github/scripts/github-actions-queue/DATA_BRANCH_README.md
@@ -9,6 +9,12 @@ documented one-time migration. `state.json` tracks live collection, and
 `backfill-state.json` tracks automatic historical collection.
 The files under `migrations/` record completed one-time data cleanups.
 
+With `DATA_BRANCH_SNAPSHOTS=true` in the source repository, each update replaces
+the branch with a parentless commit. Current files retain the historical job
+measurements; previous Git snapshots are not retained. The commit's
+`Data-snapshot-generation` trailer orders snapshots without retaining their
+parents.
+
 `report-state.json.gz` tracks the immutable files included in the hourly
 aggregates. `report-data/manifest.json` and the daily JSON files under
 `report-data/` are compact derived inputs for the
diff --git a/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md b/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md
index 3c8a5f837ce6..1018c6992553 100644
--- a/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md
+++ b/.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md
@@ -441,9 +441,12 @@ older worker cannot process active `backfill` or `reminders` items.
    intact so receipts continue suppressing duplicates.
 
 The execution change does not alter accepted-state or delivery formats, so this
-rollback does not require releasing or promoting dashboard code. Existing
-version guards still apply to any separate code rollback. Do not bypass them
-or discard delivery receipts to make older code publish.
+rollback does not require releasing or promoting dashboard code. If data
+snapshots have been activated, keep `DATA_BRANCH_SNAPSHOTS=true` and retain
+snapshot-capable code in every legacy reader and writer; returning to legacy
+execution does not restore Git history. Existing version guards still apply
+to any separate code rollback. Do not bypass them or discard delivery receipts
+to make older code publish.
 
 ### Manual maintenance
 
@@ -465,3 +468,99 @@ Delivery is not exactly once. A GitHub or Slack call can succeed immediately
 before worker death or receipt persistence failure. Existing comment markers,
 live-state reconciliation, and durable delivery receipts suppress ordinary
 duplicates, but that external-success/receipt-loss window remains.
+
+## 7. Data branch snapshot cutover
+
+`DATA_BRANCH_SNAPSHOTS` is a repository-level GitHub Actions variable shared by
+the dashboard and Actions queue collector. Empty or `false` retains ordinary
+Git history. `true` writes parentless snapshots with the complete current tree,
+including historical CI records and delivery acknowledgements. Other values
+fail explicitly.
+
+Snapshot commits have a monotonically increasing `Data-snapshot-generation`
+trailer. Dashboard fetches use it to reject stale snapshots, and every push
+uses the exact SHA loaded before processing as its force-with-lease expectation.
+A rejected dashboard update refetches and reapplies its changes. A rejected
+collector push fails the job; the next run resumes from the accepted checkpoint.
+Unchanged snapshots do not create commits. The first snapshot compacts a legacy
+branch even when its files have not changed.
+
+### Deployment order
+
+1. Merge the snapshot-capable code with the variable unset. Release and promote
+   it through `promote-pull-request-dashboard.yml` so every stable workflow pin
+   loads the new reader and writer. In owned mode, verify the immutable stable
+   code refs used by both live and maintenance drains as well as the canary
+   ref. Verify the legacy queue drain, author reminder sweep, and CI collector
+   also run the new code. Do not rerun older workflow attempts after activation.
+   Snapshot activation requires this promotion even though changing execution
+   ownership alone does not.
+2. Record the current execution mode. Pause the Netlify bridge first, then set
+   GitHub's `PR_DASHBOARD_EXECUTION_MODE` to `paused`, using the commands in
+   section 6. Keep both paused throughout compaction so webhooks remain queued
+   and recovery cannot start either lane. Temporarily disable
+   `pull-request-dashboard.yml`, `pull-request-dashboard-drain.yml`,
+   `pull-request-dashboard-refresh-author-nudges.yml`, and
+   `github-actions-queue-collector.yml`. Wait until every live and maintenance
+   drain, legacy writer, and collector has finished, including queued runs and
+   runnerless publisher jobs. Retiring a specific old run requires separate
+   cancellation approval. Do not take the backup while any writer remains.
+   Keep the stalled-run watchdog enabled; no Netlify code change is required.
+3. From a dedicated maintenance clone, fetch the latest generated branches.
+   Record each branch's head and tree SHA, and create a verified Git bundle
+   backup outside this repository. Never push backup branches or tags here:
+   any ref to old commits keeps that history reachable to clones.
+4. Set the repository variable while writers are paused:
+
+   ```sh
+   gh variable set DATA_BRANCH_SNAPSHOTS --body true --repo open-telemetry/shared-workflows
+   ```
+
+5. Compact `otelbot/github-actions-queue-data` and every
+   `otelbot/pull-request-dashboard-state/*` and
+   `otelbot/pull-request-dashboard-delivery/*` branch. For each branch, use a
+   clean temporary worktree at its recorded head and the helper from the new
+   source checkout. Configure its Git author and authenticated `origin`, then:
+
+   ```sh
+   DATA_BRANCH_SNAPSHOTS=true python3 "$SOURCE_CHECKOUT/.github/scripts/pull-request-dashboard/state_branch.py" commit-data \
+     --state-dir "$DATA_DIRECTORY" \
+     --state-branch "$BRANCH" \
+     --expected-sha "$RECORDED_HEAD" \
+     --message "Compact generated data"
+   ```
+
+   `SOURCE_CHECKOUT` contains the promoted code, `DATA_DIRECTORY` is that
+   branch's temporary worktree, and `RECORDED_HEAD` is the exact backed-up SHA.
+   The helper snapshots the index and working tree, so verify they are clean
+   before invoking it. A lease failure means a writer or operator changed the
+   branch; stop and reconcile it rather than refreshing the lease blindly.
+6. Verify every compacted remote head has no parents and its tree SHA matches
+   the recorded tree. Make a fresh normal clone and measure its object storage.
+   Other branches or tags that reference the old data can still retain history.
+   Reenable the disabled workflows while execution remains paused. Restore
+   GitHub's recorded execution mode, then deploy the bridge with that same mode
+   using the section 6 procedure. Confirm collection, dashboard publication,
+   and delivery acknowledgements continue from their existing state, and in
+   owned mode follow work in both lanes through completion.
+
+### Recovery and clone size
+
+Keep `DATA_BRANCH_SNAPSHOTS=true` after cutover. Turning it off cannot safely
+restore the ancestry-based protocol; the new writer rejects a snapshot branch
+when the variable is disabled. Older workflow code cannot participate in the
+snapshot protocol.
+
+If cutover fails before writers resume, keep them paused, restore the recorded
+heads from the external backup with explicit leases against the compacted
+heads, verify their trees, and unset `DATA_BRANCH_SNAPSHOTS` before restoring
+both execution modes and starting fresh runs.
+After processing resumes, restoring an old backup can discard CI checkpoints
+or repeat notifications. Recover the latest accepted state instead, with a
+snapshot generation higher than the current head.
+
+Compaction removes old history from the generated branch refs, not from
+existing clones. A fresh clone avoids that history once no advertised branch
+or tag retains it. Existing clones also retain local refs, reflogs, and packed
+objects; a fresh clone is the safest way to reclaim that space. GitHub may
+retain unreachable objects and pull request refs separately.
diff --git a/.github/scripts/pull-request-dashboard/state_branch.py b/.github/scripts/pull-request-dashboard/state_branch.py
index 1ad7379b7f04..c12e3cd80935 100644
--- a/.github/scripts/pull-request-dashboard/state_branch.py
+++ b/.github/scripts/pull-request-dashboard/state_branch.py
@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Manage the dashboard workflow's git-backed state branch."""
+"""Manage Git-backed dashboard state and Actions telemetry snapshots."""
 
 from __future__ import annotations
 
@@ -27,6 +27,15 @@
 FETCH_ATTEMPTS = 4
 STATE_BRANCH_PREFIX = "otelbot/pull-request-dashboard-state"
 DELIVERY_STATE_BRANCH_PREFIX = "otelbot/pull-request-dashboard-delivery"
+SNAPSHOT_GENERATION_TRAILER = "Data-snapshot-generation"
+SUBPROCESS_FLAGS = getattr(subprocess, "CREATE_NO_WINDOW", 0)
+
+
+def snapshot_mode() -> bool:
+    value = os.environ.get("DATA_BRANCH_SNAPSHOTS", "")
+    if value not in ("", "false", "true"):
+        raise ValueError("DATA_BRANCH_SNAPSHOTS must be true, false, or empty")
+    return value == "true"
 
 
 def delivery_state_branch(state_branch: str) -> str:
@@ -68,7 +77,9 @@ def accepted_state_dir(state_branch: str, required: bool) -> Iterator[Path | Non
 
 
 def run(cmd: list[str], check: bool = True, cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
-    return subprocess.run(cmd, check=check, cwd=cwd, text=True)
+    return subprocess.run(
+        cmd, check=check, cwd=cwd, text=True, creationflags=SUBPROCESS_FLAGS,
+    )
 
 
 def remote_ref(state_branch: str) -> str:
@@ -129,7 +140,10 @@ def ref_is_ancestor(ancestor: str, descendant: str) -> bool:
         capture_output=True,
         text=True,
         check=False,
+        creationflags=SUBPROCESS_FLAGS,
     )
+    if result.returncode not in (0, 1):
+        raise RuntimeError(f"failed to compare Git refs: {result.stderr.strip()}")
     return result.returncode == 0
 
 
@@ -140,6 +154,7 @@ def ref_oid(ref: str, cwd: Path | None = None) -> str:
         capture_output=True,
         text=True,
         check=False,
+        creationflags=SUBPROCESS_FLAGS,
     )
     if result.returncode != 0:
         message = result.stderr.strip() or result.stdout.strip() or f"exit code {result.returncode}"
@@ -153,6 +168,85 @@ def remote_is_behind_local(state_branch: str, fetched_ref: str) -> bool:
     return ref_is_ancestor(fetched_ref, remote_ref(state_branch))
 
 
+def snapshot_generation(ref: str, cwd: Path | None = None) -> int:
+    result = subprocess.run(
+        ["git", "cat-file", "commit", ref],
+        cwd=cwd, capture_output=True, text=True, check=True,
+        creationflags=SUBPROCESS_FLAGS,
+    )
+    headers, _, message = result.stdout.partition("\n\n")
+    has_parents = any(line.startswith("parent ") for line in headers.splitlines())
+    trailers = [
+        line for line in message.splitlines()
+        if line.startswith(f"{SNAPSHOT_GENERATION_TRAILER}:")
+    ]
+    if not trailers:
+        return 0
+    if (
+        has_parents
+        or len(trailers) != 1
+        or re.fullmatch(rf"{SNAPSHOT_GENERATION_TRAILER}: [1-9][0-9]*", trailers[0]) is None
+    ):
+        raise RuntimeError(f"invalid data snapshot commit: {ref}")
+    return int(trailers[0].split(": ", 1)[1])
+
+
+def has_snapshot_ancestor(oid: str, cwd: Path | None = None) -> bool:
+    result = subprocess.run(
+        [
+            "git", "log", "-n1", "--format=%H", "-E",
+            f"--grep=^{SNAPSHOT_GENERATION_TRAILER}:", oid,
+        ],
+        cwd=cwd, capture_output=True, text=True, check=False,
+        creationflags=SUBPROCESS_FLAGS,
+    )
+    if result.returncode != 0:
+        raise RuntimeError(f"failed to inspect Git history for {oid}: {result.stderr.strip()}")
+    return bool(result.stdout.strip())
+
+
+def update_snapshot_ref(state_branch: str, fetched_ref: str) -> None:
+    destination = remote_ref(state_branch)
+    fetched_oid = ref_oid(fetched_ref)
+    fetched_generation = snapshot_generation(fetched_oid)
+    # A fresh checkout has no local ref, so detect history appended to a snapshot from the commits themselves.
+    if not fetched_generation and has_snapshot_ancestor(fetched_oid):
+        raise RuntimeError(f"history restored after data snapshot for {state_branch}")
+    for _ in range(FETCH_ATTEMPTS):
+        local_oid = ref_oid(destination) if has_state_branch(state_branch) else ""
+        if local_oid == fetched_oid:
+            return
+        if local_oid:
+            local_generation = snapshot_generation(local_oid)
+            if local_generation and fetched_generation:
+                if local_generation == fetched_generation:
+                    raise RuntimeError(f"conflicting data snapshot generation for {state_branch}")
+                newer = fetched_generation > local_generation
+            elif fetched_generation:
+                if ref_is_ancestor(fetched_oid, local_oid):
+                    raise RuntimeError(f"history restored after data snapshot for {state_branch}")
+                newer = True
+            elif local_generation:
+                if ref_is_ancestor(local_oid, fetched_oid):
+                    raise RuntimeError(f"history restored after data snapshot for {state_branch}")
+                newer = False
+            else:
+                newer = ref_is_ancestor(local_oid, fetched_oid)
+                if not newer and not ref_is_ancestor(fetched_oid, local_oid):
+                    raise RuntimeError(f"fetched state branch {state_branch} diverged from the local ref")
+            if not newer:
+                print(f"remote {state_branch} is behind the local ref; keeping the local ref", file=sys.stderr)
+                return
+        # A concurrent fetch must not replace a newer local snapshot.
+        result = run(
+            ["git", "update-ref", destination, fetched_oid, local_oid],
+            check=False,
+        )
+        if result.returncode == 0:
+            return
+    raise RuntimeError(f"could not update local snapshot ref for {state_branch}")
+
+
 def fetch_state_branch(state_branch: str, required: bool) -> bool:
     fetched_ref = temporary_fetch_ref()
     try:
@@ -162,12 +256,14 @@ def fetch_state_branch(state_branch: str, required: bool) -> bool:
                     "git",
                     "fetch",
                     "--no-write-fetch-head",
+                    "--refmap=",
                     "origin",
                     f"{state_branch}:{fetched_ref}",
                 ],
                 capture_output=True,
                 text=True,
                 check=False,
+                creationflags=SUBPROCESS_FLAGS,
             )
             if proc.returncode == 0:
                 break
@@ -184,6 +280,10 @@ def fetch_state_branch(state_branch: str, required: bool) -> bool:
                 )
             time.sleep(retry_delay_seconds(attempt))
 
+        if snapshot_mode():
+            update_snapshot_ref(state_branch, fetched_ref)
+            return True
+
         destination = remote_ref(state_branch)
         if not has_state_branch(state_branch) or ref_is_ancestor(destination, fetched_ref):
             run(["git", "update-ref", destination, fetched_ref])
@@ -217,15 +317,16 @@ def remove_existing_state_dir(state_dir: Path) -> None:
         state_dir.unlink()
 
 
-def checkout_state(state_dir: Path, state_branch: str, require_existing: bool) -> None:
+def checkout_state(state_dir: Path, state_branch: str, require_existing: bool) -> str:
     remove_existing_state_dir(state_dir)
     fetch_state_branch(state_branch, required=require_existing)
     if has_state_branch(state_branch):
         run(["git", "worktree", "add", "-B", state_branch, str(state_dir), f"origin/{state_branch}"])
-        return
+        return ref_oid("HEAD", cwd=state_dir)
     run(["git", "worktree", "add", "--detach", str(state_dir), "HEAD"])
     run(["git", "switch", "--orphan", state_branch], cwd=state_dir)
     run(["git", "rm", "-rf", "."], cwd=state_dir, check=False)
+    return ""
 
 
 def reset_state(state_dir: Path, state_branch: str) -> bool:
@@ -235,7 +336,7 @@ def reset_state(state_dir: Path, state_branch: str) -> bool:
     return True
 
 
-def push_state(state_dir: Path, state_branch: str) -> bool:
+def push_state(state_dir: Path, state_branch: str, expected_sha: str) -> bool:
     env = dict(os.environ)
     token = os.environ.get("GITHUB_TOKEN")
     if token:
@@ -243,8 +344,48 @@ def push_state(state_dir: Path, state_branch: str) -> bool:
         env["GIT_CONFIG_COUNT"] = "1"
         env["GIT_CONFIG_KEY_0"] = "http.https://github.com/.extraheader"
         env["GIT_CONFIG_VALUE_0"] = f"AUTHORIZATION: basic {credential}"
-    cmd = ["git", "push", "--force-with-lease", "origin", state_branch]
-    return subprocess.run(cmd, cwd=state_dir, check=False, text=True, env=env).returncode == 0
+    cmd = [
+        "git", "push", f"--force-with-lease=refs/heads/{state_branch}:{expected_sha}",
+        "origin", f"HEAD:refs/heads/{state_branch}",
+    ]
+    return subprocess.run(
+        cmd, cwd=state_dir, check=False, text=True, env=env,
+        creationflags=SUBPROCESS_FLAGS,
+    ).returncode == 0
+
+
+def commit_staged_state(state_dir: Path, message: str, expected_sha: str) -> bool:
+    enabled = snapshot_mode()
+    generation = snapshot_generation(expected_sha, cwd=state_dir) if expected_sha else 0
+    if generation and not enabled:
+        raise RuntimeError("DATA_BRANCH_SNAPSHOTS must remain true for snapshot branches")
+    if enabled and expected_sha and not generation and has_snapshot_ancestor(expected_sha, cwd=state_dir):
+        raise RuntimeError("history restored after data snapshot")
+    diff = run(
+        ["git", "diff", "--cached", "--quiet"], cwd=state_dir, check=False,
+    )
+    if diff.returncode not in (0, 1):
+        raise RuntimeError(f"failed to check staged data: exit code {diff.returncode}")
+    unchanged = diff.returncode == 0
+    if unchanged and (not enabled or generation):
+        print("no state changes to push", file=sys.stderr)
+        return False
+    if not enabled:
+        run(["git", "commit", "-m", message], cwd=state_dir)
+        return True
+
+    tree = subprocess.run(
+        ["git", "write-tree"], cwd=state_dir, capture_output=True,
+        text=True, check=True, creationflags=SUBPROCESS_FLAGS,
+    ).stdout.strip()
+    snapshot = subprocess.run(
+        ["git", "commit-tree", tree],
+        input=f"{message}\n\n{SNAPSHOT_GENERATION_TRAILER}: {generation + 1}\n",
+        cwd=state_dir, capture_output=True, text=True, check=True,
+        creationflags=SUBPROCESS_FLAGS,
+    ).stdout.strip()
+    run(["git", "update-ref", "HEAD", snapshot, expected_sha], cwd=state_dir)
+    return True
 
 
 def configure_git() -> None:
@@ -269,6 +410,7 @@ def set_git_config(name: str, value: str) -> None:
             check=False,
             text=True,
             capture_output=True,
+            creationflags=SUBPROCESS_FLAGS,
         )
         if proc.returncode == 0:
             return
@@ -306,7 +448,7 @@ def push_state_changes(
     retry_snapshots: list[tuple[Path, Path]] | None = None,
 ) -> int:
     configure_git()
-    checkout_state(state_dir, state_branch, require_existing=False)
+    expected_sha = checkout_state(state_dir, state_branch, require_existing=False)
     paths_to_add = add_paths or ["."]
     snapshots = retry_snapshots or []
 
@@ -326,14 +468,12 @@ def push_state_changes(
                 ["git", "add", "--update", "--", ".publisher-lock.json"],
                 cwd=state_dir,
             )
-        if run(["git", "diff", "--cached", "--quiet"], cwd=state_dir, check=False).returncode == 0:
-            print("no state changes to push", file=sys.stderr)
+        if not commit_staged_state(state_dir, commit_message, expected_sha):
             return 0
 
-        run(["git", "commit", "-m", commit_message], cwd=state_dir)
         copy_snapshots(snapshots)
 
-        if push_state(state_dir, state_branch):
+        if push_state(state_dir, state_branch, expected_sha):
             print(f"state pushed on attempt {attempt}", file=sys.stderr)
             return 0
 
@@ -349,6 +489,7 @@ def push_state_changes(
         time.sleep(delay)
         if not reset_state(state_dir, state_branch):
             return 1
+        expected_sha = ref_oid("HEAD", cwd=state_dir)
     return 1
 
 
@@ -358,12 +499,25 @@ def main() -> int:
     checkout = subparsers.add_parser("checkout", help="check out the accepted state branch")
     checkout.add_argument("--state-branch", required=True)
     checkout.add_argument("--state-dir", type=Path, required=True)
+    commit = subparsers.add_parser("commit-data", help="commit and push a collected data tree")
+    commit.add_argument("--state-branch", required=True)
+    commit.add_argument("--state-dir", type=Path, required=True)
+    commit.add_argument("--expected-sha", required=True)
+    commit.add_argument("--message", required=True)
     args = parser.parse_args()
 
     if args.command == "checkout":
         configure_git()
         checkout_state(args.state_dir, args.state_branch, require_existing=True)
         return 0
+    if args.command == "commit-data":
+        run(["git", "add", "--all"], cwd=args.state_dir)
+        if not commit_staged_state(args.state_dir, args.message, args.expected_sha):
+            return 0
+        if not push_state(args.state_dir, args.state_branch, args.expected_sha):
+            print("data push rejected; collection will resume from the accepted checkpoint", file=sys.stderr)
+            return 1
+        return 0
     parser.error(f"unknown command: {args.command}")
     return 2
 
diff --git a/.github/scripts/pull-request-dashboard/test_data_snapshots.py b/.github/scripts/pull-request-dashboard/test_data_snapshots.py
new file mode 100644
index 000000000000..61758b196ff8
--- /dev/null
+++ b/.github/scripts/pull-request-dashboard/test_data_snapshots.py
@@ -0,0 +1,326 @@
+from __future__ import annotations
+
+from collections.abc import Iterator
+from contextlib import contextmanager
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import state_branch
+from test_state_branch import run_git
+
+
+@contextmanager
+def working_directory(directory: Path) -> Iterator[None]:
+    previous = Path.cwd()
+    os.chdir(directory)
+    try:
+        yield
+    finally:
+        os.chdir(previous)
+
+
+class DataSnapshotTest(unittest.TestCase):
+    def setUp(self) -> None:
+        temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(temporary.cleanup)
+        self.root = Path(temporary.name)
+        self.remote = self.root / "remote.git"
+        self.worker = self.root / "worker"
+        self.worker.mkdir()
+        run_git(self.root, "init", "--bare", str(self.remote))
+        run_git(self.worker, "init", "-b", "data")
+        run_git(self.worker, "config", "user.email", "test@example.com")
+        run_git(self.worker, "config", "user.name", "Test")
+        run_git(self.worker, "remote", "add", "origin", str(self.remote))
+        (self.worker / "jobs.json").write_text('{"jobs":[1]}\n', encoding="utf-8")
+        (self.worker / "ledger.json").write_text('{"sent":[1]}\n', encoding="utf-8")
+        run_git(self.worker, "add", ".")
+        run_git(self.worker, "commit", "-m", "Initial data")
+        run_git(self.worker, "push", "-u", "origin", "data")
+        self.legacy = run_git(self.worker, "rev-parse", "HEAD").stdout.strip()
+        run_git(self.worker, "switch", "--detach")
+        self.environment = patch.dict(
+            os.environ, {"DATA_BRANCH_SNAPSHOTS": "true", "GITHUB_TOKEN": ""},
+        )
+        self.environment.start()
+        self.addCleanup(self.environment.stop)
+
+    def remote_git(self, *args: str) -> str:
+        return run_git(self.root, f"--git-dir={self.remote}", *args).stdout
+
+    def snapshot(self, directory: Path, expected_sha: str) -> str:
+        run_git(directory, "add", "--all")
+        self.assertTrue(state_branch.commit_staged_state(directory, "Update data", expected_sha))
+        return run_git(directory, "rev-parse", "HEAD").stdout.strip()
+
+    def test_compacts_unchanged_legacy_tree_and_preserves_all_files(self) -> None:
+        before = run_git(self.worker, "rev-parse", "HEAD^{tree}").stdout
+        snapshot = self.snapshot(self.worker, self.legacy)
+        self.assertEqual(before, run_git(self.worker, "rev-parse", "HEAD^{tree}").stdout)
+        self.assertEqual("1\n", run_git(self.worker, "rev-list", "--count", "HEAD").stdout)
+        self.assertEqual(1, state_branch.snapshot_generation(snapshot, self.worker))
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        clone = self.root / "fresh-clone"
+        run_git(self.root, "clone", "--no-local", "--branch", "data", str(self.remote), str(clone))
+        self.assertEqual("1\n", run_git(clone, "rev-list", "--count", "HEAD").stdout)
+        legacy_object = subprocess.run(
+            ["git", "cat-file", "-e", self.legacy], cwd=clone,
+            check=False, capture_output=True, text=True,
+            creationflags=state_branch.SUBPROCESS_FLAGS,
+        )
+        self.assertNotEqual(0, legacy_object.returncode)
+        self.assertEqual('{"jobs":[1]}\n', (clone / "jobs.json").read_text(encoding="utf-8"))
+        self.assertEqual('{"sent":[1]}\n', (clone / "ledger.json").read_text(encoding="utf-8"))
+        self.assertFalse(state_branch.commit_staged_state(self.worker, "No changes", snapshot))
+        self.assertEqual(snapshot, run_git(self.worker, "rev-parse", "HEAD").stdout.strip())
+
+    def test_repeated_updates_keep_one_commit_and_increment_generation(self) -> None:
+        expected = self.legacy
+        for generation in range(1, 4):
+            (self.worker / "jobs.json").write_text(f'{{"jobs":[1,{generation}]}}\n', encoding="utf-8")
+            snapshot = self.snapshot(self.worker, expected)
+            self.assertTrue(state_branch.push_state(self.worker, "data", expected))
+            self.assertEqual("1\n", self.remote_git("rev-list", "--count", "data"))
+            self.assertEqual(generation, state_branch.snapshot_generation(snapshot, self.worker))
+            self.assertEqual('{"sent":[1]}\n', (self.worker / "ledger.json").read_text(encoding="utf-8"))
+            expected = snapshot
+
+    def test_explicit_lease_rejects_stale_writer_even_after_background_fetch(self) -> None:
+        other = self.root / "other"
+        run_git(self.root, "clone", "--branch", "data", str(self.remote), str(other))
+        run_git(other, "config", "user.email", "test@example.com")
+        run_git(other, "config", "user.name", "Test")
+        (self.worker / "jobs.json").write_text('{"jobs":[1,2]}\n', encoding="utf-8")
+        accepted = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        (other / "ledger.json").write_text('{"sent":[1,2]}\n', encoding="utf-8")
+        self.snapshot(other, self.legacy)
+        run_git(other, "fetch", "origin")
+        self.assertFalse(state_branch.push_state(other, "data", self.legacy))
+        self.assertEqual(accepted, self.remote_git("rev-parse", "data").strip())
+
+    def test_dashboard_retry_reapplies_update_to_winning_snapshot(self) -> None:
+        worktree = self.root / "dashboard"
+        real_push = state_branch.push_state
+        attempts = 0
+
+        def competing_push(directory: Path, branch: str, expected: str) -> bool:
+            nonlocal attempts
+            attempts += 1
+            if attempts == 1:
+                (self.worker / "ledger.json").write_text('{"sent":[1,2]}\n', encoding="utf-8")
+                self.snapshot(self.worker, self.legacy)
+                self.assertTrue(real_push(self.worker, branch, self.legacy))
+            return real_push(directory, branch, expected)
+
+        def update() -> int:
+            (worktree / "jobs.json").write_text('{"jobs":[1,2]}\n', encoding="utf-8")
+            return 0
+
+        with (
+            working_directory(self.worker),
+            patch.object(state_branch, "push_state", side_effect=competing_push),
+            patch.object(state_branch.time, "sleep"),
+        ):
+            status = state_branch.push_state_changes(
+                worktree, "Update state", update, state_branch="data",
+            )
+        self.assertEqual(0, status)
+        self.assertEqual(2, attempts)
+        self.assertEqual('{"sent":[1,2]}\n', (worktree / "ledger.json").read_text(encoding="utf-8"))
+        self.assertEqual('{"jobs":[1,2]}\n', (worktree / "jobs.json").read_text(encoding="utf-8"))
+        self.assertEqual("1\n", self.remote_git("rev-list", "--count", "data"))
+        self.assertEqual(2, state_branch.snapshot_generation("HEAD", worktree))
+
+    def test_fetch_ignores_stale_snapshots_and_legacy_heads(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        (self.worker / "jobs.json").write_text('{"jobs":[1,2]}\n', encoding="utf-8")
+        second = self.snapshot(self.worker, first)
+        self.assertTrue(state_branch.push_state(self.worker, "data", first))
+        with working_directory(self.worker):
+            state_branch.update_snapshot_ref("data", first)
+            state_branch.update_snapshot_ref("data", self.legacy)
+            self.assertEqual(second, state_branch.ref_oid(state_branch.remote_ref("data")))
+            self.assertTrue(state_branch.fetch_state_branch("data", required=True))
+        self.assertEqual("", run_git(self.worker, "for-each-ref", "--format=%(refname)", "refs/pull-request-dashboard-fetch").stdout)
+
+    def test_fetch_preserves_newer_local_snapshot_when_remote_is_stale(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        (self.worker / "jobs.json").write_text('{"jobs":[1,2]}\n', encoding="utf-8")
+        second = self.snapshot(self.worker, first)
+        run_git(self.worker, "update-ref", state_branch.remote_ref("data"), second)
+
+        with working_directory(self.worker):
+            self.assertTrue(state_branch.fetch_state_branch("data", required=True))
+
+        self.assertEqual(second, run_git(self.worker, "rev-parse", state_branch.remote_ref("data")).stdout.strip())
+
+    def test_fetch_replaces_legacy_ref_with_snapshot_and_accepts_newer_generation(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        (self.worker / "jobs.json").write_text('{"jobs":[1,2]}\n', encoding="utf-8")
+        second = self.snapshot(self.worker, first)
+        with working_directory(self.worker):
+            state_branch.update_snapshot_ref("data", first)
+            self.assertEqual(first, state_branch.ref_oid(state_branch.remote_ref("data")))
+            state_branch.update_snapshot_ref("data", second)
+            self.assertEqual(second, state_branch.ref_oid(state_branch.remote_ref("data")))
+
+    def test_same_generation_with_different_contents_is_an_error(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        run_git(self.worker, "update-ref", "HEAD", self.legacy)
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        run_git(self.worker, "add", "--all")
+        conflicting = self.snapshot(self.worker, self.legacy)
+        with working_directory(self.worker), self.assertRaisesRegex(RuntimeError, "conflicting"):
+            state_branch.update_snapshot_ref("data", conflicting)
+        self.assertEqual(first, run_git(self.worker, "rev-parse", "origin/data").stdout.strip())
+
+    def test_concurrent_fetch_does_not_regress_local_ref(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        second = self.snapshot(self.worker, first)
+        real_run = state_branch.run
+
+        def concurrent_update(
+            cmd: list[str], *, check: bool = True, cwd: Path | None = None,
+        ) -> subprocess.CompletedProcess[str]:
+            if cmd[:3] == ["git", "update-ref", state_branch.remote_ref("data")]:
+                run_git(self.worker, "update-ref", state_branch.remote_ref("data"), second)
+            return real_run(cmd, check=check, cwd=cwd)
+
+        with working_directory(self.worker), patch.object(state_branch, "run", side_effect=concurrent_update):
+            state_branch.update_snapshot_ref("data", first)
+        self.assertEqual(second, run_git(self.worker, "rev-parse", "origin/data").stdout.strip())
+
+    def test_history_bearing_write_after_snapshot_is_rejected(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        run_git(self.worker, "add", ".")
+        run_git(self.worker, "commit", "-m", "History-bearing update")
+        with working_directory(self.worker), self.assertRaisesRegex(RuntimeError, "history restored"):
+            state_branch.update_snapshot_ref("data", "HEAD")
+
+    def test_fresh_checkout_rejects_history_bearing_head_after_snapshot(self) -> None:
+        self.snapshot(self.worker, self.legacy)
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        run_git(self.worker, "add", ".")
+        run_git(self.worker, "commit", "-m", "History-bearing update")
+        run_git(self.worker, "update-ref", "-d", state_branch.remote_ref("data"))
+        with working_directory(self.worker), self.assertRaisesRegex(RuntimeError, "history restored"):
+            state_branch.update_snapshot_ref("data", "HEAD")
+
+    def test_commit_rejects_history_appended_to_snapshot(self) -> None:
+        self.snapshot(self.worker, self.legacy)
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        run_git(self.worker, "add", ".")
+        run_git(self.worker, "commit", "-m", "History-bearing update")
+        appended = run_git(self.worker, "rev-parse", "HEAD").stdout.strip()
+        run_git(self.worker, "add", ".")
+        with self.assertRaisesRegex(RuntimeError, "history restored"):
+            state_branch.commit_staged_state(self.worker, "Update", appended)
+
+    def test_disabled_mode_preserves_legacy_behavior_but_cannot_restore_history(self) -> None:
+        with patch.dict(os.environ, {"DATA_BRANCH_SNAPSHOTS": "false"}):
+            self.assertFalse(state_branch.commit_staged_state(self.worker, "No changes", self.legacy))
+            (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+            run_git(self.worker, "add", ".")
+            self.assertTrue(state_branch.commit_staged_state(self.worker, "Update", self.legacy))
+            self.assertEqual("2\n", run_git(self.worker, "rev-list", "--count", "HEAD").stdout)
+        first = self.snapshot(self.worker, run_git(self.worker, "rev-parse", "HEAD").stdout.strip())
+        with patch.dict(os.environ, {"DATA_BRANCH_SNAPSHOTS": ""}), self.assertRaisesRegex(RuntimeError, "must remain true"):
+            state_branch.commit_staged_state(self.worker, "No changes", first)
+
+    def test_invalid_mode_is_an_error(self) -> None:
+        with patch.dict(os.environ, {"DATA_BRANCH_SNAPSHOTS": "yes"}), self.assertRaisesRegex(ValueError, "must be"):
+            state_branch.commit_staged_state(self.worker, "Update", self.legacy)
+
+    def test_interrupted_update_does_not_modify_remote(self) -> None:
+        self.snapshot(self.worker, self.legacy)
+        self.assertEqual(self.legacy, self.remote_git("rev-parse", "data").strip())
+
+    def test_invalid_snapshot_metadata_is_rejected_even_in_shallow_checkout(self) -> None:
+        first = self.snapshot(self.worker, self.legacy)
+        self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        run_git(self.worker, "add", ".")
+        run_git(self.worker, "commit", "-m", "Invalid snapshot\n\nData-snapshot-generation: 2")
+        invalid = run_git(self.worker, "rev-parse", "HEAD").stdout.strip()
+        run_git(self.worker, "push", "origin", "HEAD:refs/heads/data")
+        shallow = self.root / "shallow"
+        run_git(self.root, "clone", "--depth=1", "--branch", "data", self.remote.as_uri(), str(shallow))
+        with self.assertRaisesRegex(RuntimeError, "invalid data snapshot"):
+            state_branch.snapshot_generation(invalid, shallow)
+
+    def test_git_helpers_supply_no_window_flag(self) -> None:
+        with patch.object(subprocess, "run", wraps=subprocess.run) as launch:
+            self.snapshot(self.worker, self.legacy)
+            self.assertTrue(state_branch.push_state(self.worker, "data", self.legacy))
+            with working_directory(self.worker):
+                self.assertTrue(state_branch.fetch_state_branch("data", required=True))
+        self.assertTrue(launch.call_args_list)
+        for invocation in launch.call_args_list:
+            self.assertEqual(
+                getattr(subprocess, "CREATE_NO_WINDOW", 0),
+                invocation.kwargs["creationflags"],
+            )
+
+    def test_collector_command_preserves_measurements_and_checkpoints(self) -> None:
+        (self.worker / "state.json").write_text('{"cursor":"2026-10-01T17:00:00Z"}\n', encoding="utf-8")
+        command = [
+            sys.executable, str(Path(state_branch.__file__).resolve()), "commit-data",
+            "--state-dir", str(self.worker), "--state-branch", "data",
+            "--expected-sha", self.legacy, "--message", "Collect queue data",
+        ]
+        result = subprocess.run(
+            command, cwd=self.worker, check=True, capture_output=True, text=True,
+            creationflags=state_branch.SUBPROCESS_FLAGS,
+        )
+        self.assertEqual(0, result.returncode)
+        self.assertEqual("1\n", self.remote_git("rev-list", "--count", "data"))
+        self.assertEqual('{"jobs":[1]}\n', self.remote_git("show", "data:jobs.json"))
+        self.assertEqual('{"cursor":"2026-10-01T17:00:00Z"}\n', self.remote_git("show", "data:state.json"))
+        accepted = run_git(self.worker, "rev-parse", "HEAD").stdout.strip()
+        resumed = self.root / "resumed"
+        run_git(self.root, "clone", "--branch", "data", str(self.remote), str(resumed))
+        run_git(resumed, "config", "user.email", "test@example.com")
+        run_git(resumed, "config", "user.name", "Test")
+        self.assertEqual('{"cursor":"2026-10-01T17:00:00Z"}\n', (resumed / "state.json").read_text(encoding="utf-8"))
+        (resumed / "state.json").write_text('{"cursor":"2026-10-01T18:00:00Z"}\n', encoding="utf-8")
+        (resumed / "new-jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        subprocess.run(
+            [
+                sys.executable, str(Path(state_branch.__file__).resolve()), "commit-data",
+                "--state-dir", str(resumed), "--state-branch", "data",
+                "--expected-sha", accepted, "--message", "Resume collection",
+            ],
+            cwd=resumed, check=True, capture_output=True, text=True,
+            creationflags=state_branch.SUBPROCESS_FLAGS,
+        )
+        self.assertEqual("1\n", self.remote_git("rev-list", "--count", "data"))
+        self.assertEqual('{"jobs":[1]}\n', self.remote_git("show", "data:jobs.json"))
+        self.assertEqual('{"jobs":[2]}\n', self.remote_git("show", "data:new-jobs.json"))
+        self.assertEqual('{"cursor":"2026-10-01T18:00:00Z"}\n', self.remote_git("show", "data:state.json"))
+
+    def test_first_snapshot_creates_missing_branch_with_empty_lease(self) -> None:
+        new_branch = "new-data"
+        run_git(self.worker, "switch", "--orphan", new_branch)
+        (self.worker / "jobs.json").write_text('{"jobs":[1]}\n', encoding="utf-8")
+        first = self.snapshot(self.worker, "")
+        self.assertTrue(state_branch.push_state(self.worker, new_branch, ""))
+        (self.worker / "jobs.json").write_text('{"jobs":[2]}\n', encoding="utf-8")
+        self.snapshot(self.worker, first)
+        self.assertFalse(state_branch.push_state(self.worker, new_branch, ""))
+        self.assertEqual("1\n", self.remote_git("rev-list", "--count", new_branch))
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/.github/scripts/pull-request-dashboard/test_owned_execution.py b/.github/scripts/pull-request-dashboard/test_owned_execution.py
index b0025f0e8335..b05076db9287 100644
--- a/.github/scripts/pull-request-dashboard/test_owned_execution.py
+++ b/.github/scripts/pull-request-dashboard/test_owned_execution.py
@@ -301,6 +301,7 @@ def test_stable_claims_are_processed_not_dispatched_and_use_current_config(self)
                     {"itemKey": claim.item_key, "outcome": "success"},
                 ])) as process,
                 mock.patch.object(drain_queue, "acknowledge_results"),
+                mock.patch.dict(os.environ, {"DATA_BRANCH_SNAPSHOTS": "true"}, clear=True),
             ):
                 dispatch = mock.Mock()
                 drain_queue.process_claim_wave(
@@ -319,6 +320,7 @@ def test_stable_claims_are_processed_not_dispatched_and_use_current_config(self)
             # Omitting config_path deliberately uses the current worker's configuration.
             self.assertNotIn("config_path", process.call_args.kwargs)
             self.assertNotIn("PR_DASHBOARD_PRIVATE_KEY", process.call_args.kwargs["processor_env"])
+            self.assertEqual(process.call_args.kwargs["processor_env"]["DATA_BRANCH_SNAPSHOTS"], "true")
 
     def test_owned_wave_requires_an_isolated_checkout_before_processing(self) -> None:
         claim = Claim("example#pr:1", 1, "example", 1, "", 0)
diff --git a/.github/scripts/pull-request-dashboard/test_rollout.py b/.github/scripts/pull-request-dashboard/test_rollout.py
index cd8b5d40fddd..0a9e044f95c5 100644
--- a/.github/scripts/pull-request-dashboard/test_rollout.py
+++ b/.github/scripts/pull-request-dashboard/test_rollout.py
@@ -148,6 +148,37 @@ def test_unit_tests_install_sdk_without_downloading_runtime(self) -> None:
         self.assertIn("python -m pip install -r requirements.txt", text)
         self.assertNotIn("download-runtime", text)
 
+    def test_all_data_writers_share_the_snapshot_activation_variable(self) -> None:
+        for path in (
+            REPO_WORKFLOW,
+            DRAIN_WORKFLOW,
+            SWEEP_WORKFLOW,
+            WORKFLOW.parent / "github-actions-queue-collector.yml",
+        ):
+            with self.subTest(workflow=path.name):
+                self.assertIn(
+                    "DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}",
+                    path.read_text(encoding="utf-8"),
+                )
+
+    def test_both_drain_jobs_receive_the_snapshot_activation_variable(self) -> None:
+        jobs = job_blocks(DRAIN_WORKFLOW.read_text(encoding="utf-8"))
+        for name in ("owned-drain", "drain"):
+            with self.subTest(job=name):
+                self.assertIn(
+                    "DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}",
+                    jobs[name],
+                )
+
+    def test_collector_uses_the_checkout_sha_as_an_explicit_push_lease(self) -> None:
+        body = (WORKFLOW.parent / "github-actions-queue-collector.yml").read_text(encoding="utf-8")
+        self.assertIn('expected_sha=$(git -C "$data_dir" rev-parse HEAD)', body)
+        self.assertIn('echo "expected_sha=$expected_sha" >> "$GITHUB_OUTPUT"', body)
+        self.assertIn("EXPECTED_SHA: ${{ steps.data.outputs.expected_sha }}", body)
+        self.assertIn("state_branch.py commit-data", body)
+        self.assertIn('--expected-sha "$EXPECTED_SHA"', body)
+        self.assertNotIn('push origin "HEAD:', body)
+
     def test_targeted_canary_job_inlines_the_workflow_canary_list(self) -> None:
         body = self.jobs["run-targeted-dashboard-canary"]
         inline = re.findall(r"fromJSON\('(\[[^']*\])'\)", body)
diff --git a/.github/scripts/pull-request-dashboard/test_state_branch.py b/.github/scripts/pull-request-dashboard/test_state_branch.py
index f171c6d54995..e47f7180e460 100644
--- a/.github/scripts/pull-request-dashboard/test_state_branch.py
+++ b/.github/scripts/pull-request-dashboard/test_state_branch.py
@@ -83,7 +83,7 @@ def test_returns_none_when_optional_state_is_missing(
 
 
 class LegacyPublisherLockCleanupTest(unittest.TestCase):
-    @patch.object(state_branch, "checkout_state")
+    @patch.object(state_branch, "checkout_state", return_value="")
     @patch.object(state_branch, "configure_git")
     def test_worker_update_removes_legacy_lock_without_waiting(
         self,
@@ -109,7 +109,7 @@ def update_state() -> int:
         self.assertEqual(1, status)
 
     @patch.object(state_branch, "push_state")
-    @patch.object(state_branch, "checkout_state")
+    @patch.object(state_branch, "checkout_state", return_value="")
     @patch.object(state_branch, "configure_git")
     def test_narrow_update_commits_legacy_lock_deletion(
         self,
@@ -136,7 +136,7 @@ def update_state() -> int:
                 state_file.write_text('{"revision":2}\n', encoding="utf-8")
                 return 0
 
-            def inspect_pushed_commit(_state_dir: Path, _state_branch: str) -> bool:
+            def inspect_pushed_commit(_state_dir: Path, _state_branch: str, _expected_sha: str) -> bool:
                 changed_paths = run_git(
                     state_dir,
                     "diff-tree",
@@ -163,7 +163,7 @@ def inspect_pushed_commit(_state_dir: Path, _state_branch: str) -> bool:
             )
 
         self.assertEqual(0, status)
-        push_state.assert_called_once_with(state_dir, "state-branch")
+        push_state.assert_called_once_with(state_dir, "state-branch", "")
 
 
 class FetchStateBranchTest(unittest.TestCase):
@@ -217,24 +217,28 @@ def test_recovers_from_transient_fetch_error(
                         "git",
                         "fetch",
                         "--no-write-fetch-head",
+                        "--refmap=",
                         "origin",
                         "state-branch:refs/temp/fetch",
                     ],
                     capture_output=True,
                     text=True,
                     check=False,
+                    creationflags=state_branch.SUBPROCESS_FLAGS,
                 ),
                 call(
                     [
                         "git",
                         "fetch",
                         "--no-write-fetch-head",
+                        "--refmap=",
                         "origin",
                         "state-branch:refs/temp/fetch",
                     ],
                     capture_output=True,
                     text=True,
                     check=False,
+                    creationflags=state_branch.SUBPROCESS_FLAGS,
                 ),
             ],
             subprocess_run.call_args_list,
@@ -289,12 +293,14 @@ def test_raises_after_transient_fetch_retries_are_exhausted(
                         "git",
                         "fetch",
                         "--no-write-fetch-head",
+                        "--refmap=",
                         "origin",
                         "state-branch:refs/temp/fetch",
                     ],
                     capture_output=True,
                     text=True,
                     check=False,
+                    creationflags=state_branch.SUBPROCESS_FLAGS,
                 )
             ]
             * 4,
@@ -335,12 +341,14 @@ def test_returns_false_for_missing_optional_branch_without_retry(
                 "git",
                 "fetch",
                 "--no-write-fetch-head",
+                "--refmap=",
                 "origin",
                 "state-branch:refs/temp/fetch",
             ],
             capture_output=True,
             text=True,
             check=False,
+            creationflags=state_branch.SUBPROCESS_FLAGS,
         )
         sleep.assert_not_called()
         run.assert_called_once_with(
@@ -374,12 +382,14 @@ def test_raises_permanent_fetch_error_without_retry(
                 "git",
                 "fetch",
                 "--no-write-fetch-head",
+                "--refmap=",
                 "origin",
                 "state-branch:refs/temp/fetch",
             ],
             capture_output=True,
             text=True,
             check=False,
+            creationflags=state_branch.SUBPROCESS_FLAGS,
         )
         sleep.assert_not_called()
         run.assert_called_once_with(
@@ -416,6 +426,7 @@ def test_keeps_local_ref_when_remote_is_behind(
                 "git",
                 "fetch",
                 "--no-write-fetch-head",
+                "--refmap=",
                 "origin",
                 "state-branch:refs/temp/fetch",
             ],
diff --git a/.github/workflows/github-actions-queue-collector.yml b/.github/workflows/github-actions-queue-collector.yml
index 00cc0c86385c..d7893af9e975 100644
--- a/.github/workflows/github-actions-queue-collector.yml
+++ b/.github/workflows/github-actions-queue-collector.yml
@@ -60,18 +60,32 @@ jobs:
         id: data
         env:
           DATA_BRANCH_README: .github/scripts/github-actions-queue/DATA_BRANCH_README.md
+          DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
         run: |
           set -euo pipefail
           data_dir="$RUNNER_TEMP/github-actions-queue-data"
+          expected_sha=
           if git ls-remote --exit-code --heads origin "refs/heads/$DATA_BRANCH" >/dev/null; then
             git fetch --no-tags --depth=1 origin "$DATA_BRANCH"
+            # In snapshot mode, a head without a snapshot trailer needs its full history so commit-data can reject appended history.
+            if [[ "$DATA_BRANCH_SNAPSHOTS" == "true" ]] &&
+              ! git log -1 --format=%B FETCH_HEAD | grep -q '^Data-snapshot-generation:'; then
+              git fetch --no-tags --depth=2147483647 origin "$DATA_BRANCH"
+            fi
             git worktree add --detach "$data_dir" FETCH_HEAD
+            expected_sha=$(git -C "$data_dir" rev-parse HEAD)
           else
+            status=$?
+            if [[ "$status" != 2 ]]; then
+              echo "::error::Could not check the data branch"
+              exit "$status"
+            fi
             git worktree add --detach "$data_dir" HEAD
             git -C "$data_dir" switch --orphan "$DATA_BRANCH"
           fi
           cp "$DATA_BRANCH_README" "$data_dir/README.md"
           echo "directory=$data_dir" >> "$GITHUB_OUTPUT"
+          echo "expected_sha=$expected_sha" >> "$GITHUB_OUTPUT"
 
       - name: Remove records without measurable queue time
         id: migration
@@ -175,19 +189,18 @@ jobs:
         env:
           CURSOR: ${{ steps.collect.outputs.cursor }}
           DATA_DIRECTORY: ${{ steps.data.outputs.directory }}
-          GH_TOKEN: ${{ github.token }}
+          EXPECTED_SHA: ${{ steps.data.outputs.expected_sha }}
+          DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
+          GITHUB_TOKEN: ${{ github.token }}
         run: |
           set -euo pipefail
           git -C "$DATA_DIRECTORY" config user.name "github-actions[bot]"
           git -C "$DATA_DIRECTORY" config user.email "41898282+github-actions[bot]@users.noreply.github.com"
-          git -C "$DATA_DIRECTORY" add --all
-          if git -C "$DATA_DIRECTORY" diff --cached --quiet; then
-            echo "No data or checkpoint changes to commit"
-            exit 0
-          fi
-          git -C "$DATA_DIRECTORY" commit -m "Collect queue data through $CURSOR"
-          gh auth setup-git
-          git -C "$DATA_DIRECTORY" push origin "HEAD:refs/heads/$DATA_BRANCH"
+          python3 .github/scripts/pull-request-dashboard/state_branch.py commit-data \
+            --state-dir "$DATA_DIRECTORY" \
+            --state-branch "$DATA_BRANCH" \
+            --expected-sha "$EXPECTED_SHA" \
+            --message "Collect queue data through $CURSOR"
 
       - name: Prepare Pages site
         id: site
diff --git a/.github/workflows/github-actions-queue-test.yml b/.github/workflows/github-actions-queue-test.yml
index ea15ce79fb26..afe80e587b9b 100644
--- a/.github/workflows/github-actions-queue-test.yml
+++ b/.github/workflows/github-actions-queue-test.yml
@@ -6,6 +6,7 @@ on:
       - main
     paths:
       - .github/scripts/github-actions-queue/**
+      - .github/scripts/pull-request-dashboard/state_branch.py
       - .github/workflows/github-actions-queue-*.yml
       - github-actions-queue/**
   pull_request:
@@ -13,6 +14,7 @@ on:
       - main
     paths:
       - .github/scripts/github-actions-queue/**
+      - .github/scripts/pull-request-dashboard/state_branch.py
       - .github/workflows/github-actions-queue-*.yml
       - github-actions-queue/**
 
diff --git a/.github/workflows/pull-request-dashboard-drain.yml b/.github/workflows/pull-request-dashboard-drain.yml
index 37581e75dbf0..db34eb4697b9 100644
--- a/.github/workflows/pull-request-dashboard-drain.yml
+++ b/.github/workflows/pull-request-dashboard-drain.yml
@@ -31,6 +31,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 50
     env:
+      DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
       DISPATCHER_GENERATION: ${{ inputs.dispatcher_generation }}
       PR_DASHBOARD_QUEUE_LANE: ${{ inputs.queue_lane || 'live' }}
       PR_DASHBOARD_QUEUE_ENDPOINT: https://otel-pull-request-dashboard.netlify.app/.netlify/functions/dashboard-queue-worker
@@ -98,6 +99,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 50
     env:
+      DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
       DISPATCHER_GENERATION: ${{ inputs.dispatcher_generation }}
       PR_DASHBOARD_QUEUE_LANE: ${{ inputs.queue_lane || 'live' }}
       PR_DASHBOARD_QUEUE_ENDPOINT: https://otel-pull-request-dashboard.netlify.app/.netlify/functions/dashboard-queue-worker
diff --git a/.github/workflows/pull-request-dashboard-refresh-author-nudges.yml b/.github/workflows/pull-request-dashboard-refresh-author-nudges.yml
index 441174f39c57..4007f041e5ac 100644
--- a/.github/workflows/pull-request-dashboard-refresh-author-nudges.yml
+++ b/.github/workflows/pull-request-dashboard-refresh-author-nudges.yml
@@ -21,6 +21,7 @@ on:
 permissions: {}
 
 env:
+  DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
   DASHBOARD_CONFIG: .github/scripts/pull-request-dashboard/repositories.json
   DASHBOARD_STATE_BRANCH_PREFIX: otelbot/pull-request-dashboard-state
 
diff --git a/.github/workflows/pull-request-dashboard-repo.yml b/.github/workflows/pull-request-dashboard-repo.yml
index 006b1b010036..accc3530d51f 100644
--- a/.github/workflows/pull-request-dashboard-repo.yml
+++ b/.github/workflows/pull-request-dashboard-repo.yml
@@ -24,6 +24,7 @@ permissions:
   contents: read
 
 env:
+  DATA_BRANCH_SNAPSHOTS: ${{ vars.DATA_BRANCH_SNAPSHOTS }}
   DASHBOARD_CONFIG: .github/scripts/pull-request-dashboard/repositories.json
   DASHBOARD_STATE_BRANCH_PREFIX: otelbot/pull-request-dashboard-state
   DASHBOARD_DELIVERY_STATE_BRANCH_PREFIX: otelbot/pull-request-dashboard-delivery
diff --git a/github-actions-queue/README.md b/github-actions-queue/README.md
index a69c713193ed..348ad26f52a5 100644
--- a/github-actions-queue/README.md
+++ b/github-actions-queue/README.md
@@ -6,6 +6,14 @@ terminal GitHub Actions job in active public repositories under
 `otelbot/github-actions-queue-data` branch in this repository and can be
 queried to find queue-time outliers.
 
+The repository variable `DATA_BRANCH_SNAPSHOTS=true` retains only the latest
+Git snapshot of the data branch. All historical job measurements in the
+current files, collection checkpoints, and report inputs remain intact.
+The variable defaults to disabled and also controls PR dashboard state writers;
+follow the [shared cutover procedure](../.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md#7-data-branch-snapshot-cutover)
+before enabling it. The hourly collection and Pages deployment schedules do
+not change.
+
 Queue time is measured per job:
 
 ```text
diff --git a/pull-request-dashboard/README.md b/pull-request-dashboard/README.md
index fbc99bc517b1..50d5b5a60fb0 100644
--- a/pull-request-dashboard/README.md
+++ b/pull-request-dashboard/README.md
@@ -8,6 +8,23 @@ Webhook-triggered incremental runs keep active dashboards close to real time. Ho
 
 The classification cache reuses prior results for unchanged review threads, minimizing Copilot token usage.
 
+## Data branch storage
+
+The current dashboard state and delivery acknowledgements live on
+`otelbot/pull-request-dashboard-state/*` and
+`otelbot/pull-request-dashboard-delivery/*`. They must survive between runs,
+but their previous Git versions are not needed.
+
+The repository variable `DATA_BRANCH_SNAPSHOTS=true` makes data writers replace
+each branch with a single parentless commit containing its complete current
+files. It also controls the Actions queue collector. It defaults to disabled.
+Enable it only after promoting snapshot-capable dashboard code to the stable
+channel and completing the
+[controlled cutover](../.github/scripts/pull-request-dashboard/WEBHOOK_SETUP.md#7-data-branch-snapshot-cutover).
+Refresh schedules, notifications, and the stalled-run watchdog are unaffected.
+
+## Classification runtime
+
 Classification uses the Python Copilot SDK with a task-specific system prompt
 and no tools or repository instructions. Each dashboard process reuses one
 client, but every batch of up to ten items gets a fresh session. This removes