diff --git a/acceptance/bin/list_code_snapshot.py b/acceptance/bin/list_code_snapshot.py
deleted file mode 100755
index b8f1263cae7..00000000000
--- a/acceptance/bin/list_code_snapshot.py
+++ /dev/null
@@ -1,72 +0,0 @@
-#!/usr/bin/env python3
-"""
-List the entries of each AI Runtime code snapshot tarball uploaded during deploy.
-
-Reads out.requests.txt, takes every ai_runtime_task.code_source_path from the
-jobs/create request, exports each workspace archive via the CLI, and prints its
-sorted tar entries (grouped per archive). Used to assert which local files each
-snapshot includes (gitignore / sync rules), across one or more tasks.
-"""
-
-import gzip
-import io
-import os
-import subprocess
-import sys
-import tarfile
-
-from print_requests import read_json_many
-
-
-def code_source_paths(requests):
-    """Every task's code_source_path from the jobs/create request(s)."""
-    result = []
-    for req in requests:
-        body = req.get("body")
-        if isinstance(body, dict) and req.get("path", "").endswith("/jobs/create"):
-            for task in body.get("tasks", []):
-                art = task.get("ai_runtime_task")
-                if art and art.get("code_source_path"):
-                    result.append(art["code_source_path"])
-    return result
-
-
-def print_entries(cli, env, remote):
-    local = "code_snapshot.tar.gz"
-    subprocess.run(
-        [cli, "workspace", "export", remote, "--format", "AUTO", "--file", local],
-        check=True,
-        env=env,
-    )
-    with open(local, "rb") as f:
-        data = gzip.decompress(f.read())
-    os.remove(local)
-
-    # Print the archive's sync-relative name (hash tokenized by test.toml repls) so
-    # multi-archive output is legible.
-    print(f"# {remote.split('/files/', 1)[-1]}")
-    with tarfile.open(fileobj=io.BytesIO(data)) as tar:
-        for name in sorted(tar.getnames()):
-            print(name)
-
-
-def main():
-    with open("out.requests.txt") as f:
-        requests = read_json_many(f.read())
-
-    paths = code_source_paths(requests)
-    if not paths:
-        sys.exit("no jobs/create request with code_source_path in out.requests.txt")
-
-    cli = os.environ["CLI"]
-    # MSYS_NO_PATHCONV stops Git Bash on Windows from rewriting the /Workspace path.
-    env = {**os.environ, "MSYS_NO_PATHCONV": "1"}
-
-    # code_source_path is an absolute workspace path (/Workspace/Users/.../files/...).
-    # Sort so multi-task output is deterministic.
-    for remote in sorted(paths):
-        print_entries(cli, env, remote)
-
-
-if __name__ == "__main__":
-    main()
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt b/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt
index dbfdc962645..bab538a81e0 100644
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt
+++ b/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt
@@ -1,4 +1,5 @@
 
 >>> [CLI] bundle deploy
-Error: code_source_path "./src" has no files to package (all excluded by .gitignore or sync.exclude, or the directory is empty)
+Building air_code_source_src...
+Error: artifact tgz: no files to pack under "[TEST_TMP_DIR]" (empty, gitignored, or no `include` match)
 
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml b/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml
index 4a1cc0476c0..6f91fce9d54 100644
--- a/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml
+++ b/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml
@@ -1,20 +1,13 @@
 bundle:
   name: ai-runtime-test
 
-sync:
-  # *.log excluded; data/*.bin force-included despite .gitignore.
-  exclude:
-    - "**/*.log"
-  include:
-    - src/data/*.bin
-
 resources:
   jobs:
     train:
       name: "[${bundle.target}] AI Runtime training"
       tasks:
-        # Two AI Runtime tasks with distinct local code dirs: each is packaged into
-        # its own content-addressed tarball, both under the repo root's .air_snapshots.
+        # Two AI Runtime tasks with distinct local code dirs: each is turned into its
+        # own tgz artifact and uploaded through the standard artifact path.
         - task_key: train
           environment_key: default
           ai_runtime_task:
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/output.txt b/acceptance/bundle/ai_runtime_task/local_code_source/output.txt
index 8b01277b57b..476d4b1908c 100644
--- a/acceptance/bundle/ai_runtime_task/local_code_source/output.txt
+++ b/acceptance/bundle/ai_runtime_task/local_code_source/output.txt
@@ -1,129 +1,29 @@
 
-=== deploy packages and uploads the local code sources
+=== deploy packages the local code sources as tgz artifacts and uploads them
 
 >>> [CLI] bundle deploy
+Building air_code_source_src...
+Building air_code_source_src2...
+Uploading .databricks/air_code_source/air_code_source_src.tar.gz...
+Uploading .databricks/air_code_source/air_code_source_src2.tar.gz...
 Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files...
 Created jobs.train
-Files: 13 uploaded, 0 deleted
+Files: 9 uploaded, 0 deleted
 Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
 
-=== each task's tarball holds only synced files (both under the repo root .air_snapshots)
+=== both code_source_paths rewritten to the uploaded artifact remote paths
 
->>> list_code_snapshot.py
-# .air_snapshots/[SNAPSHOT].tar.gz
-src2/command.sh
-src2/train.py
-# .air_snapshots/[SNAPSHOT].tar.gz
-src/.gitignore
-src/command.sh
-src/data/model.bin
-src/train.py
+>>> jq -s .[] | select(.path=="/api/2.2/jobs/create") | .body.tasks[].ai_runtime_task.code_source_path out.requests.txt
+"/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/artifacts/.internal/air_code_source_src.tar.gz"
+"/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/artifacts/.internal/air_code_source_src2.tar.gz"
 
-=== both code_source_paths point into the bundle .air_snapshots, command_paths rewritten, deps on environments spec
+=== both code tarballs uploaded
 
->>> print_requests.py --nostamp --sort --del-field raw_body //.air_snapshots/ //jobs/create
-{
-  "method": "POST",
-  "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/.air_snapshots/[SNAPSHOT].tar.gz",
-  "q": {
-    "overwrite": "true"
-  }
-}
-{
-  "method": "POST",
-  "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/.air_snapshots/[SNAPSHOT].tar.gz",
-  "q": {
-    "overwrite": "true"
-  }
-}
-{
-  "method": "POST",
-  "path": "/api/2.2/jobs/create",
-  "body": {
-    "deployment": {
-      "kind": "BUNDLE",
-      "metadata_file_path": "/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/state/metadata.json"
-    },
-    "edit_mode": "UI_LOCKED",
-    "environments": [
-      {
-        "environment_key": "default",
-        "spec": {
-          "dependencies": [
-            "torch>=2.0.0"
-          ],
-          "environment_version": "5"
-        }
-      }
-    ],
-    "format": "MULTI_TASK",
-    "max_concurrent_runs": 1,
-    "name": "[default] AI Runtime training",
-    "queue": {
-      "enabled": true
-    },
-    "tasks": [
-      {
-        "ai_runtime_task": {
-          "code_source_path": "/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/.air_snapshots/[SNAPSHOT].tar.gz",
-          "deployments": [
-            {
-              "command_path": "/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/src/command.sh",
-              "compute": {
-                "accelerator_count": 8,
-                "accelerator_type": "GPU_8xH100"
-              }
-            }
-          ],
-          "experiment": "my-training"
-        },
-        "environment_key": "default",
-        "task_key": "train"
-      },
-      {
-        "ai_runtime_task": {
-          "code_source_path": "/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/.air_snapshots/[SNAPSHOT].tar.gz",
-          "deployments": [
-            {
-              "command_path": "/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/src2/command.sh",
-              "compute": {
-                "accelerator_count": 8,
-                "accelerator_type": "GPU_8xH100"
-              }
-            }
-          ],
-          "experiment": "my-training-2"
-        },
-        "environment_key": "default",
-        "task_key": "train2"
-      }
-    ]
-  }
-}
+>>> jq -r .path
+/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/artifacts/.internal/air_code_source_src.tar.gz
+/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/artifacts/.internal/air_code_source_src2.tar.gz
 
-=== re-planning unchanged code is a no-op (no changes)
-
->>> [CLI] bundle plan
-Plan: 0 to add, 0 to change, 0 to delete, 1 unchanged
-
-=== editing a file changes the snapshot hash (content-addressed name changes)
-
->>> [CLI] bundle deploy
-Uploading bundle files to /Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files...
-Updated jobs.train
-Files: 4 uploaded, 1 deleted
-Resources: 0 created, 1 changed, 0 deleted, 0 unchanged
-
->>> print_requests.py --nostamp --sort --del-field raw_body //.air_snapshots/
-{
-  "method": "POST",
-  "path": "/api/2.0/workspace-files/import-file/Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default/files/.air_snapshots/[SNAPSHOT].tar.gz",
-  "q": {
-    "overwrite": "true"
-  }
-}
-
-=== destroy removes the deployed bundle (including the synced snapshots)
+=== destroy removes the deployed bundle
 
 >>> [CLI] bundle destroy --auto-approve
 The following resources will be deleted:
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/script b/acceptance/bundle/ai_runtime_task/local_code_source/script
index 41949cf1c85..dccc1786add 100644
--- a/acceptance/bundle/ai_runtime_task/local_code_source/script
+++ b/acceptance/bundle/ai_runtime_task/local_code_source/script
@@ -1,29 +1,18 @@
-# Two AI Runtime tasks each with a local directory code_source_path. Each is
-# packaged into a content-addressed tarball written into the bundle
-# (.air_snapshots/, at the repo root) and uploaded by normal bundle file sync;
-# code_source_path/command_path are rewritten. Pip deps ride on the job's
-# environments[].spec.dependencies (no requirements.yaml is synthesized).
+# Two AI Runtime tasks with distinct local directory code sources. Each directory is
+# turned into a tgz artifact (see bundle/config/mutator/aicode), built and uploaded
+# through the standard artifact path, and its code_source_path is rewritten to the
+# uploaded remote path. Pip deps ride on the job's environments[].spec.dependencies.
 
-title "deploy packages and uploads the local code sources\n"
+title "deploy packages the local code sources as tgz artifacts and uploads them\n"
 trace $CLI bundle deploy
 
-title "each task's tarball holds only synced files (both under the repo root .air_snapshots)\n"
-trace list_code_snapshot.py
+title "both code_source_paths rewritten to the uploaded artifact remote paths\n"
+# Filters use a leading // so Git Bash on Windows does not path-convert them.
+trace jq -s '.[] | select(.path=="/api/2.2/jobs/create") | .body.tasks[].ai_runtime_task.code_source_path' out.requests.txt
 
-title "both code_source_paths point into the bundle .air_snapshots, command_paths rewritten, deps on environments spec\n"
-# --del-field raw_body drops the binary tarball upload payload (kept readable). Filters
-# use a leading // so Git Bash on Windows does not path-convert them. --keep is not
-# passed, so print_requests.py --nostamp consumes out.requests.txt.
-trace print_requests.py --nostamp --sort --del-field raw_body '//.air_snapshots/' '//jobs/create'
+title "both code tarballs uploaded\n"
+trace jq -r .path < out.requests.txt | grep import | grep '.tar.gz' | sort
 
-title "re-planning unchanged code is a no-op (no changes)\n"
-trace $CLI bundle plan
-
-title "editing a file changes the snapshot hash (content-addressed name changes)\n"
-update_file.py src/train.py 'print("training")' 'print("training v2")'
-trace $CLI bundle deploy
-trace print_requests.py --nostamp --sort --del-field raw_body '//.air_snapshots/'
-
-title "destroy removes the deployed bundle (including the synced snapshots)\n"
+title "destroy removes the deployed bundle\n"
 trace $CLI bundle destroy --auto-approve
 rm out.requests.txt
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/.gitignore b/acceptance/bundle/ai_runtime_task/local_code_source/src/.gitignore
deleted file mode 100644
index 54553edaac1..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/.gitignore
+++ /dev/null
@@ -1,2 +0,0 @@
-ignored_by_git.txt
-data/
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/data/model.bin b/acceptance/bundle/ai_runtime_task/local_code_source/src/data/model.bin
deleted file mode 100644
index 05424f2a4c8..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/data/model.bin
+++ /dev/null
@@ -1 +0,0 @@
-weights
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/data/notes.txt b/acceptance/bundle/ai_runtime_task/local_code_source/src/data/notes.txt
deleted file mode 100644
index fb188b9ecf0..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/data/notes.txt
+++ /dev/null
@@ -1 +0,0 @@
-scratch
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/debug.log b/acceptance/bundle/ai_runtime_task/local_code_source/src/debug.log
deleted file mode 100644
index 6bfe6b19e37..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/debug.log
+++ /dev/null
@@ -1 +0,0 @@
-log
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/ignored_by_git.txt b/acceptance/bundle/ai_runtime_task/local_code_source/src/ignored_by_git.txt
deleted file mode 100644
index bd930095363..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/ignored_by_git.txt
+++ /dev/null
@@ -1 +0,0 @@
-kept
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/test.toml b/acceptance/bundle/ai_runtime_task/local_code_source/test.toml
index 0aa3f208009..f99765d07e2 100644
--- a/acceptance/bundle/ai_runtime_task/local_code_source/test.toml
+++ b/acceptance/bundle/ai_runtime_task/local_code_source/test.toml
@@ -3,10 +3,3 @@ RecordRequests = true
 Ignore = [
     '.databricks',
 ]
-
-# The archive is content-addressed: <dir>_<sha256[:16]>.tar.gz. The hash is stable
-# given the committed inputs, but collapse it to a token so the test does not pin a
-# specific digest. Matches both src_ and src2_ archives.
-[[Repls]]
-Old = '(src2?)_[0-9a-f]{16}\.tar\.gz'
-New = '$1_[SNAPSHOT].tar.gz'
diff --git a/bundle/bundle.go b/bundle/bundle.go
index c58fba10d77..1eb9ce15654 100644
--- a/bundle/bundle.go
+++ b/bundle/bundle.go
@@ -59,11 +59,6 @@ func (b *Bundle) SuppressProgress() bool {
 	return b.Quiet >= QuietAll
 }
 
-// AiCodeSnapshotDir is the sync-relative dir the aicode mutator writes AI Runtime
-// code snapshots into. Force-included in sync (see GetSyncIncludePatterns) so user
-// ignore rules can't filter the deployed job's code_source_path archives out.
-const AiCodeSnapshotDir = ".air_snapshots"
-
 // Filename where resources are stored for DATABRICKS_BUNDLE_ENGINE=direct
 const resourcesFilename = "resources.json"
 
@@ -224,12 +219,6 @@ type Bundle struct {
 	// comparison with remote state, but local file validation would incorrectly fail.
 	SkipLocalFileValidation bool
 
-	// HasAiRuntimeCodeSnapshot is set by the aicode.PackageCodeSource build-phase
-	// mutator when it packages a local AI Runtime code_source into the bundle's
-	// snapshot dir. GetSyncIncludePatterns reads it to force-sync that dir only for
-	// bundles that actually use the feature, rather than for every bundle.
-	HasAiRuntimeCodeSnapshot bool
-
 	// Tagging is used to normalize tag keys and values.
 	// The implementation depends on the cloud being targeted.
 	Tagging tags.Cloud
@@ -419,13 +408,6 @@ func (b *Bundle) GetSyncIncludePatterns(ctx context.Context) ([]string, error) {
 		return nil, err
 	}
 	includes := append(b.Config.Sync.Include, filepath.ToSlash(filepath.Join(internalDirRel, "*.*")))
-	// Force-sync generated AI Runtime code snapshots so a user ignore rule (e.g.
-	// "*.tar.gz" in .gitignore) can't filter them out — the deployed job's
-	// code_source_path points at these archives (see bundle/config/mutator/aicode).
-	// Scoped to bundles that actually package one, so it's not a global include.
-	if b.HasAiRuntimeCodeSnapshot {
-		includes = append(includes, AiCodeSnapshotDir+"/*")
-	}
 	return includes, nil
 }
 
diff --git a/bundle/bundle_test.go b/bundle/bundle_test.go
index 09c3636758f..9bd667afe62 100644
--- a/bundle/bundle_test.go
+++ b/bundle/bundle_test.go
@@ -71,29 +71,6 @@ func TestBundleLocalStateDir(t *testing.T) {
 	assert.Equal(t, filepath.Join(projectDir, ".databricks", "bundle", "default"), cacheDir)
 }
 
-func TestGetSyncIncludePatternsScopesSnapshotDir(t *testing.T) {
-	ctx := t.Context()
-	projectDir := t.TempDir()
-	f, err := os.Create(filepath.Join(projectDir, "databricks.yml"))
-	require.NoError(t, err)
-	f.Close()
-
-	b, err := Load(ctx, projectDir)
-	require.NoError(t, err)
-	b.Config.Bundle.Target = "default"
-
-	// Without an AI Runtime code snapshot, the dir is not force-included.
-	includes, err := b.GetSyncIncludePatterns(ctx)
-	require.NoError(t, err)
-	assert.NotContains(t, includes, AiCodeSnapshotDir+"/*")
-
-	// Once the aicode mutator sets the flag, it is.
-	b.HasAiRuntimeCodeSnapshot = true
-	includes, err = b.GetSyncIncludePatterns(ctx)
-	require.NoError(t, err)
-	assert.Contains(t, includes, AiCodeSnapshotDir+"/*")
-}
-
 func TestBundleLocalStateDirOverride(t *testing.T) {
 	ctx := t.Context()
 	projectDir := t.TempDir()
diff --git a/bundle/config/mutator/aicode/package_code_source.go b/bundle/config/mutator/aicode/package_code_source.go
index b2f78a9aebd..57654dfd092 100644
--- a/bundle/config/mutator/aicode/package_code_source.go
+++ b/bundle/config/mutator/aicode/package_code_source.go
@@ -1,52 +1,47 @@
-// Package aicode packages a local directory referenced by an AI Runtime task's
-// code_source_path into a content-addressed tarball inside the bundle, and rewrites
-// code_source_path to the workspace path that tarball occupies once synced. Remote
-// values are left untouched.
+// Package aicode routes an AI Runtime task's local-directory code_source_path through
+// the standard artifact path: it synthesizes a `tgz` artifact that packages the
+// directory and rewrites code_source_path to the tarball the artifact builds. Remote
+// values and local files (a pre-built tarball from an explicit `artifacts` block) are
+// left untouched.
 //
-// The archive is overlaid on the sync tree and uploaded by normal bundle file sync
-// in the deploy phase; the mutator performs no workspace writes, so it is safe in
-// the build phase (which runs before `bundle plan`). Living in the bundle means
-// `bundle destroy` cleans it, and the content-addressed name lets incremental sync
-// skip re-uploading unchanged code.
-//
-// Not done via mutator.TranslatePaths (which handles command_path): that runs in
-// initialize, which also runs on `bundle validate`, so the archive would be
-// materialized during validate. Build phase is deploy-only.
+// It runs before artifacts.Prepare, so the synthesized artifact is prepared, built, and
+// uploaded by the normal artifact pipeline — there is no sync-root overlay. Because it
+// only edits config (no packaging or workspace writes), it is safe in the initialize
+// phase; the tarball itself is produced later by artifacts.Build.
 package aicode
 
 import (
-	"bytes"
 	"context"
 	"errors"
 	"fmt"
 	"io/fs"
+	"maps"
 	"os"
 	"path"
 	"path/filepath"
-	"slices"
 	"strings"
 
 	"github.com/databricks/cli/bundle"
-	"github.com/databricks/cli/bundle/deploy/files"
+	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/libraries"
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
-	"github.com/databricks/cli/libs/fileset"
 	"github.com/databricks/cli/libs/log"
-	libsync "github.com/databricks/cli/libs/sync"
-	"github.com/databricks/cli/libs/vfs"
 )
 
-// codeSourcePattern is the config location of an AI Runtime task's
-// code_source_path. It matches a direct task only — the same scope aicode.Validate
-// operates on. ai_runtime_task nested under a for_each_task is not a supported
-// combination yet (Validate rejects it); when it is, both should gain it together.
+// codeSourcePattern is the config location of an AI Runtime task's code_source_path. It
+// matches a direct task only — the same scope aicode.Validate operates on.
 var codeSourcePattern = dyn.NewPattern(
 	dyn.Key("resources"), dyn.Key("jobs"), dyn.AnyKey(),
 	dyn.Key("tasks"), dyn.AnyIndex(),
 	dyn.Key("ai_runtime_task"), dyn.Key("code_source_path"),
 )
 
+// codeArtifactOutputDir is where synthesized tgz artifacts write their tarball. It lives
+// under .databricks (transient, not synced) so the built file is uploaded once via the
+// artifact path and never swept into a sync or into the archive it produces.
+const codeArtifactOutputDir = ".databricks/air_code_source"
+
 // codeSource is a single local code_source_path occurrence to package.
 type codeSource struct {
 	configPath dyn.Path
@@ -67,144 +62,90 @@ func (m *packageCodeSource) Name() string {
 
 func (m *packageCodeSource) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
 	sources, diags := collectLocalCodeSources(b)
-	if diags.HasError() {
-		return diags
-	}
-	if len(sources) == 0 {
+	if diags.HasError() || len(sources) == 0 {
 		return diags
 	}
 
-	// remotePaths maps each config location to the synced workspace path its archive
-	// will occupy; overlayFiles maps each archive's sync-relative path to its bytes.
-	// Both are built before any config mutation so packaging failures are reported
-	// first. The archives are added to the sync root as in-memory overlay files
-	// (see below) rather than written to disk, so the user's working tree is not
-	// dirtied by deploy.
-	remotePaths := make(map[string]string, len(sources))
-	overlayFiles := make(map[string][]byte, len(sources))
+	// artifacts maps the synthesized artifact name to its spec; outputs maps each
+	// code_source_path config location to the tarball it should point at. The
+	// code_source_path and the artifact's `files` output share the same local path —
+	// that shared path is what links them when libraries upload rewrites both to the
+	// same remote location.
+	artifacts := make(map[string]*config.Artifact, len(sources))
+	outputs := make(map[string]string, len(sources))
+	// keyDir records which relDir each artifact key was derived from. artifactKey
+	// sanitizes non-alphanumerics to '_', so distinct directories ("a/b" and "a_b")
+	// can collide on one key — which would collapse them to a single tarball and make
+	// both tasks silently ship the same code. Detect that and error instead.
+	keyDir := make(map[string]string, len(sources))
 	for _, cs := range sources {
-		relArchive, archive, err := packageOne(ctx, b, cs)
-		if err != nil {
-			diags = diags.Extend(diag.FromErr(err))
-			return diags
+		relDir := strings.TrimPrefix(filepath.ToSlash(cs.value), "./")
+		key := artifactKey(relDir)
+		if prev, ok := keyDir[key]; ok && prev != relDir {
+			return diags.Extend(diag.Errorf("code_source directories %q and %q map to the same artifact name %q; rename one so they differ by more than a non-alphanumeric character", prev, relDir, key))
 		}
-		overlayFiles[relArchive] = archive
-		// The workspace path the archive occupies once file sync uploads it. Matches
-		// how command_path is translated (workspace.file_path + sync-relative path).
-		remotePaths[cs.configPath.String()] = path.Join(b.Config.Workspace.FilePath, relArchive)
-	}
-
-	// Overlay the archives onto the sync root: bundle file sync walks and uploads
-	// them like real files, but they never touch the user's working tree.
-	syncRoot, err := vfs.Overlay(b.SyncRoot, overlayFiles)
-	if err != nil {
-		return diags.Extend(diag.FromErr(err))
+		keyDir[key] = relDir
+		outRel := path.Join(codeArtifactOutputDir, key+".tar.gz")
+		// Paths are set absolute: a synthesized artifact carries no config location, so
+		// artifacts.Prepare cannot resolve relative paths against the bundle root for it.
+		// The runtime extracts to /databricks/code_source/<dir>, so entries must nest
+		// under the directory basename — hence path = the directory's parent and
+		// include = its basename, so the tgz builder names entries "<basename>/...".
+		artifacts[key] = &config.Artifact{
+			Type:    config.ArtifactTarball,
+			Path:    filepath.Join(b.SyncRootPath, filepath.FromSlash(path.Dir(relDir))),
+			Include: []string{path.Base(relDir)},
+			Files:   []config.ArtifactFile{{Source: filepath.Join(b.SyncRootPath, filepath.FromSlash(outRel))}},
+		}
+		// code_source_path resolves (via the sync root) to the same absolute output, so
+		// libraries upload links the two and rewrites both to the same remote path.
+		outputs[cs.configPath.String()] = "./" + outRel
+		log.Debugf(ctx, "synthesized tgz artifact %q for code_source_path %q", key, cs.value)
 	}
-	b.SyncRoot = syncRoot
 
-	// Signal GetSyncIncludePatterns to force-sync the snapshot dir for this bundle,
-	// so a user ignore rule can't filter the archives out of the upload set.
-	b.HasAiRuntimeCodeSnapshot = true
-
-	err = b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) {
+	// Rewrite code_source_path first (via the dynamic tree); the typed Artifacts set
+	// below then survives to the dynamic tree on mutator exit. Doing it in the other
+	// order would let Mutate's ToTyped pass drop the freshly-set artifacts.
+	err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) {
 		for _, cs := range sources {
-			remote := remotePaths[cs.configPath.String()]
+			out := outputs[cs.configPath.String()]
 			var err error
-			root, err = dyn.SetByPath(root, cs.configPath, dyn.NewValue(remote, []dyn.Location{cs.location}))
+			root, err = dyn.SetByPath(root, cs.configPath, dyn.NewValue(out, []dyn.Location{cs.location}))
 			if err != nil {
-				return root, fmt.Errorf("failed to update code_source_path %q to %q: %w", cs.value, remote, err)
+				return root, fmt.Errorf("failed to update code_source_path %q to %q: %w", cs.value, out, err)
 			}
 		}
 		return root, nil
 	})
 	if err != nil {
-		diags = diags.Extend(diag.FromErr(err))
-	}
-
-	return diags
-}
-
-// snapshotSubdir is the sync-relative dir the archives are placed under (dedicated
-// so a snapshot is never nested in the dir it snapshots). See bundle.AiCodeSnapshotDir.
-const snapshotSubdir = bundle.AiCodeSnapshotDir
-
-// packageOne packages the local directory for a single code source into a
-// reproducible, content-addressed tarball and returns its sync-relative path plus
-// the archive bytes. It performs no disk or workspace write: the caller overlays the
-// bytes onto the sync root and the deploy-phase file sync uploads them.
-func packageOne(ctx context.Context, b *bundle.Bundle, cs codeSource) (string, []byte, error) {
-	localDir := filepath.Join(b.SyncRootPath, filepath.FromSlash(cs.value))
-	dirName := filepath.Base(localDir)
-
-	// relBase is the code directory relative to the sync root, used both to scope the
-	// sync file list to this directory and to re-base archive entry names under it.
-	relBase, err := filepath.Rel(b.SyncRootPath, localDir)
-	if err != nil {
-		return "", nil, fmt.Errorf("code_source_path %q: %w", cs.value, err)
+		return diags.Extend(diag.FromErr(err))
 	}
-	relBase = filepath.ToSlash(relBase)
 
-	files, err := codeSourceFiles(ctx, b, relBase)
-	if err != nil {
-		return "", nil, fmt.Errorf("failed to list files for code_source_path %q: %w", cs.value, err)
-	}
-	// An empty file list means every file under the directory was filtered out
-	// (gitignore / sync.exclude) or the directory is empty. Packaging it would deploy
-	// a job with no code, so fail with an actionable message instead.
-	if len(files) == 0 {
-		return "", nil, fmt.Errorf("code_source_path %q has no files to package (all excluded by .gitignore or sync.exclude, or the directory is empty)", cs.value)
+	if b.Config.Artifacts == nil {
+		b.Config.Artifacts = make(map[string]*config.Artifact, len(artifacts))
 	}
+	maps.Copy(b.Config.Artifacts, artifacts)
 
-	// Build the archive in memory so its content hash can name the file; the hash is
-	// computed while gzipping, so this adds no extra pass over the files.
-	var buf bytes.Buffer
-	sha, err := buildCodeSnapshot(b.SyncRoot, relBase, files, dirName, &buf)
-	if err != nil {
-		return "", nil, fmt.Errorf("failed to package code_source_path %q: %w", cs.value, err)
-	}
-	// Content-addressed name + incremental file sync means an unchanged archive keeps
-	// the same synced path and is not re-uploaded.
-	relArchive := path.Join(snapshotSubdir, fmt.Sprintf("%s_%s.tar.gz", dirName, sha[:16]))
-	log.Debugf(ctx, "packaged code snapshot %s for code_source_path %q", relArchive, cs.value)
-	return relArchive, buf.Bytes(), nil
+	return diags
 }
 
-// codeSourceFiles returns the files under the code directory (relBase, relative to
-// the sync root) that should go into the snapshot. It reuses the bundle's sync
-// options so the file list is filtered exactly like bundle file sync: .gitignore
-// aware, plus the top-level sync.include/exclude globs. Scoping Paths to relBase
-// restricts the walk (and the returned relative paths) to the code directory.
-func codeSourceFiles(ctx context.Context, b *bundle.Bundle, relBase string) ([]fileset.File, error) {
-	opts, err := files.GetSyncOptions(ctx, b)
-	if err != nil {
-		return nil, err
-	}
-	// Scope the file list to the code directory (relBase) while keeping the
-	// bundle's include/exclude globs, so filtering matches bundle file sync.
-	fl, err := libsync.NewFileList(ctx, opts.WorktreeRoot, opts.LocalRoot, []string{relBase}, opts.Include, opts.Exclude)
-	if err != nil {
-		return nil, err
-	}
-	all, err := fl.Files(ctx)
-	if err != nil {
-		return nil, err
-	}
-
-	// sync.include is force-added regardless of the scoped walk, so the list can
-	// contain files outside the code directory. Keep only what is under relBase, or
-	// those strays make an all-filtered directory look non-empty and an empty archive
-	// ships.
-	if relBase == "." {
-		return all, nil
-	}
-	prefix := relBase + "/"
-	return slices.DeleteFunc(all, func(f fileset.File) bool {
-		return !strings.HasPrefix(f.Relative, prefix)
-	}), nil
+// artifactKey is a stable artifact name for a code directory (relative to the
+// bundle). Two tasks pointing at the same directory collapse to one artifact. The
+// sanitization is lossy, so distinct directories can collide on one key; the caller
+// guards against that (see keyDir in Apply).
+func artifactKey(relDir string) string {
+	safe := strings.Map(func(r rune) rune {
+		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') {
+			return r
+		}
+		return '_'
+	}, relDir)
+	return "air_code_source_" + safe
 }
 
-// collectLocalCodeSources returns every AI Runtime task code_source_path that
-// points at a local directory. Already-remote values are skipped.
+// collectLocalCodeSources returns every AI Runtime task code_source_path that points at
+// a local directory. Remote values and local files (handled by the artifact path) are
+// skipped.
 func collectLocalCodeSources(b *bundle.Bundle) ([]codeSource, diag.Diagnostics) {
 	var sources []codeSource
 	var diags diag.Diagnostics
@@ -218,10 +159,9 @@ func collectLocalCodeSources(b *bundle.Bundle) ([]codeSource, diag.Diagnostics)
 			if !libraries.IsLocalPath(value) {
 				return v, nil
 			}
-			// Only package a local *directory*. A local file (e.g. a pre-built
-			// tarball delivered via an `artifacts` block) is left alone so it flows
-			// through the standard artifact-upload path as a file. aicode.Validate
-			// applies the same directory check, so the two stay in agreement.
+			// Only package a local *directory*. A local file (e.g. a pre-built tarball
+			// delivered via an `artifacts` block) is left alone so it flows through the
+			// standard artifact-upload path as a file.
 			localDir := filepath.Join(b.SyncRootPath, filepath.FromSlash(value))
 			isDir, err := isExistingDir(localDir)
 			if err != nil {
@@ -245,10 +185,9 @@ func collectLocalCodeSources(b *bundle.Bundle) ([]codeSource, diag.Diagnostics)
 	return sources, diags
 }
 
-// isExistingDir reports whether path is an existing directory. A not-exist error
-// is not an error here (the path is simply not a directory this mutator packages),
-// but any other stat failure — notably a permission error on the parent — is
-// surfaced so it is not silently swallowed into "skip".
+// isExistingDir reports whether path is an existing directory. A not-exist error is not
+// an error here (the path is simply not a directory this mutator packages), but any
+// other stat failure — notably a permission error on the parent — is surfaced.
 func isExistingDir(path string) (bool, error) {
 	info, err := os.Stat(path)
 	if err != nil {
diff --git a/bundle/config/mutator/aicode/package_code_source_test.go b/bundle/config/mutator/aicode/package_code_source_test.go
index fca33a0a7bf..44c72bcfb3d 100644
--- a/bundle/config/mutator/aicode/package_code_source_test.go
+++ b/bundle/config/mutator/aicode/package_code_source_test.go
@@ -19,11 +19,10 @@ import (
 // bundleWithCodeSource builds a bundle rooted at dir whose single AI Runtime task
 // points at codeSourcePath.
 //
-// The end-to-end package/upload/rewrite behavior (local dir -> tarball -> upload ->
-// rewritten code_source_path) runs the full mutator pipeline (sync file list,
-// workspace filer) and is covered by acceptance tests under
-// acceptance/bundle/ai_runtime_task. This unit test covers only the pure
-// config-collection seam that does not touch the pipeline.
+// The end-to-end build/upload behavior (tarball built by artifacts.Build, uploaded by
+// libraries) runs the full pipeline and is covered by acceptance tests under
+// acceptance/bundle/ai_runtime_task. These unit tests cover the config-only seam:
+// which paths are collected, and the artifact synthesis + code_source_path rewrite.
 func bundleWithCodeSource(t *testing.T, dir, codeSourcePath string) *bundle.Bundle {
 	t.Helper()
 	b := &bundle.Bundle{
@@ -54,6 +53,63 @@ func bundleWithCodeSource(t *testing.T, dir, codeSourcePath string) *bundle.Bund
 	return b
 }
 
+// A local-directory code_source_path is turned into a tgz artifact and the field is
+// rewritten to the tarball that artifact builds. path/include are chosen so archive
+// entries nest under the directory basename (the runtime's code_source layout), and
+// the rewritten path equals the artifact's files.source so the upload rail links them.
+func TestPackageCodeSourceSynthesizesArtifact(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "src"), 0o700))
+	b := bundleWithCodeSource(t, dir, "./src")
+
+	diags := PackageCodeSource().Apply(t.Context(), b)
+	require.Empty(t, diags)
+
+	outRel := ".databricks/air_code_source/air_code_source_src.tar.gz"
+	a := b.Config.Artifacts["air_code_source_src"]
+	require.NotNil(t, a)
+	assert.Equal(t, config.ArtifactTarball, a.Type)
+	// path/files are absolute: a synthesized artifact has no location for Prepare to
+	// resolve relative paths against.
+	assert.Equal(t, dir, a.Path)
+	assert.Equal(t, []string{"src"}, a.Include)
+	require.Len(t, a.Files, 1)
+	assert.Equal(t, filepath.Join(dir, filepath.FromSlash(outRel)), a.Files[0].Source)
+
+	// code_source_path is rewritten to the bundle-relative output, which resolves to
+	// the same absolute file so the upload rail links them.
+	assert.Equal(t, "./"+outRel, b.Config.Resources.Jobs["train"].Tasks[0].AiRuntimeTask.CodeSourcePath)
+}
+
+// Two distinct code_source directories that sanitize to the same artifact key
+// ("a/b" and "a_b" both become air_code_source_a_b) are rejected rather than
+// silently collapsed into one tarball.
+func TestPackageCodeSourceErrorsOnKeyCollision(t *testing.T) {
+	dir := t.TempDir()
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "a", "b"), 0o700))
+	require.NoError(t, os.MkdirAll(filepath.Join(dir, "a_b"), 0o700))
+	b := &bundle.Bundle{
+		BundleRootPath: dir,
+		SyncRootPath:   dir,
+		Config: config.Root{
+			Bundle: config.Bundle{Target: "default"},
+			Resources: config.Resources{
+				Jobs: map[string]*resources.Job{
+					"train": {JobSettings: jobs.JobSettings{Tasks: []jobs.Task{
+						{TaskKey: "t1", AiRuntimeTask: &jobs.AiRuntimeTask{Experiment: "exp", CodeSourcePath: "./a/b"}},
+						{TaskKey: "t2", AiRuntimeTask: &jobs.AiRuntimeTask{Experiment: "exp", CodeSourcePath: "./a_b"}},
+					}}},
+				},
+			},
+		},
+	}
+	bundletest.SetLocation(b, ".", []dyn.Location{{File: filepath.Join(dir, "databricks.yml")}})
+
+	diags := PackageCodeSource().Apply(t.Context(), b)
+	require.True(t, diags.HasError())
+	assert.ErrorContains(t, diags.Error(), "map to the same artifact name")
+}
+
 func TestCollectLocalCodeSourcesFindsLocalDir(t *testing.T) {
 	dir := t.TempDir()
 	require.NoError(t, os.MkdirAll(filepath.Join(dir, "src"), 0o700))
diff --git a/bundle/config/mutator/aicode/snapshot_package.go b/bundle/config/mutator/aicode/snapshot_package.go
deleted file mode 100644
index af1d5d583e0..00000000000
--- a/bundle/config/mutator/aicode/snapshot_package.go
+++ /dev/null
@@ -1,125 +0,0 @@
-package aicode
-
-import (
-	"archive/tar"
-	"compress/gzip"
-	"crypto/sha256"
-	"encoding/hex"
-	"fmt"
-	"io"
-	"path"
-	"slices"
-	"strings"
-	"time"
-
-	"github.com/databricks/cli/libs/fileset"
-	"github.com/databricks/cli/libs/vfs"
-)
-
-// tarEpoch is a fixed modification time stamped on every tar entry so the archive
-// is content-addressed: identical file contents always produce identical bytes
-// (and therefore an identical SHA-256), regardless of file mtimes or when the
-// archive was built. This is what lets an unchanged code directory resolve to the
-// same uploaded filename across deploys and skip re-upload. The technique mirrors
-// bundle/deploy/snapshot/path.go (which does the same for the immutable-folder zip).
-// Reproducible per platform, not across them (see addFileToArchive).
-var tarEpoch = time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC)
-
-// appleDoublePrefix is the basename prefix of macOS AppleDouble metadata files.
-// The AIR CLI excludes these; we match it so macOS archives carry no extra entries.
-const appleDoublePrefix = "._"
-
-// buildCodeSnapshot writes a reproducible gzipped tarball of the given files to out
-// and returns its SHA-256 hex digest. syncRoot is the root the files' Relative paths
-// are against (the bundle sync root); relBase is the code directory relative to that
-// root; prefix is the archive's top-level directory name. Each file at
-// "<relBase>/<rest>" is written to the archive as "<prefix>/<rest>", so the archive
-// expands to <prefix>/... — matching the runtime's /databricks/code_source/<dir>
-// extraction contract.
-//
-// The file list is produced by the bundle's sync walker, so it honors .gitignore
-// (including nested files) and the top-level sync.include/exclude globs — the same
-// filtering as bundle file sync.
-func buildCodeSnapshot(syncRoot vfs.Path, relBase string, files []fileset.File, prefix string, out io.Writer) (string, error) {
-	// Sort by relative path so the archive byte stream (and thus its hash) does not
-	// depend on iteration order.
-	slices.SortFunc(files, func(a, b fileset.File) int {
-		return strings.Compare(a.Relative, b.Relative)
-	})
-
-	hash := sha256.New()
-	gzw := gzip.NewWriter(io.MultiWriter(out, hash))
-	tw := tar.NewWriter(gzw)
-
-	for _, f := range files {
-		if err := addFileToArchive(tw, syncRoot, relBase, f, prefix); err != nil {
-			return "", err
-		}
-	}
-
-	if err := tw.Close(); err != nil {
-		return "", err
-	}
-	if err := gzw.Close(); err != nil {
-		return "", err
-	}
-	return hex.EncodeToString(hash.Sum(nil)), nil
-}
-
-func addFileToArchive(tw *tar.Writer, syncRoot vfs.Path, relBase string, f fileset.File, prefix string) error {
-	// f.Relative is relative to syncRoot and slash-separated. Re-base it under the
-	// code directory so the entry nests under the archive prefix.
-	rel := f.Relative
-	if relBase != "." {
-		trimmed, ok := strings.CutPrefix(rel, relBase+"/")
-		if !ok {
-			// Not under the code dir; the sync file list is scoped to it, so this
-			// should not happen, but skip defensively rather than mis-place a file.
-			return nil
-		}
-		rel = trimmed
-	}
-
-	if strings.HasPrefix(path.Base(rel), appleDoublePrefix) {
-		return nil
-	}
-
-	rc, err := syncRoot.Open(f.Relative)
-	if err != nil {
-		return fmt.Errorf("open %s: %w", f.Relative, err)
-	}
-	defer rc.Close()
-
-	info, err := rc.Stat()
-	if err != nil {
-		return fmt.Errorf("stat %s: %w", f.Relative, err)
-	}
-
-	// Only regular files are archived. The walker never yields directories, and
-	// symlinks inside a code snapshot are out of scope.
-	if !info.Mode().IsRegular() {
-		return nil
-	}
-
-	// Preserve the owner execute bit so a bundled helper stays executable, and
-	// normalize the rest to a canonical mode. Windows has no execute bit, so files are
-	// archived 0644 there and the archive hash differs from a Unix-built one.
-	mode := int64(0o644)
-	if info.Mode().Perm()&0o100 != 0 {
-		mode = 0o755
-	}
-	hdr := &tar.Header{
-		Typeflag: tar.TypeReg,
-		Name:     path.Join(prefix, rel),
-		Size:     info.Size(),
-		Mode:     mode,
-		ModTime:  tarEpoch,
-	}
-	if err := tw.WriteHeader(hdr); err != nil {
-		return fmt.Errorf("tar header for %s: %w", rel, err)
-	}
-	if _, err := io.Copy(tw, rc); err != nil {
-		return fmt.Errorf("write %s: %w", rel, err)
-	}
-	return nil
-}
diff --git a/bundle/config/mutator/aicode/snapshot_package_test.go b/bundle/config/mutator/aicode/snapshot_package_test.go
deleted file mode 100644
index 5342924fd58..00000000000
--- a/bundle/config/mutator/aicode/snapshot_package_test.go
+++ /dev/null
@@ -1,165 +0,0 @@
-package aicode
-
-import (
-	"archive/tar"
-	"bytes"
-	"compress/gzip"
-	"io"
-	"os"
-	"path/filepath"
-	"runtime"
-	"testing"
-
-	"github.com/databricks/cli/libs/fileset"
-	"github.com/databricks/cli/libs/vfs"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// writeTree materializes files (relative slash path -> content) under a fresh
-// temp dir and returns a vfs.Path rooted at it plus the fileset for its contents.
-func writeTree(t *testing.T, files map[string]string) (vfs.Path, []fileset.File) {
-	t.Helper()
-	dir := t.TempDir()
-	for name, content := range files {
-		p := filepath.Join(dir, filepath.FromSlash(name))
-		require.NoError(t, os.MkdirAll(filepath.Dir(p), 0o755))
-		require.NoError(t, os.WriteFile(p, []byte(content), 0o644))
-	}
-	root := vfs.MustNew(dir)
-	fs, err := fileset.New(root).Files()
-	require.NoError(t, err)
-	return root, fs
-}
-
-// tarEntries reads a gzipped tarball and returns entry name -> content.
-func tarEntries(t *testing.T, b []byte) map[string]string {
-	t.Helper()
-	gzr, err := gzip.NewReader(bytes.NewReader(b))
-	require.NoError(t, err)
-	tr := tar.NewReader(gzr)
-	out := map[string]string{}
-	for {
-		hdr, err := tr.Next()
-		if err == io.EOF {
-			break
-		}
-		require.NoError(t, err)
-		content, err := io.ReadAll(tr)
-		require.NoError(t, err)
-		out[hdr.Name] = string(content)
-	}
-	return out
-}
-
-// tarModes reads a gzipped tarball and returns entry name -> permission bits.
-func tarModes(t *testing.T, b []byte) map[string]int64 {
-	t.Helper()
-	gzr, err := gzip.NewReader(bytes.NewReader(b))
-	require.NoError(t, err)
-	tr := tar.NewReader(gzr)
-	out := map[string]int64{}
-	for {
-		hdr, err := tr.Next()
-		if err == io.EOF {
-			break
-		}
-		require.NoError(t, err)
-		out[hdr.Name] = hdr.Mode & 0o777
-	}
-	return out
-}
-
-// An executable file keeps the execute bit (0o755) so a bundled helper the user
-// invokes still runs; a non-executable file is normalized to 0o644.
-func TestBuildCodeSnapshotPreservesExecuteBit(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		// Windows file modes don't carry a Unix execute bit, so there's nothing to
-		// preserve; files simply archive as 0o644. The bit only matters on the
-		// Unix hosts that run the deployed workload.
-		t.Skip("execute bit is a Unix concept; not represented on Windows")
-	}
-	dir := t.TempDir()
-	require.NoError(t, os.WriteFile(filepath.Join(dir, "run.sh"), []byte("#!/bin/sh\n"), 0o755))
-	require.NoError(t, os.WriteFile(filepath.Join(dir, "train.py"), []byte("x"), 0o644))
-	root := vfs.MustNew(dir)
-	files, err := fileset.New(root).Files()
-	require.NoError(t, err)
-
-	var buf bytes.Buffer
-	_, err = buildCodeSnapshot(root, ".", files, "code", &buf)
-	require.NoError(t, err)
-
-	modes := tarModes(t, buf.Bytes())
-	assert.Equal(t, int64(0o755), modes["code/run.sh"], "executable helper must keep its execute bit")
-	assert.Equal(t, int64(0o644), modes["code/train.py"], "non-executable file is normalized to 0o644")
-}
-
-func TestBuildCodeSnapshotPrefixesEntries(t *testing.T) {
-	root, files := writeTree(t, map[string]string{
-		"train.py":        "print('train')",
-		"pkg/util.py":     "x = 1",
-		"._resource_fork": "apple double",
-	})
-
-	var buf bytes.Buffer
-	sha, err := buildCodeSnapshot(root, ".", files, "mycode", &buf)
-	require.NoError(t, err)
-	require.NotEmpty(t, sha)
-
-	entries := tarEntries(t, buf.Bytes())
-	// Entries are prefixed with the code dir basename (runtime extracts to
-	// /databricks/code_source/<dir>).
-	assert.Equal(t, "print('train')", entries["mycode/train.py"])
-	assert.Equal(t, "x = 1", entries["mycode/pkg/util.py"])
-	assert.NotContains(t, entries, "mycode/._resource_fork", "AppleDouble metadata must be excluded")
-}
-
-func TestBuildCodeSnapshotRebasesUnderRelBase(t *testing.T) {
-	// Files listed relative to a sync root; only the "src" subtree is packaged and
-	// re-based so entries nest under the prefix (not the intermediate "src/").
-	root, files := writeTree(t, map[string]string{
-		"src/train.py":    "t",
-		"src/pkg/util.py": "u",
-		"other/ignore.py": "o",
-	})
-
-	var buf bytes.Buffer
-	_, err := buildCodeSnapshot(root, "src", files, "src", &buf)
-	require.NoError(t, err)
-
-	entries := tarEntries(t, buf.Bytes())
-	assert.Contains(t, entries, "src/train.py")
-	assert.Contains(t, entries, "src/pkg/util.py")
-	// A file outside relBase is not under "src/", so it is skipped.
-	assert.NotContains(t, entries, "src/other/ignore.py")
-	assert.NotContains(t, entries, "other/ignore.py")
-}
-
-func TestBuildCodeSnapshotIsReproducible(t *testing.T) {
-	files := map[string]string{"a.py": "aaa", "sub/b.py": "bbb"}
-	root1, fs1 := writeTree(t, files)
-	root2, fs2 := writeTree(t, files)
-
-	var buf1, buf2 bytes.Buffer
-	sha1, err := buildCodeSnapshot(root1, ".", fs1, "code", &buf1)
-	require.NoError(t, err)
-	sha2, err := buildCodeSnapshot(root2, ".", fs2, "code", &buf2)
-	require.NoError(t, err)
-
-	assert.Equal(t, sha1, sha2, "identical content must produce an identical hash")
-	assert.Equal(t, buf1.Bytes(), buf2.Bytes(), "identical content must produce identical bytes")
-}
-
-func TestBuildCodeSnapshotHashChangesWithContent(t *testing.T) {
-	root1, fs1 := writeTree(t, map[string]string{"main.py": "v1"})
-	root2, fs2 := writeTree(t, map[string]string{"main.py": "v2"})
-
-	var buf1, buf2 bytes.Buffer
-	sha1, err := buildCodeSnapshot(root1, ".", fs1, "code", &buf1)
-	require.NoError(t, err)
-	sha2, err := buildCodeSnapshot(root2, ".", fs2, "code", &buf2)
-	require.NoError(t, err)
-
-	assert.NotEqual(t, sha1, sha2)
-}
diff --git a/bundle/config/mutator/aicode/validate.go b/bundle/config/mutator/aicode/validate.go
index 492203bc9f4..69e9d6300f2 100644
--- a/bundle/config/mutator/aicode/validate.go
+++ b/bundle/config/mutator/aicode/validate.go
@@ -3,7 +3,6 @@ package aicode
 import (
 	"context"
 	"fmt"
-	"os"
 	"path/filepath"
 
 	"github.com/databricks/cli/bundle"
@@ -12,7 +11,6 @@ import (
 	"github.com/databricks/cli/libs/diag"
 	"github.com/databricks/cli/libs/dyn"
 	"github.com/databricks/databricks-sdk-go/service/jobs"
-	ignore "github.com/sabhiram/go-gitignore"
 )
 
 // Validate checks AI Runtime tasks that reference a local code_source_path so
@@ -33,20 +31,11 @@ func (v *validate) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 
 	jobsPath := dyn.NewPath(dyn.Key("resources"), dyn.Key("jobs"))
 
-	// packagesCode records whether any task actually has a code_source_path this
-	// mutator will package. The bundle-level guards below (which reject configs that
-	// would drop the generated snapshot from sync) only matter in that case, so
-	// they're gated on it to avoid spurious errors on unrelated bundles.
-	packagesCode := false
-
 	for name, job := range b.Config.Resources.Jobs {
 		jobPath := jobsPath.Append(dyn.Key(name))
 
 		for i, task := range job.Tasks {
 			taskPath := jobPath.Append(dyn.Key("tasks"), dyn.Index(i))
-			if task.AiRuntimeTask != nil && v.packagesLocalDir(b, task.AiRuntimeTask.CodeSourcePath) {
-				packagesCode = true
-			}
 
 			// A local code_source_path under a for_each_task is not packaged by this
 			// mutator (aicode collects only direct tasks). Reject it rather than let a
@@ -74,75 +63,9 @@ func (v *validate) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics
 		}
 	}
 
-	if packagesCode {
-		diags = diags.Extend(validateSnapshotDir(b))
-	}
-
-	return diags
-}
-
-// packagesLocalDir reports whether codeSourcePath is one this mutator packages: a
-// local path that is an existing directory. A local *file* (a pre-built tarball from
-// an `artifacts` block) is uploaded by the artifact path instead, so it packages no
-// snapshot and the snapshot-directory guards must not apply to it.
-func (v *validate) packagesLocalDir(b *bundle.Bundle, codeSourcePath string) bool {
-	if codeSourcePath == "" || !libraries.IsLocalPath(codeSourcePath) {
-		return false
-	}
-	// A stat error is reported by validateTask; treat it as "not a directory" here.
-	isDir, err := isExistingDir(filepath.Join(b.SyncRootPath, filepath.FromSlash(codeSourcePath)))
-	return err == nil && isDir
-}
-
-// validateSnapshotDir rejects two configs that would silently drop the generated
-// code archive from sync (leaving the job pointing at an un-uploaded path):
-//
-//   - A real file/dir at bundle.AiCodeSnapshotDir collides with the overlay (sync
-//     carries the user's entry, not the archive).
-//   - A sync.exclude matching that path removes the archive (exclude is applied
-//     after include, so it beats the force-include).
-func validateSnapshotDir(b *bundle.Bundle) diag.Diagnostics {
-	var diags diag.Diagnostics
-	syncExcludePath := dyn.NewPath(dyn.Key("sync"), dyn.Key("exclude"))
-
-	// A user-owned file or directory at the reserved path collides with the overlay.
-	local := filepath.Join(b.SyncRootPath, bundle.AiCodeSnapshotDir)
-	if _, err := os.Stat(local); err == nil {
-		diags = diags.Append(diag.Diagnostic{
-			Severity: diag.Error,
-			Summary:  fmt.Sprintf("%q is reserved for AI Runtime code snapshots and must not exist in the bundle", bundle.AiCodeSnapshotDir),
-			Detail:   "Remove it; the deploy generates code archives under this path.",
-		})
-	}
-
-	// A sync.exclude matching the reserved path would drop the generated archive from
-	// the upload (exclude wins over the force-include).
-	if matchesSnapshotDir(b.Config.Sync.Exclude) {
-		diags = diags.Append(diag.Diagnostic{
-			Severity:  diag.Error,
-			Summary:   fmt.Sprintf("sync.exclude must not match %q, which holds the AI Runtime code snapshot", bundle.AiCodeSnapshotDir),
-			Detail:    "Remove the pattern that excludes it; otherwise the deployed job's code_source_path would not be uploaded.",
-			Locations: b.Config.GetLocations(syncExcludePath.String()),
-			Paths:     []dyn.Path{syncExcludePath},
-		})
-	}
-
 	return diags
 }
 
-// matchesSnapshotDir reports whether any sync.exclude pattern would remove a file
-// under the reserved snapshot directory, using the same gitignore-style matcher the
-// sync engine applies to exclude patterns (see libs/fileset).
-func matchesSnapshotDir(exclude []string) bool {
-	if len(exclude) == 0 {
-		return false
-	}
-	matcher := ignore.CompileIgnoreLines(exclude...)
-	// A representative archive path; the mutator names archives
-	// <AiCodeSnapshotDir>/<dir>_<sha>.tar.gz.
-	return matcher.MatchesPath(bundle.AiCodeSnapshotDir + "/probe.tar.gz")
-}
-
 func (v *validate) validateTask(b *bundle.Bundle, gitSource *jobs.GitSource, codeSourcePath string, codePath dyn.Path) diag.Diagnostics {
 	// Only local code_source_path values are packaged at deploy; remote values
 	// are used as-is and need no validation here.
diff --git a/bundle/config/mutator/aicode/validate_test.go b/bundle/config/mutator/aicode/validate_test.go
index 8ab48060af1..1839e8c227d 100644
--- a/bundle/config/mutator/aicode/validate_test.go
+++ b/bundle/config/mutator/aicode/validate_test.go
@@ -136,46 +136,3 @@ func TestValidateForEachTaskCodeSourceRejected(t *testing.T) {
 	require.Len(t, diags, 1)
 	assert.Contains(t, diags[0].Summary, "for_each_task")
 }
-
-// A sync.exclude pattern matching the reserved snapshot dir would drop the
-// generated archive from the upload (exclude wins over include), so it is rejected.
-func TestValidateSyncExcludeMatchingSnapshotDir(t *testing.T) {
-	for _, pattern := range []string{".air_snapshots/*", "**/*.tar.gz", ".air_snapshots"} {
-		b := bundleForValidate(t, "src", nil)
-		mkCodeDir(t, b, "src")
-		b.Config.Sync.Exclude = []string{pattern}
-		diags := Validate().Apply(t.Context(), b)
-		require.Len(t, diags, 1, "pattern %q should be rejected", pattern)
-		assert.Contains(t, diags[0].Summary, "sync.exclude")
-	}
-}
-
-// An unrelated sync.exclude is fine — the guard only fires for patterns that would
-// filter the snapshot dir.
-func TestValidateSyncExcludeUnrelatedIsAllowed(t *testing.T) {
-	b := bundleForValidate(t, "src", nil)
-	mkCodeDir(t, b, "src")
-	b.Config.Sync.Exclude = []string{"*.log", "build/**"}
-	assert.Empty(t, Validate().Apply(t.Context(), b))
-}
-
-// A real file/dir at the reserved snapshot path collides with the overlay, so it is
-// rejected.
-func TestValidateReservedSnapshotDirCollision(t *testing.T) {
-	b := bundleForValidate(t, "src", nil)
-	mkCodeDir(t, b, "src")
-	require.NoError(t, os.MkdirAll(filepath.Join(b.SyncRootPath, ".air_snapshots"), 0o700))
-	diags := Validate().Apply(t.Context(), b)
-	require.Len(t, diags, 1)
-	assert.Contains(t, diags[0].Summary, "reserved")
-}
-
-// The snapshot-dir guards only fire when a task actually packages a local
-// code_source — an unrelated bundle with a stray .air_snapshots is not this
-// mutator's concern.
-func TestValidateSnapshotGuardsSkippedWithoutLocalCodeSource(t *testing.T) {
-	b := bundleForValidate(t, "/Volumes/main/default/code/x.tar.gz", nil)
-	require.NoError(t, os.MkdirAll(filepath.Join(b.SyncRootPath, ".air_snapshots"), 0o700))
-	b.Config.Sync.Exclude = []string{".air_snapshots/*"}
-	assert.Empty(t, Validate().Apply(t.Context(), b))
-}
diff --git a/bundle/phases/build.go b/bundle/phases/build.go
index 580a18f7ab6..db376e07e28 100644
--- a/bundle/phases/build.go
+++ b/bundle/phases/build.go
@@ -7,7 +7,6 @@ import (
 	"github.com/databricks/cli/bundle/artifacts"
 	"github.com/databricks/cli/bundle/config"
 	"github.com/databricks/cli/bundle/config/mutator"
-	"github.com/databricks/cli/bundle/config/mutator/aicode"
 	"github.com/databricks/cli/bundle/libraries"
 	"github.com/databricks/cli/bundle/scripts"
 	"github.com/databricks/cli/bundle/trampoline"
@@ -28,13 +27,6 @@ func Build(ctx context.Context, b *bundle.Bundle) LibLocationMap {
 		artifacts.Build(),
 		scripts.Execute(config.ScriptPostBuild),
 
-		// Package any AI Runtime task code_source_path that points at a local
-		// directory into a content-addressed tarball overlaid on the sync root, and
-		// rewrite the field to the synced workspace path. No requirements.yaml is
-		// synthesized: the runtime installs pip deps from the job's serverless
-		// environment (environments[].spec.dependencies) directly.
-		aicode.PackageCodeSource(),
-
 		mutator.ResolveVariableReferencesWithoutResources(
 			"artifacts",
 		),
diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go
index b15e1c30df6..5bbb1c2b829 100644
--- a/bundle/phases/initialize.go
+++ b/bundle/phases/initialize.go
@@ -219,6 +219,13 @@ func Initialize(ctx context.Context, b *bundle.Bundle) {
 		// Applies the artifacts_dynamic_version preset to enable dynamic versioning on all artifacts
 		mutator.ApplyArtifactsDynamicVersion(),
 
+		// Turn any AI Runtime task code_source_path that points at a local directory
+		// into a `tgz` artifact and rewrite the field to the tarball that artifact
+		// builds, so the code is packaged and uploaded through the standard artifact
+		// path. Runs before artifacts.Prepare so the synthesized artifact is prepared
+		// and built like any other. Remote values and local files are left untouched.
+		aicode.PackageCodeSource(),
+
 		// Reads (typed): b.Config.Artifacts, b.BundleRootPath (checks artifact configurations and bundle path)
 		// Updates (typed): b.Config.Artifacts (auto-creates Python wheel artifact if none defined but setup.py exists)
 		// Updates (dynamic): artifacts.*.{path,build_command,files.*.source} (sets default paths, build commands, and makes relative paths absolute)