diff --git a/.nextchanges/air/code-source-flattening.md b/.nextchanges/air/code-source-flattening.md
new file mode 100644
index 00000000000..4bb18e32315
--- /dev/null
+++ b/.nextchanges/air/code-source-flattening.md
@@ -0,0 +1 @@
+* Support snapshot fields directly under `code_source` without requiring `type` or a nested `snapshot` block. ([#6927](https://github.com/databricks/cli/pull/6927))
diff --git a/acceptance/air/config-help/output.txt b/acceptance/air/config-help/output.txt
index c0949f97a43..abc1126bb54 100644
--- a/acceptance/air/config-help/output.txt
+++ b/acceptance/air/config-help/output.txt
@@ -114,29 +114,34 @@ config.compute.accelerator_type
   Type:     string
   Required: no
 
-=== conditionally required leaf
->>> [CLI] air run -h config.code_source.type
-config.code_source.type
-  Kind of code source. Must be "snapshot", the only supported type.
+=== code source fields
+>>> [CLI] air run -h config.code_source
+config.code_source
+  Local code to upload and make available to the run.
 
-  Type:     string
-  Required: when code_source is set
+  Fields:
+    root_path      Root of the code source to archive.
+    remote_volume  Volume to upload the archive to.
+    git            Pin the snapshot to a specific git revision.
+    include_paths  Restrict the archive to these paths, relative to root_path and without "..".
 
-=== union field reports both accepted shapes
->>> [CLI] air run -h config.code_source.snapshot.git.remote
-config.code_source.snapshot.git.remote
+Use "-h config.code_source.<field>" for details on a field.
+
+=== union field
+>>> [CLI] air run -h config.code_source.git.remote
+config.code_source.git.remote
   No longer supported: the snapshot archives your local copy. Only false is accepted; use commit to pin a revision.
 
   Type:     bool or string
   Required: no
 
-=== git-pinned subdirectory snapshots document root_path scoping
->>> [CLI] air run -h config.code_source.snapshot.root_path
-config.code_source.snapshot.root_path
+=== git-pinned subdirectory documents root_path scoping
+>>> [CLI] air run -h config.code_source.root_path
+config.code_source.root_path
   Root of the code source to archive. A git-pinned subdirectory packages only that subtree.
 
   Type:     string
-  Required: when code_source.snapshot is set
+  Required: when code_source is set
 
 === the config. prefix is optional
 >>> [CLI] air run -h compute.num_accelerators
diff --git a/acceptance/air/config-help/script b/acceptance/air/config-help/script
index e38fb1dd5e9..4f76006eeec 100644
--- a/acceptance/air/config-help/script
+++ b/acceptance/air/config-help/script
@@ -22,14 +22,14 @@ trace $CLI air run -h config.environment.version
 title "leaf field"
 trace $CLI air run -h config.compute.accelerator_type
 
-title "conditionally required leaf"
-trace $CLI air run -h config.code_source.type
+title "code source fields"
+trace $CLI air run -h config.code_source
 
-title "union field reports both accepted shapes"
-trace $CLI air run -h config.code_source.snapshot.git.remote
+title "union field"
+trace $CLI air run -h config.code_source.git.remote
 
-title "git-pinned subdirectory snapshots document root_path scoping"
-trace $CLI air run -h config.code_source.snapshot.root_path
+title "git-pinned subdirectory documents root_path scoping"
+trace $CLI air run -h config.code_source.root_path
 
 title "the config. prefix is optional"
 trace $CLI air run -h compute.num_accelerators
diff --git a/cmd/air/convert_to_dabs.go b/cmd/air/convert_to_dabs.go
index dd762446786..13421271367 100644
--- a/cmd/air/convert_to_dabs.go
+++ b/cmd/air/convert_to_dabs.go
@@ -127,7 +127,7 @@ func convertToDabs(ctx context.Context, cfg *runConfig, configPath, bundleDir st
 		// artifact location is set bundle-wide via workspace.artifact_path, not
 		// per-code-source, so a per-source Volume isn't representable here.
 		if snap.RemoteVolume != nil {
-			return nil, nil, errors.New("code_source.snapshot.remote_volume is not supported by convert-to-dabs; set workspace.artifact_path in the bundle instead")
+			return nil, nil, errors.New("remote_volume in code_source is not supported by convert-to-dabs; set workspace.artifact_path in the bundle instead")
 		}
 	}
 
diff --git a/cmd/air/convert_to_dabs_test.go b/cmd/air/convert_to_dabs_test.go
index 6e209224d6c..bd786a0db4e 100644
--- a/cmd/air/convert_to_dabs_test.go
+++ b/cmd/air/convert_to_dabs_test.go
@@ -318,7 +318,7 @@ code_source:
 	loaded, err := loadRunConfig(path)
 	require.NoError(t, err)
 	_, _, err = convertToDabs(t.Context(), loaded, path, filepath.Dir(path))
-	require.ErrorContains(t, err, "remote_volume is not supported")
+	require.ErrorContains(t, err, "remote_volume in code_source is not supported")
 }
 
 // env_variables / secrets / parameters ride as sidecar files (the ai_runtime_task
diff --git a/cmd/air/runconfig.go b/cmd/air/runconfig.go
index b466f2784f0..8ad651d4cbc 100644
--- a/cmd/air/runconfig.go
+++ b/cmd/air/runconfig.go
@@ -325,52 +325,72 @@ func (s *stringOrInt) UnmarshalYAML(node *yaml.Node) error {
 
 // codeSourceConfig is the `code_source` block. Only the "snapshot" type exists.
 type codeSourceConfig struct {
-	Type     string                `yaml:"type" help:"Kind of code source. Must be \"snapshot\", the only supported type." required:"when code_source is set"`
-	Snapshot *snapshotSourceConfig `yaml:"snapshot" help:"Which local directory to archive and upload." required:"when code_source.type is \"snapshot\""`
+	Type           *string               `yaml:"type,omitempty" hidden:"yes"`
+	Snapshot       *snapshotSourceConfig `yaml:",inline"`
+	LegacySnapshot *snapshotSourceConfig `yaml:"snapshot,omitempty" hidden:"yes"`
 }
 
 func (c *codeSourceConfig) validate() error {
-	if c.Type != "snapshot" {
-		return fmt.Errorf("code_source.type must be 'snapshot', got %q", c.Type)
+	if c.Type != nil && strings.TrimSpace(*c.Type) == "" {
+		return errors.New("code_source.type cannot be empty; omit it to use snapshot")
 	}
-	if c.Snapshot == nil {
-		return errors.New("code_source.type='snapshot' requires a snapshot configuration")
+	if c.Type != nil && *c.Type != "snapshot" {
+		return fmt.Errorf("code_source.type must be 'snapshot', got %q", *c.Type)
 	}
-	return c.Snapshot.validate()
+	if c.Snapshot != nil && c.LegacySnapshot != nil {
+		return errors.New("code_source cannot combine the nested 'snapshot' block with direct fields; remove the 'snapshot' wrapper and place all fields directly under code_source")
+	}
+
+	snapshot := c.Snapshot
+	prefix := "code_source"
+	if snapshot == nil {
+		snapshot = c.LegacySnapshot
+		prefix = "code_source.snapshot"
+	}
+	if snapshot == nil {
+		return errors.New("code_source requires snapshot configuration with root_path")
+	}
+	if err := snapshot.validate(prefix); err != nil {
+		return err
+	}
+
+	c.Snapshot = snapshot
+	c.LegacySnapshot = nil
+	return nil
 }
 
 // snapshotSourceConfig describes a local directory to tar and upload.
 type snapshotSourceConfig struct {
-	RootPath     string   `yaml:"root_path" help:"Root of the code source to archive. A git-pinned subdirectory packages only that subtree." required:"when code_source.snapshot is set"`
+	RootPath     string   `yaml:"root_path" help:"Root of the code source to archive. A git-pinned subdirectory packages only that subtree." required:"when code_source is set"`
 	RemoteVolume *string  `yaml:"remote_volume" help:"Volume to upload the archive to. Must start with /Volumes/."`
 	Git          *gitRef  `yaml:"git" help:"Pin the snapshot to a specific git revision."`
 	IncludePaths []string `yaml:"include_paths" help:"Restrict the archive to these paths, relative to root_path and without \"..\". Omit to include everything."`
 }
 
-func (s *snapshotSourceConfig) validate() error {
+func (s *snapshotSourceConfig) validate(prefix string) error {
 	if strings.TrimSpace(s.RootPath) == "" {
-		return errors.New("code_source.snapshot.root_path cannot be empty")
+		return fmt.Errorf("%s.root_path cannot be empty", prefix)
 	}
 
 	if s.RemoteVolume != nil && !strings.HasPrefix(*s.RemoteVolume, "/Volumes/") {
-		return errors.New("code_source.snapshot.remote_volume must start with '/Volumes/'")
+		return fmt.Errorf("%s.remote_volume must start with '/Volumes/'", prefix)
 	}
 
 	// A non-nil but empty include_paths is an explicit mistake (omit it instead).
 	if s.IncludePaths != nil && len(s.IncludePaths) == 0 {
-		return errors.New("code_source.snapshot.include_paths cannot be an empty list; either omit it or provide paths")
+		return fmt.Errorf("%s.include_paths cannot be an empty list; either omit it or provide paths", prefix)
 	}
 	for _, p := range s.IncludePaths {
 		p = strings.TrimSpace(p)
 		if p == "" {
-			return errors.New("code_source.snapshot.include_paths entry cannot be empty")
+			return fmt.Errorf("%s.include_paths entry cannot be empty", prefix)
 		}
 		if strings.HasPrefix(p, "/") {
-			return fmt.Errorf("code_source.snapshot.include_paths must be relative paths, got: %s", p)
+			return fmt.Errorf("%s.include_paths must be relative paths, got: %s", prefix, p)
 		}
 		// No parent traversal: snapshots must stay within root_path.
 		if slices.Contains(strings.Split(p, "/"), "..") {
-			return fmt.Errorf("code_source.snapshot.include_paths cannot contain '..' traversal, got: %s", p)
+			return fmt.Errorf("%s.include_paths cannot contain '..' traversal, got: %s", prefix, p)
 		}
 	}
 
@@ -520,6 +540,7 @@ type configField struct {
 	typeName string
 	required string
 	help     string
+	hidden   bool
 	// freeForm marks a user-keyed map (parameters/secrets/env_variables): no
 	// children, yet any sub-path into it is valid.
 	freeForm bool
@@ -555,7 +576,7 @@ func resolveConfigField(path string) (configField, error) {
 			return configField{}, fmt.Errorf("%q is not an object, so it has no field %q", current.path, part)
 		}
 		child, ok := findConfigChild(current.children, part)
-		if !ok {
+		if !ok || child.hidden {
 			return configField{}, unknownConfigFieldError(current, part, strings.Split(trimmed, ".")[:i+1])
 		}
 		current = child
@@ -596,6 +617,9 @@ func cutLast(s, sep string) (before, after string, found bool) {
 func unknownConfigFieldError(parent configField, part string, matched []string) error {
 	names := make([]string, 0, len(parent.children))
 	for _, c := range parent.children {
+		if c.hidden {
+			continue
+		}
 		names = append(names, configLeafName(c.path))
 	}
 	slices.Sort(names)
@@ -653,7 +677,13 @@ func describeStruct(t reflect.Type, prefix string) []configField {
 		if tag == "" || tag == "-" {
 			continue
 		}
-		name, _, _ := strings.Cut(tag, ",")
+		name, options, _ := strings.Cut(tag, ",")
+		if slices.Contains(strings.Split(options, ","), "inline") {
+			if nested := underlyingConfigStruct(f.Type); nested != nil {
+				out = append(out, describeStruct(nested, prefix)...)
+			}
+			continue
+		}
 		if name == "" || name == "-" {
 			continue
 		}
@@ -663,6 +693,7 @@ func describeStruct(t reflect.Type, prefix string) []configField {
 			typeName: configTypeName(f.Type),
 			required: f.Tag.Get("required"),
 			help:     f.Tag.Get("help"),
+			hidden:   f.Tag.Get("hidden") == "yes",
 			freeForm: freeFormConfigFields[name],
 		}
 		if nested := underlyingConfigStruct(f.Type); nested != nil && !field.freeForm {
@@ -737,10 +768,16 @@ func renderConfigField(w io.Writer, f configField) {
 
 	width := 0
 	for _, c := range f.children {
+		if c.hidden {
+			continue
+		}
 		width = max(width, len(configLeafName(c.path)))
 	}
 	fmt.Fprintf(w, "\n  Fields:\n")
 	for _, c := range f.children {
+		if c.hidden {
+			continue
+		}
 		fmt.Fprintf(w, "    %-*s  %s\n", width, configLeafName(c.path), configFieldSummary(c))
 	}
 	fmt.Fprintf(w, "\nUse \"-h %s.<field>\" for details on a field.\n", f.path)
diff --git a/cmd/air/runconfig_load.go b/cmd/air/runconfig_load.go
index 2f91e6bc382..ad7610436a2 100644
--- a/cmd/air/runconfig_load.go
+++ b/cmd/air/runconfig_load.go
@@ -102,6 +102,15 @@ func finishRunConfigLoad(path string, document *yaml.Node, validationYAML []byte
 	if err != nil {
 		return nil, err
 	}
+	// YAML decodes omitted and null pointers alike. Preserve explicit nulls so validation rejects them.
+	if cfg.CodeSource != nil {
+		if cfg.CodeSource.Type == nil && codeSourceFieldPresent(document, "type") {
+			cfg.CodeSource.Type = new(string)
+		}
+		if cfg.CodeSource.LegacySnapshot == nil && codeSourceFieldPresent(document, "snapshot") {
+			cfg.CodeSource.LegacySnapshot = &snapshotSourceConfig{}
+		}
+	}
 	if err := validateRunConfig(cfg); err != nil {
 		return nil, err
 	}
@@ -115,6 +124,15 @@ func finishRunConfigLoad(path string, document *yaml.Node, validationYAML []byte
 	return cfg, nil
 }
 
+func codeSourceFieldPresent(document *yaml.Node, field string) bool {
+	codeSource, _ := mappingValue(document.Content[0], "code_source")
+	if codeSource == nil || codeSource.Kind != yaml.MappingNode {
+		return false
+	}
+	fieldNode, _ := mappingValue(codeSource, field)
+	return fieldNode != nil
+}
+
 func stripYAMLComments(node *yaml.Node) {
 	node.HeadComment = ""
 	node.LineComment = ""
diff --git a/cmd/air/runconfig_override.go b/cmd/air/runconfig_override.go
index feb54f2af47..3fbccd076cb 100644
--- a/cmd/air/runconfig_override.go
+++ b/cmd/air/runconfig_override.go
@@ -67,6 +67,9 @@ func checkOverridePath(parts []string, node configField, fullPath string) error
 	if !ok {
 		names := make([]string, 0, len(node.children))
 		for _, c := range node.children {
+			if c.hidden {
+				continue
+			}
 			names = append(names, configLeafName(c.path))
 		}
 		slices.Sort(names)
diff --git a/cmd/air/runconfig_override_test.go b/cmd/air/runconfig_override_test.go
index 37e7b4991b8..c0dc4937138 100644
--- a/cmd/air/runconfig_override_test.go
+++ b/cmd/air/runconfig_override_test.go
@@ -138,13 +138,15 @@ func TestLoadRunConfigWithOverrides(t *testing.T) {
 		assert.Equal(t, "0", cfg.EnvVariables["RANK"])
 	})
 
-	t.Run("intermediate maps are auto-created", func(t *testing.T) {
-		cfg, err := loadRunConfigWithOverrides(t.Context(), writeConfig(t, overrideBaseConfig), []string{"code_source.type=snapshot", "code_source.snapshot.root_path=/Workspace/x"})
-		require.NoError(t, err)
-		require.NotNil(t, cfg.CodeSource)
-		require.NotNil(t, cfg.CodeSource.Snapshot)
-		assert.Equal(t, "/Workspace/x", cfg.CodeSource.Snapshot.RootPath)
-	})
+	for _, path := range []string{"code_source.root_path", "code_source.snapshot.root_path"} {
+		t.Run(path+" override", func(t *testing.T) {
+			cfg, err := loadRunConfigWithOverrides(t.Context(), writeConfig(t, overrideBaseConfig), []string{path + "=/Workspace/x"})
+			require.NoError(t, err)
+			require.NotNil(t, cfg.CodeSource)
+			require.NotNil(t, cfg.CodeSource.Snapshot)
+			assert.Equal(t, "/Workspace/x", cfg.CodeSource.Snapshot.RootPath)
+		})
+	}
 
 	t.Run("unity catalog image override applies", func(t *testing.T) {
 		cfg, err := loadRunConfigWithOverrides(t.Context(), writeConfig(t, overrideBaseConfig), []string{"environment.unity_catalog_image=main.air.training:prod"})
diff --git a/cmd/air/runconfig_test.go b/cmd/air/runconfig_test.go
index e370c1763f0..3d2fddadd88 100644
--- a/cmd/air/runconfig_test.go
+++ b/cmd/air/runconfig_test.go
@@ -94,6 +94,56 @@ permissions:
 	assert.Len(t, cfg.Permissions, 2)
 }
 
+func TestLoadRunConfig_CodeSourceShapes(t *testing.T) {
+	tests := []struct {
+		name string
+		yaml string
+	}{
+		{"flattened", "code_source:\n  root_path: src\n"},
+		{"flattened with type", "code_source:\n  type: snapshot\n  root_path: src\n"},
+		{"nested", "code_source:\n  snapshot:\n    root_path: src\n"},
+		{"nested with type", "code_source:\n  type: snapshot\n  snapshot:\n    root_path: src\n"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			cfg, err := loadRunConfig(writeConfig(t, minimalConfig+tt.yaml))
+			require.NoError(t, err)
+			require.NotNil(t, cfg.CodeSource.Snapshot)
+			assert.Equal(t, "src", cfg.CodeSource.Snapshot.RootPath)
+			assert.Nil(t, cfg.CodeSource.LegacySnapshot)
+		})
+	}
+}
+
+func TestLoadRunConfig_CodeSourceErrors(t *testing.T) {
+	tests := []struct {
+		name    string
+		yaml    string
+		errFrag string
+	}{
+		{"empty", "code_source: {}\n", "requires snapshot configuration"},
+		{"type only", "code_source:\n  type: snapshot\n", "requires snapshot configuration"},
+		{"empty type", "code_source:\n  type: \"\"\n  root_path: src\n", "cannot be empty; omit it to use snapshot"},
+		{"null type", "code_source:\n  type: null\n  root_path: src\n", "cannot be empty; omit it to use snapshot"},
+		{"unsupported type", "code_source:\n  type: git\n  root_path: src\n", "must be 'snapshot'"},
+		{"mixed shapes", "code_source:\n  root_path: src\n  snapshot:\n    root_path: src\n", "remove the 'snapshot' wrapper and place all fields directly under code_source"},
+		{"mixed with null snapshot", "code_source:\n  root_path: src\n  snapshot: null\n", "remove the 'snapshot' wrapper and place all fields directly under code_source"},
+		{"flattened error path", "code_source:\n  root_path: \"\"\n", "code_source.root_path cannot be empty"},
+		{"nested error path", "code_source:\n  snapshot:\n    root_path: \"\"\n", "code_source.snapshot.root_path cannot be empty"},
+		{"null snapshot", "code_source:\n  snapshot: null\n", "code_source.snapshot.root_path cannot be empty"},
+		{"unknown flattened field", "code_source:\n  root_path: src\n  bogus: true\n", "field bogus not found"},
+		{"unknown nested field", "code_source:\n  snapshot:\n    root_path: src\n    bogus: true\n", "field bogus not found"},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			_, err := loadRunConfig(writeConfig(t, minimalConfig+tt.yaml))
+			require.ErrorContains(t, err, tt.errFrag)
+		})
+	}
+}
+
 // TestLoadRunConfigUnityCatalogImage covers parsing environment.unity_catalog_image,
 // which is mutually exclusive with dependencies/version.
 func TestLoadRunConfigUnityCatalogImage(t *testing.T) {
@@ -427,7 +477,7 @@ func TestSnapshotSourceConfigValidate(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			err := tt.snap.validate()
+			err := tt.snap.validate("code_source")
 			if tt.errFrag == "" {
 				assert.NoError(t, err)
 				return
@@ -488,10 +538,9 @@ func TestResolveConfigField(t *testing.T) {
 		{"bare path", "compute.accelerator_type", "config.compute.accelerator_type", "string", ""},
 		{"top-level required", "config.experiment_name", "config.experiment_name", "string", "yes"},
 		{"int leaf", "config.max_retries", "config.max_retries", "int", ""},
-		{"conditionally required", "config.code_source.type", "config.code_source.type", "string", "when code_source is set"},
 		{"through a slice", "config.permissions.level", "config.permissions.level", "string", "when a grant is listed"},
 		{"free-form map", "config.parameters", "config.parameters", "map of string to any", ""},
-		{"deeply nested", "config.code_source.snapshot.root_path", "config.code_source.snapshot.root_path", "string", "when code_source.snapshot is set"},
+		{"flattened snapshot", "config.code_source.root_path", "config.code_source.root_path", "string", "when code_source is set"},
 	}
 
 	for _, tt := range tests {
@@ -513,7 +562,7 @@ func TestResolveConfigField_PolymorphicTypes(t *testing.T) {
 	tests := []struct{ path, wantType string }{
 		{"config.environment.dependencies", "list of strings"},
 		{"config.environment.version", "string or int"},
-		{"config.code_source.snapshot.git.remote", "bool or string"},
+		{"config.code_source.git.remote", "bool or string"},
 	}
 
 	for _, tt := range tests {
@@ -580,8 +629,13 @@ func TestResolveConfigField_Errors(t *testing.T) {
 		},
 		{
 			name:      "nested typo reports the resolved prefix",
-			path:      "config.code_source.snapshot.rootpath",
-			wantParts: []string{`unknown config field "config.code_source.snapshot.rootpath"`, `did you mean "root_path"?`},
+			path:      "config.code_source.rootpath",
+			wantParts: []string{`unknown config field "config.code_source.rootpath"`, `did you mean "root_path"?`},
+		},
+		{
+			name:      "legacy fields are hidden",
+			path:      "config.code_source.snapshot",
+			wantParts: []string{`unknown config field "config.code_source.snapshot"`},
 		},
 	}
 
@@ -641,6 +695,9 @@ func TestConfigFieldsAllDocumented(t *testing.T) {
 	var walk func(fields []configField)
 	walk = func(fields []configField) {
 		for _, f := range fields {
+			if f.hidden {
+				continue
+			}
 			assert.NotEmpty(t, f.help, "%s is missing a help: struct tag", f.path)
 			assert.NotEmpty(t, f.typeName, "%s has no type name", f.path)
 			walk(f.children)
@@ -708,9 +765,17 @@ func TestConfigSchemaSharedByHelpAndOverride(t *testing.T) {
 	paths := []string{
 		"compute.num_accelerators",
 		"environment.unity_catalog_image",
-		"code_source.snapshot.root_path",
+		"code_source.root_path",
 		"env_variables.MY_VAR", // free-form sub-path
 	}
+
+	for _, p := range []string{"code_source.type", "code_source.snapshot.root_path"} {
+		t.Run(p+" compatibility override", func(t *testing.T) {
+			require.NoError(t, validateOverridePaths([]overrideEntry{{path: p, raw: "x"}}))
+			_, err := resolveConfigField(p)
+			require.Error(t, err)
+		})
+	}
 	for _, p := range paths {
 		t.Run(p, func(t *testing.T) {
 			require.NoError(t, validateOverridePaths([]overrideEntry{{path: p, raw: "x"}}))