feat(decision): add opt-in --vcs-bundle option to decision task

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Change Log
 
+## [Unreleased]
+
+### Added
+
+- `taskgraph decision --vcs-bundle` writes a native git/hg bundle of the checkout to `public/vcs.bundle`, letting downstream tasks seed a checkout from it instead of cloning from the remote
+
 ## [24.1.1] - 2026-06-11
 
 ### Fixed

src/taskgraph/decision.py

@@ -98,6 +98,16 @@ def taskgraph_decision(options, parameters=None):
         opt_handler.setFormatter(logging.root.handlers[0].formatter)
     opt_log.addHandler(opt_handler)
 
+    # If requested, write a native VCS bundle of the checkout as a public
+    # artifact so downstream tasks can seed a checkout from it instead of
+    # cloning from the remote.
+    if options.get("vcs_bundle"):
+        repo = get_repository(os.getcwd())
+        ARTIFACTS_DIR.mkdir(parents=True, exist_ok=True)
+        bundle_path = ARTIFACTS_DIR / "vcs.bundle"
+        logger.info(f"creating vcs bundle at {bundle_path}")
+        repo.create_bundle(bundle_path)
+
     parameters = parameters or (
         lambda graph_config: get_decision_parameters(graph_config, options)
     )

src/taskgraph/main.py

@@ -941,6 +941,13 @@ def load_task(args):
     action="store_true",
     help="Allow user to override computed decision task parameters.",
 )
+@argument(
+    "--vcs-bundle",
+    action="store_true",
+    default=False,
+    help="Write a native git/hg bundle of the checkout to public/vcs.bundle, so "
+    "downstream tasks can seed a checkout from it instead of cloning.",
+)
 def decision(options):
     from taskgraph.decision import taskgraph_decision  # noqa: PLC0415
 

src/taskgraph/util/vcs.py

@@ -215,6 +215,14 @@ def does_revision_exist_locally(self, revision: str) -> bool:
         If this function returns an unexpected value, then make sure
         the revision was fetched from the remote repository."""
 
+    @abstractmethod
+    def create_bundle(self, dest_path) -> None:
+        """Write a native VCS bundle of the local checkout to ``dest_path``.
+
+        The bundle captures the full history and all refs so a downstream
+        task can seed a checkout from it (e.g. ``git clone``/``hg clone``)
+        instead of cloning from the remote."""
+
     def get_note(
         self,
         note: str,
@@ -384,6 +392,11 @@ def does_revision_exist_locally(self, revision):
                 return False
             raise
 
+    def create_bundle(self, dest_path):
+        # `--all` bundles every changeset (full history) so a downstream
+        # `hg clone <dest_path>` can seed a checkout offline.
+        self.run("bundle", "--all", str(dest_path))
+
 
 class GitRepository(Repository):
     @property
@@ -602,6 +615,16 @@ def does_revision_exist_locally(self, revision):
                 return False
             raise
 
+    def create_bundle(self, dest_path):
+        if self.is_shallow:
+            raise RuntimeError(
+                "Cannot create a git bundle from a shallow clone; a full "
+                "checkout is required to capture the complete history."
+            )
+        # `--all` includes every ref (and HEAD) so a downstream
+        # `git clone <dest_path>` can resolve a branch and seed offline.
+        self.run("bundle", "create", str(dest_path), "--all")
+
     def get_note(
         self,
         note: str,

test/test_decision.py

@@ -198,3 +198,37 @@ def test_decision_parameters_note_invalid_json(mock_files_changed, mock_get_note
         Exception, match="Failed to parse refs/notes/decision-parameters as JSON"
     ):
         decision.get_decision_parameters(FAKE_GRAPH_CONFIG, options)
+
+
+class _StopDecision(Exception):
+    """Sentinel raised by the mocked generator to halt taskgraph_decision early."""
+
+
+@pytest.mark.parametrize(
+    "vcs_bundle, expect_bundle",
+    [
+        pytest.param(True, True, id="flag-set"),
+        pytest.param(False, False, id="flag-absent"),
+    ],
+)
+@unittest.mock.patch.object(decision, "TaskGraphGenerator", side_effect=_StopDecision)
+@unittest.mock.patch.object(decision, "get_repository")
+def test_taskgraph_decision_vcs_bundle(
+    mock_get_repository, mock_tgg, vcs_bundle, expect_bundle, monkeypatch, tmp_path
+):
+    monkeypatch.setenv("TASK_ID", "decision-task-id")
+    monkeypatch.setattr(decision, "ARTIFACTS_DIR", tmp_path / "artifacts")
+    mock_repo = mock_get_repository.return_value
+
+    # The (mocked) generator raises, halting the decision right after the point
+    # where the VCS bundle would have been created. We only care about whether
+    # the bundle was written beforehand, not the rest of graph generation.
+    with pytest.raises(_StopDecision):
+        decision.taskgraph_decision({"vcs_bundle": vcs_bundle}, parameters=object())
+
+    if expect_bundle:
+        mock_repo.create_bundle.assert_called_once_with(
+            tmp_path / "artifacts" / "vcs.bundle"
+        )
+    else:
+        mock_repo.create_bundle.assert_not_called()

test/test_util_vcs.py

@@ -515,6 +515,22 @@ def test_does_revision_exist_locally(repo):
     assert not repo.does_revision_exist_locally("deadbeef")
 
 
+def test_create_bundle(repo, tmp_path):
+    dest = tmp_path / "vcs.bundle"
+    repo.create_bundle(dest)
+
+    assert dest.exists()
+    assert dest.stat().st_size > 0
+
+    # Round-trip: a fresh clone seeded from the bundle must land on the same
+    # head revision, proving the bundle carries the full history and refs.
+    out = tmp_path / "out"
+    subprocess.check_call([repo.tool, "clone", str(dest), str(out)])
+
+    cloned = get_repository(str(out))
+    assert cloned.head_rev == repo.head_rev
+
+
 def test_get_changed_files_shallow_clone(git_repo, tmp_path, default_git_branch):
     tmp_repo = Path(git_repo)