diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/assets/model.bin b/acceptance/bundle/ai_runtime_task/empty_code_source/assets/model.bin
deleted file mode 100644
index e69de29bb2d..00000000000
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/databricks.yml b/acceptance/bundle/ai_runtime_task/empty_code_source/databricks.yml
deleted file mode 100644
index 562924ddc06..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/databricks.yml
+++ /dev/null
@@ -1,29 +0,0 @@
-bundle:
-  name: ai-runtime-empty
-
-# An unrelated force-include elsewhere in the bundle. sync.include is added to the
-# file list regardless of the scoped walk, so it must not make an all-filtered code
-# directory look non-empty.
-sync:
-  include:
-    - assets/*.bin
-
-resources:
-  jobs:
-    train:
-      name: "[${bundle.target}] AI Runtime training"
-      tasks:
-        - task_key: train
-          environment_key: default
-          ai_runtime_task:
-            experiment: my-training
-            code_source_path: ./src
-            deployments:
-              - command_path: src/command.sh
-                compute:
-                  accelerator_type: GPU_8xH100
-                  accelerator_count: 8
-      environments:
-        - environment_key: default
-          spec:
-            environment_version: "5"
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/out.test.toml b/acceptance/bundle/ai_runtime_task/empty_code_source/out.test.toml
deleted file mode 100644
index e1af1a235ad..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/out.test.toml
+++ /dev/null
@@ -1,3 +0,0 @@
-Cloud = false
-EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
-EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt b/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt
deleted file mode 100644
index bab538a81e0..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/output.txt
+++ /dev/null
@@ -1,5 +0,0 @@
-
->>> [CLI] bundle deploy
-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/empty_code_source/script b/acceptance/bundle/ai_runtime_task/empty_code_source/script
deleted file mode 100644
index 6c9858c4368..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/script
+++ /dev/null
@@ -1,5 +0,0 @@
-# When code_source_path resolves to a directory whose contents are all filtered out
-# (here: a src/.gitignore of "*"), there is nothing to package. Deploy must fail
-# with an actionable message rather than shipping an empty code archive.
-# A sync.include elsewhere in the bundle must not defeat that guard (see databricks.yml).
-musterr trace $CLI bundle deploy
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/src/.gitignore b/acceptance/bundle/ai_runtime_task/empty_code_source/src/.gitignore
deleted file mode 100644
index 72e8ffc0db8..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/src/.gitignore
+++ /dev/null
@@ -1 +0,0 @@
-*
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/src/command.sh b/acceptance/bundle/ai_runtime_task/empty_code_source/src/command.sh
deleted file mode 100644
index 7ce3949767c..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/src/command.sh
+++ /dev/null
@@ -1,2 +0,0 @@
-cd $CODE_SOURCE_PATH
-python train.py
diff --git a/acceptance/bundle/ai_runtime_task/empty_code_source/src/train.py b/acceptance/bundle/ai_runtime_task/empty_code_source/src/train.py
deleted file mode 100644
index 399ce99aa9f..00000000000
--- a/acceptance/bundle/ai_runtime_task/empty_code_source/src/train.py
+++ /dev/null
@@ -1 +0,0 @@
-print("x")
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml b/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml
deleted file mode 100644
index 6f91fce9d54..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/databricks.yml
+++ /dev/null
@@ -1,36 +0,0 @@
-bundle:
-  name: ai-runtime-test
-
-resources:
-  jobs:
-    train:
-      name: "[${bundle.target}] AI Runtime training"
-      tasks:
-        # 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:
-            experiment: my-training
-            code_source_path: ./src
-            deployments:
-              - command_path: src/command.sh
-                compute:
-                  accelerator_type: GPU_8xH100
-                  accelerator_count: 8
-        - task_key: train2
-          environment_key: default
-          ai_runtime_task:
-            experiment: my-training-2
-            code_source_path: ./src2
-            deployments:
-              - command_path: src2/command.sh
-                compute:
-                  accelerator_type: GPU_8xH100
-                  accelerator_count: 8
-      environments:
-        - environment_key: default
-          spec:
-            environment_version: "5"
-            dependencies:
-              - torch>=2.0.0
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/out.test.toml b/acceptance/bundle/ai_runtime_task/local_code_source/out.test.toml
deleted file mode 100644
index e1af1a235ad..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/out.test.toml
+++ /dev/null
@@ -1,3 +0,0 @@
-Cloud = false
-EnvMatrix.DATABRICKS_BUNDLE_ENGINE = ["terraform", "direct"]
-EnvMatrix.DMS = ["", "true"]
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/output.txt b/acceptance/bundle/ai_runtime_task/local_code_source/output.txt
deleted file mode 100644
index 476d4b1908c..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/output.txt
+++ /dev/null
@@ -1,34 +0,0 @@
-
-=== 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: 9 uploaded, 0 deleted
-Resources: 1 created, 0 changed, 0 deleted, 0 unchanged
-
-=== both code_source_paths rewritten to the uploaded artifact remote paths
-
->>> 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 tarballs uploaded
-
->>> 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
-
-=== destroy removes the deployed bundle
-
->>> [CLI] bundle destroy --auto-approve
-The following resources will be deleted:
-  delete resources.jobs.train
-
-All files and directories at the following location will be deleted: /Workspace/Users/[USERNAME]/.bundle/ai-runtime-test/default
-
-Destroy: 1 deleted
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/script b/acceptance/bundle/ai_runtime_task/local_code_source/script
deleted file mode 100644
index dccc1786add..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/script
+++ /dev/null
@@ -1,18 +0,0 @@
-# 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 the local code sources as tgz artifacts and uploads them\n"
-trace $CLI bundle deploy
-
-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 tarballs uploaded\n"
-trace jq -r .path < out.requests.txt | grep import | grep '.tar.gz' | sort
-
-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/command.sh b/acceptance/bundle/ai_runtime_task/local_code_source/src/command.sh
deleted file mode 100644
index 7ce3949767c..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/command.sh
+++ /dev/null
@@ -1,2 +0,0 @@
-cd $CODE_SOURCE_PATH
-python train.py
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src/train.py b/acceptance/bundle/ai_runtime_task/local_code_source/src/train.py
deleted file mode 100644
index d8b062e5dfe..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src/train.py
+++ /dev/null
@@ -1 +0,0 @@
-print("training")
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src2/command.sh b/acceptance/bundle/ai_runtime_task/local_code_source/src2/command.sh
deleted file mode 100644
index 7ce3949767c..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src2/command.sh
+++ /dev/null
@@ -1,2 +0,0 @@
-cd $CODE_SOURCE_PATH
-python train.py
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/src2/train.py b/acceptance/bundle/ai_runtime_task/local_code_source/src2/train.py
deleted file mode 100644
index 4ff3f662781..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/src2/train.py
+++ /dev/null
@@ -1 +0,0 @@
-print("train2")
diff --git a/acceptance/bundle/ai_runtime_task/local_code_source/test.toml b/acceptance/bundle/ai_runtime_task/local_code_source/test.toml
deleted file mode 100644
index f99765d07e2..00000000000
--- a/acceptance/bundle/ai_runtime_task/local_code_source/test.toml
+++ /dev/null
@@ -1,5 +0,0 @@
-RecordRequests = true
-
-Ignore = [
-    '.databricks',
-]
diff --git a/acceptance/experimental/air/convert-to-dabs/output.txt b/acceptance/experimental/air/convert-to-dabs/output.txt
index f7b48740045..fa4413ffc07 100644
--- a/acceptance/experimental/air/convert-to-dabs/output.txt
+++ b/acceptance/experimental/air/convert-to-dabs/output.txt
@@ -19,13 +19,21 @@ persistent job that is not garbage-collected. When you are done, remove the
 job and its uploaded files with:
   [CLI] bundle destroy
 
-=== emitted databricks.yml (code_source_path points at ./src; no code is copied)
+=== emitted databricks.yml (code_source packaged as a tgz artifact; no code is copied)
 >>> cat databricks.yml
 bundle:
   name: torchrun-a10-smoke-test
 sync:
   paths:
     - generated_artifacts
+artifacts:
+  code_source:
+    type: tgz
+    path: .
+    include:
+      - src
+    files:
+      - source: ./dist/code_source.tgz
 targets:
   dev:
     mode: development
@@ -45,7 +53,7 @@ resources:
                 compute:
                   accelerator_type: GPU_1xA10
                   accelerator_count: 1
-            code_source_path: ./src
+            code_source_path: ./dist/code_source.tgz
       environments:
         - environment_key: default
           spec:
diff --git a/acceptance/experimental/air/convert-to-dabs/script b/acceptance/experimental/air/convert-to-dabs/script
index 77b6e06938a..0998acb0724 100644
--- a/acceptance/experimental/air/convert-to-dabs/script
+++ b/acceptance/experimental/air/convert-to-dabs/script
@@ -1,7 +1,7 @@
 title "convert an AIR run YAML into a DABs bundle (in place, next to the source)"
 trace $CLI experimental air convert-to-dabs train.yaml
 
-title "emitted databricks.yml (code_source_path points at ./src; no code is copied)"
+title "emitted databricks.yml (code_source packaged as a tgz artifact; no code is copied)"
 trace cat databricks.yml
 
 title "the generated command.sh carries the run command"
diff --git a/bundle/config/mutator/aicode/package_code_source.go b/bundle/config/mutator/aicode/package_code_source.go
deleted file mode 100644
index 57654dfd092..00000000000
--- a/bundle/config/mutator/aicode/package_code_source.go
+++ /dev/null
@@ -1,200 +0,0 @@
-// 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.
-//
-// 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 (
-	"context"
-	"errors"
-	"fmt"
-	"io/fs"
-	"maps"
-	"os"
-	"path"
-	"path/filepath"
-	"strings"
-
-	"github.com/databricks/cli/bundle"
-	"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/log"
-)
-
-// 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
-	location   dyn.Location
-	// value is the raw code_source_path string as written in config.
-	value string
-}
-
-func PackageCodeSource() bundle.Mutator {
-	return &packageCodeSource{}
-}
-
-type packageCodeSource struct{}
-
-func (m *packageCodeSource) Name() string {
-	return "aicode.PackageCodeSource"
-}
-
-func (m *packageCodeSource) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
-	sources, diags := collectLocalCodeSources(b)
-	if diags.HasError() || len(sources) == 0 {
-		return diags
-	}
-
-	// 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 {
-		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))
-		}
-		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)
-	}
-
-	// 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 {
-			out := outputs[cs.configPath.String()]
-			var err error
-			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, out, err)
-			}
-		}
-		return root, nil
-	})
-	if err != nil {
-		return diags.Extend(diag.FromErr(err))
-	}
-
-	if b.Config.Artifacts == nil {
-		b.Config.Artifacts = make(map[string]*config.Artifact, len(artifacts))
-	}
-	maps.Copy(b.Config.Artifacts, artifacts)
-
-	return diags
-}
-
-// 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. 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
-
-	err := b.Config.Mutate(func(root dyn.Value) (dyn.Value, error) {
-		return dyn.MapByPattern(root, codeSourcePattern, func(p dyn.Path, v dyn.Value) (dyn.Value, error) {
-			value, ok := v.AsString()
-			if !ok {
-				return v, fmt.Errorf("expected string, got %s", v.Kind())
-			}
-			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.
-			localDir := filepath.Join(b.SyncRootPath, filepath.FromSlash(value))
-			isDir, err := isExistingDir(localDir)
-			if err != nil {
-				return v, fmt.Errorf("code_source_path %q: %w", value, err)
-			}
-			if !isDir {
-				return v, nil
-			}
-			sources = append(sources, codeSource{
-				configPath: p,
-				location:   v.Location(),
-				value:      value,
-			})
-			return v, nil
-		})
-	})
-	if err != nil {
-		diags = diags.Extend(diag.FromErr(err))
-	}
-
-	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.
-func isExistingDir(path string) (bool, error) {
-	info, err := os.Stat(path)
-	if err != nil {
-		if errors.Is(err, fs.ErrNotExist) {
-			return false, nil
-		}
-		return false, err
-	}
-	return info.IsDir(), nil
-}
diff --git a/bundle/config/mutator/aicode/package_code_source_test.go b/bundle/config/mutator/aicode/package_code_source_test.go
deleted file mode 100644
index 44c72bcfb3d..00000000000
--- a/bundle/config/mutator/aicode/package_code_source_test.go
+++ /dev/null
@@ -1,145 +0,0 @@
-package aicode
-
-import (
-	"os"
-	"path/filepath"
-	"testing"
-
-	"github.com/databricks/cli/bundle"
-	"github.com/databricks/cli/bundle/config"
-	"github.com/databricks/cli/bundle/config/resources"
-	"github.com/databricks/cli/bundle/internal/bundletest"
-	"github.com/databricks/cli/libs/dyn"
-	"github.com/databricks/databricks-sdk-go/service/iam"
-	"github.com/databricks/databricks-sdk-go/service/jobs"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-// bundleWithCodeSource builds a bundle rooted at dir whose single AI Runtime task
-// points at codeSourcePath.
-//
-// 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{
-		BundleRootPath: dir,
-		SyncRootPath:   dir,
-		Config: config.Root{
-			Bundle: config.Bundle{Target: "default"},
-			Workspace: config.Workspace{
-				CurrentUser: &config.User{User: &iam.User{UserName: "me@databricks.com"}},
-			},
-			Resources: config.Resources{
-				Jobs: map[string]*resources.Job{
-					"train": {
-						JobSettings: jobs.JobSettings{
-							Tasks: []jobs.Task{
-								{
-									TaskKey:       "train",
-									AiRuntimeTask: &jobs.AiRuntimeTask{Experiment: "exp", CodeSourcePath: codeSourcePath},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	bundletest.SetLocation(b, ".", []dyn.Location{{File: filepath.Join(dir, "databricks.yml")}})
-	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))
-	b := bundleWithCodeSource(t, dir, "./src")
-	sources, diags := collectLocalCodeSources(b)
-	require.Empty(t, diags)
-	require.Len(t, sources, 1)
-	assert.Equal(t, "./src", sources[0].value)
-}
-
-func TestCollectLocalCodeSourcesSkipsRemotePaths(t *testing.T) {
-	for _, remote := range []string{
-		"/Workspace/Users/me/code.tar.gz",
-		"/Volumes/main/default/code/existing.tar.gz",
-	} {
-		b := bundleWithCodeSource(t, t.TempDir(), remote)
-		sources, diags := collectLocalCodeSources(b)
-		require.Empty(t, diags)
-		assert.Empty(t, sources, "remote code_source_path %q must not be collected", remote)
-	}
-}
-
-// A local path that resolves to a file (not a directory) — e.g. a pre-built
-// tarball delivered via an `artifacts` block — is NOT collected: it flows through
-// the standard artifact-upload path as a file rather than being packaged here.
-func TestCollectLocalCodeSourcesSkipsLocalFile(t *testing.T) {
-	dir := t.TempDir()
-	require.NoError(t, os.WriteFile(filepath.Join(dir, "code.tgz"), []byte("x"), 0o600))
-	b := bundleWithCodeSource(t, dir, "code.tgz")
-	sources, diags := collectLocalCodeSources(b)
-	require.Empty(t, diags)
-	assert.Empty(t, sources, "a local tarball file must flow through artifact upload, not aicode packaging")
-}
diff --git a/bundle/config/mutator/aicode/validate.go b/bundle/config/mutator/aicode/validate.go
deleted file mode 100644
index 69e9d6300f2..00000000000
--- a/bundle/config/mutator/aicode/validate.go
+++ /dev/null
@@ -1,141 +0,0 @@
-package aicode
-
-import (
-	"context"
-	"fmt"
-	"path/filepath"
-
-	"github.com/databricks/cli/bundle"
-	"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/databricks-sdk-go/service/jobs"
-)
-
-// Validate checks AI Runtime tasks that reference a local code_source_path so
-// that misconfigurations surface at `bundle validate` time with an actionable
-// message, rather than as an obscure failure mid-deploy. It performs no uploads.
-func Validate() bundle.ReadOnlyMutator {
-	return &validate{}
-}
-
-type validate struct{ bundle.RO }
-
-func (v *validate) Name() string {
-	return "aicode.Validate"
-}
-
-func (v *validate) Apply(ctx context.Context, b *bundle.Bundle) diag.Diagnostics {
-	var diags diag.Diagnostics
-
-	jobsPath := dyn.NewPath(dyn.Key("resources"), dyn.Key("jobs"))
-
-	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))
-
-			// 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
-			// nested ai_runtime_task deploy an un-packaged local path.
-			if task.ForEachTask != nil && task.ForEachTask.Task.AiRuntimeTask != nil {
-				nestedCode := task.ForEachTask.Task.AiRuntimeTask.CodeSourcePath
-				if nestedCode != "" && libraries.IsLocalPath(nestedCode) {
-					p := taskPath.Append(dyn.Key("for_each_task"), dyn.Key("task"),
-						dyn.Key("ai_runtime_task"), dyn.Key("code_source_path"))
-					diags = diags.Append(diag.Diagnostic{
-						Severity:  diag.Error,
-						Summary:   "ai_runtime_task with a local code_source_path is not supported inside a for_each_task",
-						Detail:    "Set code_source_path to a workspace or volume path",
-						Locations: b.Config.GetLocations(p.String()),
-						Paths:     []dyn.Path{p},
-					})
-				}
-			}
-
-			if task.AiRuntimeTask == nil {
-				continue
-			}
-			codePath := taskPath.Append(dyn.Key("ai_runtime_task"), dyn.Key("code_source_path"))
-			diags = diags.Extend(v.validateTask(b, job.GitSource, task.AiRuntimeTask.CodeSourcePath, codePath))
-		}
-	}
-
-	return diags
-}
-
-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.
-	if codeSourcePath == "" || !libraries.IsLocalPath(codeSourcePath) {
-		return nil
-	}
-
-	locations := b.Config.GetLocations(codePath.String())
-	reject := func(summary, detail string) diag.Diagnostics {
-		return diag.Diagnostics{{
-			Severity:  diag.Error,
-			Summary:   summary,
-			Detail:    detail,
-			Locations: locations,
-			Paths:     []dyn.Path{codePath},
-		}}
-	}
-
-	// The packaged directory must live inside the bundle sync root: it is uploaded as
-	// part of the bundle, so a path escaping the root (e.g. "../shared") can't be
-	// synced. Reject it here with a clear message rather than letting it fail later as
-	// an opaque io/fs "invalid argument" when the file list is built.
-	if rel, err := filepath.Rel(b.SyncRootPath, filepath.Join(b.SyncRootPath, filepath.FromSlash(codeSourcePath))); err != nil || !filepath.IsLocal(rel) {
-		return reject(
-			fmt.Sprintf("code_source_path %q is outside the bundle root", codeSourcePath),
-			"code_source_path must point at a directory inside the bundle, or at a workspace or volume path",
-		)
-	}
-
-	// This mutator packages a local *directory*. A local path that is not an existing
-	// directory is left alone so it flows through the standard artifact-upload path:
-	// a pre-built tarball delivered via an `artifacts` block is produced during the
-	// build phase (so it does not exist yet at validate time) and is uploaded as a
-	// file, not packaged here. Only when the path is an existing directory do the
-	// packaging-specific constraints below apply. A stat failure other than not-exist
-	// (e.g. unreadable parent) is surfaced rather than silently skipped.
-	localDir := filepath.Join(b.SyncRootPath, filepath.FromSlash(codeSourcePath))
-	isDir, err := isExistingDir(localDir)
-	if err != nil {
-		return reject(fmt.Sprintf("failed to inspect code_source_path %q: %v", codeSourcePath, err), "")
-	}
-	if !isDir {
-		return nil
-	}
-
-	// The deploy engine retrieves task files from git when git_source is set, so
-	// packaging a local directory would be silently ignored. Reject the combination.
-	if gitSource != nil {
-		return reject(
-			"ai_runtime_task with a local code_source_path cannot be combined with git_source",
-			"Remove git_source or set code_source_path to a workspace or volume path",
-		)
-	}
-
-	// Immutable-folder deployments upload a single content-addressed snapshot and
-	// do not support the per-task code packaging this mutator performs.
-	if b.IsImmutableFolder() {
-		return reject("ai_runtime_task with a local code_source_path is not supported with experimental.immutable_folder", "")
-	}
-
-	// Source-linked deployment runs jobs against the source files in place and does
-	// not copy them to the workspace file path (files.Upload is a no-op). This
-	// mutator relies on file sync uploading the packaged snapshot, so the two are
-	// incompatible: reject rather than deploy a job pointing at an un-uploaded path.
-	if config.IsExplicitlyEnabled(b.Config.Presets.SourceLinkedDeployment) {
-		return reject(
-			"ai_runtime_task with a local code_source_path is not supported with source-linked deployment",
-			"Disable source-linked deployment, or set code_source_path to a workspace or volume path",
-		)
-	}
-
-	return nil
-}
diff --git a/bundle/config/mutator/aicode/validate_test.go b/bundle/config/mutator/aicode/validate_test.go
deleted file mode 100644
index 1839e8c227d..00000000000
--- a/bundle/config/mutator/aicode/validate_test.go
+++ /dev/null
@@ -1,138 +0,0 @@
-package aicode
-
-import (
-	"os"
-	"path/filepath"
-	"testing"
-
-	"github.com/databricks/cli/bundle"
-	"github.com/databricks/cli/bundle/config"
-	"github.com/databricks/cli/bundle/config/resources"
-	"github.com/databricks/cli/bundle/internal/bundletest"
-	"github.com/databricks/cli/libs/dyn"
-	"github.com/databricks/databricks-sdk-go/service/jobs"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
-)
-
-func bundleForValidate(t *testing.T, codeSourcePath string, gitSource *jobs.GitSource) *bundle.Bundle {
-	t.Helper()
-	dir := t.TempDir()
-	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{
-							GitSource: gitSource,
-							Tasks: []jobs.Task{
-								{
-									TaskKey:       "train",
-									AiRuntimeTask: &jobs.AiRuntimeTask{CodeSourcePath: codeSourcePath},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	bundletest.SetLocation(b, ".", []dyn.Location{{File: filepath.Join(dir, "databricks.yml")}})
-	return b
-}
-
-// mkCodeDir creates a code_source directory (with one file) under the bundle's
-// sync root, so the path resolves to an existing directory this mutator packages.
-func mkCodeDir(t *testing.T, b *bundle.Bundle, rel string) {
-	t.Helper()
-	dir := filepath.Join(b.SyncRootPath, rel)
-	require.NoError(t, os.MkdirAll(dir, 0o700))
-	require.NoError(t, os.WriteFile(filepath.Join(dir, "train.py"), []byte("print()\n"), 0o600))
-}
-
-// A local path that is not an existing directory is left alone: it flows through
-// the standard artifact-upload path (e.g. a pre-built tarball built by an
-// `artifacts` block, which does not exist yet at validate time).
-func TestValidateNonDirectoryCodeSourceIsSkipped(t *testing.T) {
-	// Missing path (nothing on disk yet).
-	b := bundleForValidate(t, "does-not-exist", nil)
-	assert.Empty(t, Validate().Apply(t.Context(), b))
-
-	// Existing local file (a pre-built tarball), not a directory.
-	b = bundleForValidate(t, "code.tgz", nil)
-	require.NoError(t, os.WriteFile(filepath.Join(b.SyncRootPath, "code.tgz"), []byte("x"), 0o600))
-	assert.Empty(t, Validate().Apply(t.Context(), b))
-}
-
-func TestValidateGitSourceConflict(t *testing.T) {
-	b := bundleForValidate(t, "src", &jobs.GitSource{GitUrl: "https://example.invalid/repo"})
-	mkCodeDir(t, b, "src")
-	diags := Validate().Apply(t.Context(), b)
-	require.Len(t, diags, 1)
-	assert.Contains(t, diags[0].Summary, "cannot be combined with git_source")
-}
-
-func TestValidateRemoteCodeSourceIsSkipped(t *testing.T) {
-	b := bundleForValidate(t, "/Volumes/main/default/code/x.tar.gz", nil)
-	diags := Validate().Apply(t.Context(), b)
-	assert.Empty(t, diags)
-}
-
-// A code_source_path escaping the bundle sync root is rejected with a clear
-// message (it can't be synced), rather than failing later as an opaque io/fs error.
-func TestValidateCodeSourceOutsideBundleRoot(t *testing.T) {
-	b := bundleForValidate(t, "../shared", nil)
-	diags := Validate().Apply(t.Context(), b)
-	require.Len(t, diags, 1)
-	assert.Contains(t, diags[0].Summary, "outside the bundle root")
-}
-
-// Source-linked deployment doesn't copy files to the workspace file path, so the
-// packaged snapshot would never be uploaded; the combination is rejected.
-func TestValidateSourceLinkedConflict(t *testing.T) {
-	b := bundleForValidate(t, "src", nil)
-	mkCodeDir(t, b, "src")
-	enabled := true
-	b.Config.Presets.SourceLinkedDeployment = &enabled
-	diags := Validate().Apply(t.Context(), b)
-	require.Len(t, diags, 1)
-	assert.Contains(t, diags[0].Summary, "source-linked deployment")
-}
-
-// A local code_source_path nested under a for_each_task is not packaged by the
-// mutator, so it is rejected rather than silently skipped.
-func TestValidateForEachTaskCodeSourceRejected(t *testing.T) {
-	dir := t.TempDir()
-	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: "fanout",
-									ForEachTask: &jobs.ForEachTask{
-										Task: jobs.Task{
-											AiRuntimeTask: &jobs.AiRuntimeTask{CodeSourcePath: "src"},
-										},
-									},
-								},
-							},
-						},
-					},
-				},
-			},
-		},
-	}
-	bundletest.SetLocation(b, ".", []dyn.Location{{File: filepath.Join(dir, "databricks.yml")}})
-	diags := Validate().Apply(t.Context(), b)
-	require.Len(t, diags, 1)
-	assert.Contains(t, diags[0].Summary, "for_each_task")
-}
diff --git a/bundle/phases/initialize.go b/bundle/phases/initialize.go
index 2942dd4d05e..7c18212846f 100644
--- a/bundle/phases/initialize.go
+++ b/bundle/phases/initialize.go
@@ -10,7 +10,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"
 	pythonmutator "github.com/databricks/cli/bundle/config/mutator/python"
 	"github.com/databricks/cli/bundle/config/validate"
 	"github.com/databricks/cli/bundle/deploy/metadata"
@@ -203,19 +202,6 @@ func Initialize(ctx context.Context, b *bundle.Bundle) {
 
 		mutator.TranslatePaths(),
 
-		// Reads (typed): resources.jobs.*.tasks[*].ai_runtime_task.code_source_path, job git_source
-		// Validates that AI Runtime tasks referencing a local code_source_path point at an existing
-		// directory and are not combined with git_source or immutable-folder deployments, so these
-		// misconfigurations are caught at validate time rather than mid-deploy.
-		aicode.Validate(),
-
-		// 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.Experimental.PythonWheelWrapper, b.Config.Presets.SourceLinkedDeployment (checks Python wheel wrapper and deployment mode settings)
 		// Reads (dynamic): resources.jobs.*.tasks (checks for tasks with local libraries and incompatible DBR versions)
 		// Provides warnings when Python wheel tasks are used with DBR < 13.3 or when wheel wrapper is incompatible with source-linked deployment
diff --git a/experimental/air/cmd/convert_to_dabs.go b/experimental/air/cmd/convert_to_dabs.go
index 461071fb126..06d2e0311ac 100644
--- a/experimental/air/cmd/convert_to_dabs.go
+++ b/experimental/air/cmd/convert_to_dabs.go
@@ -28,16 +28,12 @@ import (
 // env_vars.json / secret_env_vars.json / hyperparameters.yaml sidecars into
 // generated_artifacts/. It does NOT package, snapshot, or upload anything.
 //
-// code_source_path is emitted as the source directory relative to the bundle (the
-// bundle root defaults to the YAML's directory, which contains it). At deploy the
-// aicode mutator (bundle/config/mutator/aicode) packages that directory and uploads
-// it — so convert never touches the code. Dependencies are folded into the job's
-// environments[] spec, which the runtime installs from directly; no requirements.yaml
-// is emitted.
-//
-// When the snapshot pins a git ref or narrows to include_paths, convert instead emits
-// a `tgz` artifact (the DABs artifact snapshotter): DABs builds the tarball from that
-// ref/subset at deploy, and code_source_path points at the built tarball.
+// The code_source is emitted as a `tgz` artifact (the DABs artifact snapshotter):
+// DABs builds the tarball from the source directory at deploy and code_source_path
+// points at the built tarball — so convert never touches the code. A git ref or
+// include_paths just add `git` / a narrowed `include` to the same artifact.
+// Dependencies are folded into the job's environments[] spec, which the runtime
+// installs from directly; no requirements.yaml is emitted.
 
 // dabsTargetName is the single default target emitted; a development-mode target
 // is the conventional starting point for a generated bundle.
@@ -92,10 +88,9 @@ does not contact the workspace.`,
 		}
 
 		// Default the bundle to the input YAML's directory. The bundle's sync root
-		// must contain the code_source so `code_source_path` resolves within it (the
-		// deploy-time aicode mutator packages the source in place), and root_path is
-		// resolved relative to the YAML, so the YAML's dir is the natural bundle root.
-		// An explicit --output-dir overrides.
+		// must contain the code_source so the `tgz` artifact can package it, and
+		// root_path is resolved relative to the YAML, so the YAML's dir is the natural
+		// bundle root. An explicit --output-dir overrides.
 		dir := outputDir
 		if dir == "" {
 			dir = filepath.Dir(yamlPath)
@@ -118,7 +113,7 @@ does not contact the workspace.`,
 // mapping is unit-testable in isolation. Returns the bundle root as a
 // map[string]dyn.Value (ready for yamlsaver) and the loose artifacts (command.sh +
 // env/secret/param sidecars) to write under generated_artifacts/. It does not touch the
-// code_source; the deploy-time aicode mutator packages it in place.
+// code_source; the emitted `tgz` artifact packages it at deploy.
 func convertToDabs(ctx context.Context, cfg *runConfig, configPath, bundleDir string) (map[string]dyn.Value, []uploadItem, error) {
 	// idempotency_token is intentionally not mapped: it dedups a single runs/submit
 	// call, which has no analogue for a persistent, repeatedly-runnable bundle job.
@@ -140,9 +135,8 @@ func convertToDabs(ctx context.Context, cfg *runConfig, configPath, bundleDir st
 		}
 	}
 
-	// The source dir relative to the bundle. It becomes code_source_path directly in
-	// the plain case, or the `tgz` artifact's `path` when a git ref / include_paths is
-	// pinned (see codeArtifactFor).
+	// The source dir relative to the bundle, used as the `tgz` artifact's `path`
+	// (see codeArtifactFor); code_source_path points at the tarball the artifact builds.
 	codeDirPath, err := bundleCodeSourcePath(ctx, cfg, configPath, bundleDir)
 	if err != nil {
 		return nil, nil, err
@@ -189,11 +183,12 @@ type codeArtifact struct {
 	gitCommit *string
 }
 
-// codeArtifactFor returns the artifact to emit when the snapshot pins a git ref or
-// narrows to include_paths, else nil — the plain directory case, packaged by the
-// deploy-time aicode mutator. It errors when the code dir resolves to the bundle root
-// (no basename to nest under). codeDirPath is the source dir relative to the bundle
-// ("./"-prefixed).
+// codeArtifactFor returns the `tgz` artifact to emit for a code_source, or nil when
+// there is no code_source. Every code_source is packaged as a `tgz` artifact and built
+// and uploaded through the standard artifact path; a git ref / include_paths just add
+// `git` / narrowed `include` to the same artifact. It errors when the code dir resolves
+// to the bundle root (no basename to nest under). codeDirPath is the source dir relative
+// to the bundle ("./"-prefixed).
 //
 // The artifact snapshotter names entries relative to `path`, and the runtime extracts
 // to /databricks/code_source/<dir>, so the code dir's basename must be the top-level
@@ -201,7 +196,7 @@ type codeArtifact struct {
 // subpaths, so entries come out as "<basename>/..." — the layout the air CLI produced.
 func codeArtifactFor(cfg *runConfig, codeDirPath string) (*codeArtifact, error) {
 	snap := codeSnapshot(cfg)
-	if snap == nil || (snap.Git == nil && len(snap.IncludePaths) == 0) {
+	if snap == nil {
 		return nil, nil
 	}
 	codeDirRel := strings.TrimPrefix(codeDirPath, "./")
@@ -211,7 +206,7 @@ func codeArtifactFor(cfg *runConfig, codeDirPath string) (*codeArtifact, error)
 	// generated_artifacts/, the output tarball) into it. Reject rather than emit that;
 	// the user should point root_path at a subdirectory.
 	if codeDirRel == "." {
-		return nil, fmt.Errorf("code_source root_path %q resolves to the bundle root; convert-to-dabs cannot translate a git or include_paths snapshot there (no code directory to package under %s/<dir>). Point root_path at a subdirectory", snap.RootPath, runtimeCodeSourceRoot)
+		return nil, fmt.Errorf("code_source root_path %q resolves to the bundle root; convert-to-dabs cannot package code there (no code directory to nest under %s/<dir>). Point root_path at a subdirectory", snap.RootPath, runtimeCodeSourceRoot)
 	}
 	dirName := path.Base(codeDirRel)
 	art := &codeArtifact{
@@ -241,9 +236,9 @@ func codeSnapshot(cfg *runConfig) *snapshotSourceConfig {
 }
 
 // bundleCodeSourcePath resolves the code_source directory to a "./"-prefixed path
-// relative to the bundle dir, for emission as ai_runtime_task.code_source_path.
-// Returns "" when the config has no code_source. The path must be inside the bundle
-// (the deploy-time mutator packages it in place and only handles in-bundle dirs).
+// relative to the bundle dir, used as the `tgz` artifact's `path`. Returns "" when the
+// config has no code_source. The path must be inside the bundle (the artifact
+// snapshotter only packages in-bundle dirs).
 func bundleCodeSourcePath(ctx context.Context, cfg *runConfig, configPath, bundleDir string) (string, error) {
 	snap := codeSnapshot(cfg)
 	if snap == nil {
@@ -306,8 +301,8 @@ func buildBundleValue(ctx context.Context, cfg *runConfig, configPath, codeSourc
 	}
 	line := 3
 	if codeSourcePath != "" {
-		// The source dir relative to the bundle; the aicode mutator packages it at
-		// deploy and rewrites this field to the uploaded workspace path.
+		// Points at the `tgz` artifact's built tarball; deploy uploads it and rewrites
+		// this to the uploaded workspace path.
 		aiRuntimeTask["code_source_path"] = nv(codeSourcePath, line)
 		line++
 	}
@@ -335,10 +330,10 @@ func buildBundleValue(ctx context.Context, cfg *runConfig, configPath, codeSourc
 	}
 	task["ai_runtime_task"] = nv(aiRuntimeTask, taskLine)
 
-	// environments[]: version + the dependency set. The aicode.SynthesizeRequirements
-	// mutator regenerates requirements.yaml from this spec at deploy time, so the full
-	// dependency set (whether authored inline or in a requirements file) must live
-	// here — convert emits no requirements.yaml of its own. Resolve the version
+	// environments[]: version + the dependency set. The runtime installs deps from this
+	// spec directly, so the full dependency set (whether authored inline or in a
+	// requirements file) must live here — convert emits no requirements.yaml of its
+	// own. Resolve the version
 	// through the same path `air run` uses (config, else env override, else the
 	// default channel) so a config without an explicit version still pins the version
 	// the workload would have run with — not an empty spec.
@@ -378,8 +373,8 @@ func buildBundleValue(ctx context.Context, cfg *runConfig, configPath, codeSourc
 			"name": nv(name, 1),
 		}, 1),
 		// sync.paths replaces the default of syncing the whole bundle root. The code
-		// directory is omitted deliberately: deploy still packages it into the
-		// snapshot tarball, so syncing it too would upload the tree twice.
+		// directory is omitted deliberately: the `tgz` artifact packages it, so syncing
+		// it too would upload the tree twice.
 		"sync": nv(map[string]dyn.Value{
 			"paths": nv([]dyn.Value{nv(generatedArtifactsDir, 1)}, 1),
 		}, 2),
@@ -395,14 +390,14 @@ func buildBundleValue(ctx context.Context, cfg *runConfig, configPath, codeSourc
 			}, 1),
 		}, 5),
 	}
-	// The `tgz` artifact snapshotter, when the snapshot pins a git ref / include_paths.
+	// The `tgz` artifact that packages the code_source (nil only when there is none).
 	if art != nil {
 		rootValue["artifacts"] = nv(buildArtifactsValue(art), 3)
 	}
 	return rootValue
 }
 
-// buildArtifactsValue builds the `artifacts` block for a git/include snapshot: a
+// buildArtifactsValue builds the `artifacts` block for the code_source: a
 // single `tgz` artifact whose `path` is the code-source root, carrying the git ref
 // and/or include subpaths, and whose `files` output is the tarball code_source_path
 // points at.
@@ -437,10 +432,9 @@ func buildArtifactsValue(art *codeArtifact) map[string]dyn.Value {
 }
 
 // bundleEnvironmentDeps resolves the runtime version and the inline dependency
-// list to emit in the bundle's environments[] spec. The aicode mutator synthesizes
-// requirements.yaml from that spec at deploy, so the whole set must be here.
-// Dependencies are inline-only (a requirements-file path is rejected at config
-// load), so an unset list yields no dependencies.
+// list to emit in the bundle's environments[] spec. The runtime installs deps from
+// that spec, so the whole set must be here. Dependencies are inline-only (a
+// requirements-file path is rejected at config load), so an unset list yields none.
 func bundleEnvironmentDeps(ctx context.Context, cfg *runConfig) (version string, deps []string) {
 	cfgVersion, _ := cfg.runtimeVersion()
 	version = dlRuntimeImage(ctx, cfgVersion)
@@ -522,7 +516,7 @@ func buildPermissionsValue(perms []permission) dyn.Value {
 
 // writeBundle writes the bundle into dir: databricks.yml plus the loose launch
 // artifacts (command.sh + env/secret/param sidecars). It does not touch the code
-// source — the deploy-time aicode mutator packages it in place. Unless force is set
+// source — the emitted `tgz` artifact packages it at deploy. Unless force is set
 // it refuses to overwrite existing files, so a re-run can't silently clobber a
 // bundle the user has edited. Returns the relative paths written, for the
 // next-steps message.
diff --git a/experimental/air/cmd/convert_to_dabs_test.go b/experimental/air/cmd/convert_to_dabs_test.go
index d169538fb06..0a81ddef844 100644
--- a/experimental/air/cmd/convert_to_dabs_test.go
+++ b/experimental/air/cmd/convert_to_dabs_test.go
@@ -141,9 +141,17 @@ environment:
 	art := task + ".ai_runtime_task"
 	assert.Equal(t, name, get(t, root, art+".experiment").MustString())
 	assert.Equal(t, "run-42", get(t, root, art+".mlflow_run").MustString())
-	// code_source_path is the source dir relative to the bundle; the deploy-time
-	// aicode mutator packages it in place.
-	assert.Equal(t, "./src", get(t, root, art+".code_source_path").MustString())
+	// code_source is packaged as a `tgz` artifact; code_source_path points at the
+	// built tarball, and the artifact packs the code dir (path = its parent, include =
+	// its basename, so entries nest under "src/").
+	assert.Equal(t, "./dist/code_source.tgz", get(t, root, art+".code_source_path").MustString())
+	codeArt := "artifacts." + codeSourceArtifactKey
+	assert.Equal(t, "tgz", get(t, root, codeArt+".type").MustString())
+	assert.Equal(t, ".", get(t, root, codeArt+".path").MustString())
+	codeInc := get(t, root, codeArt+".include").MustSequence()
+	require.Len(t, codeInc, 1)
+	assert.Equal(t, "src", codeInc[0].MustString())
+	assert.Equal(t, "./dist/code_source.tgz", get(t, root, codeArt+".files[0].source").MustString())
 
 	dep := art + ".deployments[0]"
 	assert.Equal(t, "./"+generatedArtifactsDir+"/"+commandScriptName, get(t, root, dep+".command_path").MustString())
@@ -168,8 +176,8 @@ environment:
 // Optional fields are omitted rather than emitted empty: no code_source means no
 // code_source_path; unset retries/timeout means no wrapper fields.
 // sync.paths lists only the generated-artifacts dir. The code directory must be
-// absent: deploy packages it into the snapshot tarball, so syncing it as loose
-// files too would upload the whole tree a second time.
+// absent: the `tgz` artifact packages it, so syncing it as loose files too would
+// upload the whole tree a second time.
 func TestConvertToDabsSyncPathsExcludesCodeDir(t *testing.T) {
 	cfg := minimalConfig + `
 code_source:
@@ -189,9 +197,10 @@ code_source:
 	require.Len(t, paths, 1)
 	assert.Equal(t, generatedArtifactsDir, paths[0].MustString())
 
-	// The task still points at the code dir; only the sync set omits it.
+	// The task points at the built `tgz` artifact; the code dir is packaged through the
+	// artifact path, so the sync set omits it.
 	task := "resources.jobs." + loaded.ExperimentName + ".tasks[0].ai_runtime_task"
-	assert.Equal(t, "./src", get(t, root, task+".code_source_path").MustString())
+	assert.Equal(t, "./dist/code_source.tgz", get(t, root, task+".code_source_path").MustString())
 }
 
 func TestConvertToDabsOmitsUnsetFields(t *testing.T) {
@@ -282,9 +291,9 @@ parameters:
 	assert.False(t, has(root, art+".secrets"))
 }
 
-// code_source_path is emitted as the source dir relative to the bundle and no code
-// is copied — the deploy-time mutator packages it in place. writeBundle produces
-// only databricks.yml + launch artifacts, not a code_source copy.
+// code_source_path points at the built `tgz` artifact and no code is copied — the
+// artifact snapshotter packages the source dir at deploy. writeBundle produces only
+// databricks.yml + launch artifacts, not a code_source copy.
 func TestConvertToDabsDoesNotCopyCode(t *testing.T) {
 	dir := t.TempDir()
 	require.NoError(t, os.MkdirAll(filepath.Join(dir, "src"), 0o700))
@@ -301,11 +310,11 @@ func TestConvertToDabsDoesNotCopyCode(t *testing.T) {
 	written, err := writeBundle(t.Context(), loaded, path, dir, false)
 	require.NoError(t, err)
 
-	// code_source_path points at the existing source dir; nothing is copied.
+	// code_source_path points at the built tarball; nothing is copied.
 	root, _, err := convertToDabs(t.Context(), loaded, path, dir)
 	require.NoError(t, err)
 	art := "resources.jobs." + loaded.ExperimentName + ".tasks[0].ai_runtime_task"
-	assert.Equal(t, "./src", get(t, root, art+".code_source_path").MustString())
+	assert.Equal(t, "./dist/code_source.tgz", get(t, root, art+".code_source_path").MustString())
 	assert.NotContains(t, written, "code_source/")
 }
 
@@ -360,8 +369,8 @@ func TestConvertToDabsGitPinEmitsArtifact(t *testing.T) {
 }
 
 // A requirements-FILE dependency set (environment.dependencies is a path) is folded
-// into the environments[] spec so the deploy-time aicode mutator can regenerate
-// requirements.yaml from it. Convert emits no requirements.yaml artifact of its own.
+// into the environments[] spec, which the runtime installs deps from directly.
+// Convert emits no requirements.yaml artifact of its own.
 // usage_policy_id is a resolved budget policy id and maps to the job's
 // budget_policy_id (usage_policy_name, which needs resolution, is rejected).
 func TestConvertToDabsMapsUsagePolicyID(t *testing.T) {