diff --git a/code/go/internal/validator/semantic/validate_fleet_reserved_vars.go b/code/go/internal/validator/semantic/validate_fleet_reserved_vars.go
new file mode 100644
index 000000000..34dbc070d
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_fleet_reserved_vars.go
@@ -0,0 +1,348 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"fmt"
+	"io/fs"
+	"path"
+	"slices"
+	"strings"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+const (
+	useAPMVarName        = "use_apm"
+	datasetVarName       = "data_stream.dataset"
+	tracesDataStreamType = "traces"
+)
+
+// varScope identifies where in a package manifest a variable is declared.
+type varScope string
+
+const (
+	scopeRoot           varScope = "root"            // package-level vars
+	scopePolicyTemplate varScope = "policy template" // policy_templates[].vars
+	scopeInput          varScope = "input"           // policy_templates[].inputs[].vars
+	scopeStream         varScope = "stream"          // data stream stream entries (the valid scope for reserved vars)
+)
+
+// reservedVarRule defines the constraints for a single Fleet-reserved variable.
+// Adding a new reserved variable is a matter of adding an entry to reservedVarRules.
+type reservedVarRule struct {
+	// allowedScopes lists the scopes where this variable may be declared.
+	allowedScopes []varScope
+	// validate checks type and eligibility constraints at an allowed scope.
+	// May be nil if only scope enforcement is needed.
+	validate func(v varDef, ctx varValidationContext) specerrors.ValidationErrors
+}
+
+// isAllowedAt reports whether scope is in the rule's allowedScopes.
+func (r reservedVarRule) isAllowedAt(scope varScope) bool {
+	return slices.Contains(r.allowedScopes, scope)
+}
+
+// scopeViolationMsg returns a human-readable description of where the variable
+// must be declared, derived from allowedScopes (e.g. "stream level" or
+// "root or stream level").
+func (r reservedVarRule) scopeViolationMsg() string {
+	names := make([]string, len(r.allowedScopes))
+	for i, s := range r.allowedScopes {
+		names[i] = string(s) + " level"
+	}
+	return strings.Join(names, " or ")
+}
+
+var reservedVarRules = map[string]reservedVarRule{
+	useAPMVarName:  {allowedScopes: []varScope{scopeStream}, validate: validateUseAPMVar},
+	datasetVarName: {allowedScopes: []varScope{scopeStream}, validate: validateDatasetVar},
+}
+
+type varDef struct {
+	Name string `yaml:"name"`
+	Type string `yaml:"type"`
+}
+
+// varValidationContext carries all available context at the point a variable is
+// validated. filePath, contextStr, manifest, and scope are always set.
+// policyTemplate is set at scopePolicyTemplate, scopeInput, and scopeStream
+// for input packages. stream is set at scopeStream only.
+type varValidationContext struct {
+	manifest       packageManifest
+	scope          varScope
+	filePath       string
+	contextStr     string
+	policyTemplate *policyTemplate
+	stream         *normalizedStream
+}
+
+// normalizedStream holds stream-specific fields used for content validation.
+// It is a unified representation regardless of whether the stream originates
+// from an input package policy template or an integration data stream manifest,
+// mirroring Fleet's own stream normalization approach.
+type normalizedStream struct {
+	inputType          string
+	dataStreamType     string
+	dynamicSignalTypes bool
+}
+
+// policyInput parses the type and vars from an integration policy template input entry.
+type policyInput struct {
+	Type string   `yaml:"type"`
+	Vars []varDef `yaml:"vars"`
+}
+
+// policyTemplate parses fields from a package policy template. Input (singular)
+// is used by input packages; Inputs (plural) is used by integration packages.
+type policyTemplate struct {
+	Name               string        `yaml:"name"`
+	Input              string        `yaml:"input"`
+	Type               string        `yaml:"type"`
+	DynamicSignalTypes bool          `yaml:"dynamic_signal_types"`
+	Vars               []varDef      `yaml:"vars"`
+	Inputs             []policyInput `yaml:"inputs"`
+}
+
+// packageManifest captures all fields needed for reserved variable validation
+// across both input and integration packages.
+type packageManifest struct {
+	Type            string           `yaml:"type"`
+	Vars            []varDef         `yaml:"vars"`
+	PolicyTemplates []policyTemplate `yaml:"policy_templates"`
+}
+
+// streamEntry parses the fields needed from a data stream manifest stream entry.
+type streamEntry struct {
+	Input              string   `yaml:"input"`
+	DynamicSignalTypes bool     `yaml:"dynamic_signal_types"`
+	Vars               []varDef `yaml:"vars"`
+}
+
+// ValidateFleetReservedVars validates that Fleet-reserved variables, when
+// explicitly defined in package manifests, conform to Fleet's expectations.
+func ValidateFleetReservedVars(fsys fspath.FS) specerrors.ValidationErrors {
+	manifestPath := "manifest.yml"
+	data, err := fs.ReadFile(fsys, manifestPath)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to read manifest: %w", fsys.Path(manifestPath), err)}
+	}
+
+	var manifest packageManifest
+	if err := yaml.Unmarshal(data, &manifest); err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to parse manifest: %w", fsys.Path(manifestPath), err)}
+	}
+
+	switch manifest.Type {
+	case inputPackageType:
+		return validateInputReservedVars(fsys, manifest)
+	case integrationPackageType:
+		return validateIntegrationReservedVars(fsys, manifest)
+	}
+
+	return nil
+}
+
+// reservedVarChecker accumulates validation errors while walking the var
+// declarations in a package manifest. Its check method is the single call
+// site for all reserved variable validation, regardless of scope.
+type reservedVarChecker struct {
+	errs specerrors.ValidationErrors
+}
+
+// check validates every var in vars using the provided context.
+func (c *reservedVarChecker) check(vars []varDef, ctx varValidationContext) {
+	for _, v := range vars {
+		c.errs = append(c.errs, validateReservedVar(v, ctx)...)
+	}
+}
+
+func validateInputReservedVars(fsys fspath.FS, manifest packageManifest) specerrors.ValidationErrors {
+	filePath := fsys.Path("manifest.yml")
+	c := &reservedVarChecker{}
+
+	c.check(manifest.Vars, varValidationContext{
+		manifest:   manifest,
+		scope:      scopeRoot,
+		filePath:   filePath,
+		contextStr: "package root vars",
+	})
+
+	// Policy template vars are stream context for input packages: Fleet synthesizes
+	// a data stream from each policy template, placing these vars into streams[0].vars.
+	for _, pt := range manifest.PolicyTemplates {
+		stream := normalizeInputStream(pt)
+		c.check(pt.Vars, varValidationContext{
+			manifest:       manifest,
+			scope:          scopeStream,
+			filePath:       filePath,
+			contextStr:     fmt.Sprintf("policy template %q", pt.Name),
+			policyTemplate: &pt,
+			stream:         &stream,
+		})
+	}
+
+	return c.errs
+}
+
+func validateIntegrationReservedVars(fsys fspath.FS, manifest packageManifest) specerrors.ValidationErrors {
+	filePath := fsys.Path("manifest.yml")
+	c := &reservedVarChecker{}
+
+	c.check(manifest.Vars, varValidationContext{
+		manifest:   manifest,
+		scope:      scopeRoot,
+		filePath:   filePath,
+		contextStr: "package root vars",
+	})
+
+	for _, pt := range manifest.PolicyTemplates {
+		c.check(pt.Vars, varValidationContext{
+			manifest:       manifest,
+			scope:          scopePolicyTemplate,
+			filePath:       filePath,
+			contextStr:     fmt.Sprintf("policy template %q vars", pt.Name),
+			policyTemplate: &pt,
+		})
+		for _, input := range pt.Inputs {
+			c.check(input.Vars, varValidationContext{
+				manifest:       manifest,
+				scope:          scopeInput,
+				filePath:       filePath,
+				contextStr:     fmt.Sprintf("policy template %q input %q vars", pt.Name, input.Type),
+				policyTemplate: &pt,
+			})
+		}
+	}
+
+	dataStreams, err := listDataStreams(fsys)
+	if err != nil {
+		return append(c.errs, specerrors.NewStructuredErrorf("failed to list data streams: %w", err))
+	}
+
+	for _, ds := range dataStreams {
+		manifestPath := path.Join(dataStreamDir, ds, "manifest.yml")
+		data, err := fs.ReadFile(fsys, manifestPath)
+		if err != nil {
+			c.errs = append(c.errs, specerrors.NewStructuredErrorf(
+				"file \"%s\" is invalid: failed to read manifest: %w", fsys.Path(manifestPath), err))
+			continue
+		}
+
+		var dsManifest struct {
+			Type    string        `yaml:"type"`
+			Streams []streamEntry `yaml:"streams"`
+		}
+		if err := yaml.Unmarshal(data, &dsManifest); err != nil {
+			c.errs = append(c.errs, specerrors.NewStructuredErrorf(
+				"file \"%s\" is invalid: failed to parse manifest: %w", fsys.Path(manifestPath), err))
+			continue
+		}
+
+		dsFilePath := fsys.Path(manifestPath)
+		for _, entry := range dsManifest.Streams {
+			stream := normalizeIntegrationStream(entry, dsManifest.Type)
+			c.check(entry.Vars, varValidationContext{
+				manifest:   manifest,
+				scope:      scopeStream,
+				filePath:   dsFilePath,
+				contextStr: fmt.Sprintf("stream with input type %q", entry.Input),
+				stream:     &stream,
+			})
+		}
+	}
+
+	return c.errs
+}
+
+func normalizeInputStream(pt policyTemplate) normalizedStream {
+	return normalizedStream{
+		inputType:          pt.Input,
+		dataStreamType:     pt.Type,
+		dynamicSignalTypes: pt.DynamicSignalTypes,
+	}
+}
+
+func normalizeIntegrationStream(entry streamEntry, dsType string) normalizedStream {
+	return normalizedStream{
+		inputType:          entry.Input,
+		dataStreamType:     dsType,
+		dynamicSignalTypes: entry.DynamicSignalTypes,
+	}
+}
+
+// validateReservedVar is the single entry point for all reserved variable
+// validation. It checks scope constraints via the rule's allowedScopes and,
+// when the scope is valid, runs per-variable content validation.
+func validateReservedVar(v varDef, ctx varValidationContext) specerrors.ValidationErrors {
+	rule, ok := reservedVarRules[v.Name]
+	if !ok {
+		return nil
+	}
+
+	var errs specerrors.ValidationErrors
+
+	if !rule.isAllowedAt(ctx.scope) {
+		errs = append(errs, specerrors.NewStructuredErrorf(
+			"file \"%s\" is invalid: %s: variable \"%s\" must only be declared at %s",
+			ctx.filePath, ctx.contextStr, v.Name, rule.scopeViolationMsg()))
+	}
+
+	if rule.isAllowedAt(ctx.scope) && rule.validate != nil {
+		errs = append(errs, rule.validate(v, ctx)...)
+	}
+
+	return errs
+}
+
+// validateUseAPMVar enforces that use_apm is:
+//   - defined on an otelcol input
+//   - of type "bool"
+//   - only present on "traces" data streams or when "dynamic_signal_types" is true
+func validateUseAPMVar(v varDef, ctx varValidationContext) specerrors.ValidationErrors {
+	stream := ctx.stream
+	var errs specerrors.ValidationErrors
+
+	if stream.inputType != otelcolInputType {
+		errs = append(errs, newReservedVarError(ctx.filePath, ctx.contextStr, useAPMVarName,
+			fmt.Sprintf("%q input", otelcolInputType), fmt.Sprintf("%q", stream.inputType)))
+	}
+	if v.Type != "bool" {
+		errs = append(errs, newReservedVarError(ctx.filePath, ctx.contextStr, useAPMVarName,
+			`type "bool"`, fmt.Sprintf("%q", v.Type)))
+	}
+	if stream.dataStreamType != tracesDataStreamType && !stream.dynamicSignalTypes {
+		errs = append(errs, newReservedVarError(ctx.filePath, ctx.contextStr, useAPMVarName,
+			fmt.Sprintf("%q data stream type or \"dynamic_signal_types: true\"", tracesDataStreamType),
+			fmt.Sprintf("%q data stream type", stream.dataStreamType)))
+	}
+
+	return errs
+}
+
+// validateDatasetVar enforces that data_stream.dataset is of type "text".
+func validateDatasetVar(v varDef, ctx varValidationContext) specerrors.ValidationErrors {
+	if v.Type != "text" {
+		return specerrors.ValidationErrors{
+			newReservedVarError(ctx.filePath, ctx.contextStr, datasetVarName,
+				`type "text"`, fmt.Sprintf("%q", v.Type)),
+		}
+	}
+	return nil
+}
+
+// newReservedVarError constructs a validation error for a Fleet-reserved
+// variable that doesn't satisfy its constraint. expected describes the
+// requirement (e.g. `type "bool"`) and got is the actual value found.
+func newReservedVarError(filePath, contextStr, varName, expected, got string) *specerrors.StructuredError {
+	return specerrors.NewStructuredErrorf(
+		"file \"%s\" is invalid: %s: variable \"%s\" must be %s, got %s",
+		filePath, contextStr, varName, expected, got)
+}
diff --git a/code/go/internal/validator/semantic/validate_fleet_reserved_vars_test.go b/code/go/internal/validator/semantic/validate_fleet_reserved_vars_test.go
new file mode 100644
index 000000000..3adf3fefc
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_fleet_reserved_vars_test.go
@@ -0,0 +1,159 @@
+// Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+// or more contributor license agreements. Licensed under the Elastic License;
+// you may not use this file except in compliance with the Elastic License.
+
+package semantic
+
+import (
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+)
+
+func TestValidateFleetReservedVars(t *testing.T) {
+	t.Run("non_input_or_integration_package_type_is_skipped", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: content
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Empty(t, errs, "expected no validation errors for non-input/integration package types")
+	})
+
+	t.Run("non_reserved_variable_names_are_ignored", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+policy_templates:
+  - name: sample
+    input: otelcol
+    vars:
+      - name: foobar
+        type: bool
+      - name: some_other_var
+        type: text
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Empty(t, errs, "expected no validation errors for non-reserved variable names")
+	})
+
+	t.Run("use_apm_wrong_input_and_wrong_type_both_reported", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+policy_templates:
+  - name: sample
+    type: traces
+    input: logfile
+    vars:
+      - name: use_apm
+        type: text
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Len(t, errs, 2, "expected both input type and variable type violations to be reported")
+		assert.Contains(t, errs[0].Error(), `variable "use_apm" must be "otelcol" input, got "logfile"`)
+		assert.Contains(t, errs[1].Error(), `variable "use_apm" must be type "bool", got "text"`)
+	})
+
+	t.Run("use_apm_wrong_eligibility_reported", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+policy_templates:
+  - name: sample
+    type: logs
+    input: otelcol
+    vars:
+      - name: use_apm
+        type: bool
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Len(t, errs, 1, "expected one eligibility violation to be reported")
+		assert.Contains(t, errs[0].Error(), `variable "use_apm" must be "traces" data stream type or "dynamic_signal_types: true", got "logs" data stream type`)
+	})
+
+	t.Run("use_apm_dynamic_signal_types_bypasses_eligibility", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+policy_templates:
+  - name: sample
+    type: logs
+    input: otelcol
+    dynamic_signal_types: true
+    vars:
+      - name: use_apm
+        type: bool
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Empty(t, errs, "expected no errors when dynamic_signal_types is true")
+	})
+
+	t.Run("reserved_var_at_input_package_root_level_is_flagged", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: input
+vars:
+  - name: data_stream.dataset
+    type: text
+policy_templates:
+  - name: sample
+    type: traces
+    input: otelcol
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Len(t, errs, 1, "expected one scope violation for root-level reserved var in input package")
+		assert.Contains(t, errs[0].Error(), `package root vars: variable "data_stream.dataset" must only be declared at stream level`)
+	})
+
+	t.Run("reserved_vars_at_integration_non_stream_scopes_are_flagged", func(t *testing.T) {
+		d := t.TempDir()
+
+		err := os.WriteFile(filepath.Join(d, "manifest.yml"), []byte(`
+type: integration
+vars:
+  - name: use_apm
+    type: bool
+policy_templates:
+  - name: sample
+    vars:
+      - name: data_stream.dataset
+        type: text
+    inputs:
+      - type: logfile
+        vars:
+          - name: use_apm
+            type: bool
+`), 0o644)
+		require.NoError(t, err)
+
+		errs := ValidateFleetReservedVars(fspath.DirFS(d))
+		require.Len(t, errs, 3, "expected scope violations at root, policy template, and input levels")
+		assert.Contains(t, errs[0].Error(), `package root vars: variable "use_apm" must only be declared at stream level`)
+		assert.Contains(t, errs[1].Error(), `policy template "sample" vars: variable "data_stream.dataset" must only be declared at stream level`)
+		assert.Contains(t, errs[2].Error(), `policy template "sample" input "logfile" vars: variable "use_apm" must only be declared at stream level`)
+	})
+}
diff --git a/code/go/internal/validator/spec.go b/code/go/internal/validator/spec.go
index 44b7995f1..608d4e935 100644
--- a/code/go/internal/validator/spec.go
+++ b/code/go/internal/validator/spec.go
@@ -230,6 +230,7 @@ func (s Spec) rules(pkgType string, rootSpec spectypes.ItemSpec) validationRules
 		{fn: semantic.ValidateDurationVariables, since: semver.MustParse("3.5.0")},
 		{fn: semantic.ValidateInputPackagesPolicyTemplates, types: []string{"input"}},
 		{fn: semantic.ValidateInputDynamicSignalTypes, since: semver.MustParse("3.6.0")},
+		{fn: semantic.ValidateFleetReservedVars, types: []string{"integration", "input"}, since: semver.MustParse("3.6.1")},
 		{fn: semantic.ValidateMinimumAgentVersion},
 		{fn: semantic.ValidateIntegrationPolicyTemplates, types: []string{"integration"}},
 		{fn: semantic.ValidatePipelineTags, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
diff --git a/code/go/pkg/validator/validator_test.go b/code/go/pkg/validator/validator_test.go
index 68e41b6d4..8cef3c775 100644
--- a/code/go/pkg/validator/validator_test.go
+++ b/code/go/pkg/validator/validator_test.go
@@ -37,6 +37,8 @@ func TestValidateFile(t *testing.T) {
 		"good_input":                             {},
 		"good_input_otel":                        {},
 		"good_input_qualifier":                   {},
+		"good_input_fleet_reserved_vars":         {},
+		"good_integration_fleet_reserved_vars":   {},
 		"good_input_dynamic_signal_type":         {},
 		"good_input_profiles":                    {},
 		"good_input_template_paths":              {},
@@ -447,6 +449,24 @@ func TestValidateFile(t *testing.T) {
 				"policy template \"otel_logs\": type field must not be set when dynamic_signal_types is true",
 			},
 		},
+		"bad_input_fleet_reserved_vars": {
+			"manifest.yml",
+			[]string{
+				"package root vars: variable \"data_stream.dataset\" must only be declared at stream level",
+				"policy template \"sample\": variable \"use_apm\" must be type \"bool\", got \"text\"",
+				"policy template \"sample\": variable \"data_stream.dataset\" must be type \"text\", got \"bool\"",
+			},
+		},
+		"bad_integration_fleet_reserved_vars": {
+			"data_stream/sample/manifest.yml",
+			[]string{
+				"stream with input type \"otelcol\": variable \"use_apm\" must be type \"bool\", got \"text\"",
+				"stream with input type \"otelcol\": variable \"use_apm\" must be \"traces\" data stream type or \"dynamic_signal_types: true\", got \"logs\" data stream type",
+				"stream with input type \"logfile\": variable \"use_apm\" must be \"otelcol\" input, got \"logfile\"",
+				"stream with input type \"logfile\": variable \"use_apm\" must be \"traces\" data stream type or \"dynamic_signal_types: true\", got \"logs\" data stream type",
+				"stream with input type \"logfile\": variable \"data_stream.dataset\" must be type \"text\", got \"yaml\"",
+			},
+		},
 		"bad_input_template_path": {
 			"manifest.yml",
 			[]string{
diff --git a/spec/changelog.yml b/spec/changelog.yml
index b02417985..da66b2d70 100644
--- a/spec/changelog.yml
+++ b/spec/changelog.yml
@@ -17,6 +17,9 @@
     - description: Add support for named inputs in policy templates to disambiguate multiple inputs of the same type.
       type: enhancement
       link: https://github.com/elastic/package-spec/pull/1135
+    - description: Add validation for Fleet-reserved variables (use_apm, data_stream.dataset) to enforce correct types and constraints when explicitly overridden in package manifests.
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1134
 - version: 3.6.0
   changes:
     - description: Add pipeline tag validations.
diff --git a/test/packages/bad_input_fleet_reserved_vars/agent/input/input.yml.hbs b/test/packages/bad_input_fleet_reserved_vars/agent/input/input.yml.hbs
new file mode 100644
index 000000000..bd1ec6c14
--- /dev/null
+++ b/test/packages/bad_input_fleet_reserved_vars/agent/input/input.yml.hbs
@@ -0,0 +1 @@
+receivers: {}
diff --git a/test/packages/bad_input_fleet_reserved_vars/changelog.yml b/test/packages/bad_input_fleet_reserved_vars/changelog.yml
new file mode 100644
index 000000000..e00f88133
--- /dev/null
+++ b/test/packages/bad_input_fleet_reserved_vars/changelog.yml
@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1
diff --git a/test/packages/bad_input_fleet_reserved_vars/docs/README.md b/test/packages/bad_input_fleet_reserved_vars/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/bad_input_fleet_reserved_vars/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/bad_input_fleet_reserved_vars/manifest.yml b/test/packages/bad_input_fleet_reserved_vars/manifest.yml
new file mode 100644
index 000000000..99616d6bf
--- /dev/null
+++ b/test/packages/bad_input_fleet_reserved_vars/manifest.yml
@@ -0,0 +1,43 @@
+format_version: 3.6.1
+name: bad_input_fleet_reserved_vars
+title: "Bad OTel input with incorrectly typed Fleet-reserved variables"
+version: 0.0.1
+source:
+  license: "Elastic-2.0"
+description: "Input package that declares Fleet-reserved variables with wrong types."
+type: input
+vars:
+  - name: data_stream.dataset
+    type: text
+    title: Dataset name
+    show_user: false
+categories:
+  - custom
+conditions:
+  kibana:
+    version: "^9.2.0"
+  elastic:
+    subscription: "basic"
+policy_templates:
+  - name: sample
+    type: traces
+    title: Sample OTel metrics
+    description: Collect sample OTel metrics
+    input: otelcol
+    template_path: input.yml.hbs
+    vars:
+      - name: use_apm
+        type: text
+        title: Enable Elastic APM Enrichment
+        show_user: true
+        required: false
+        default: "false"
+      - name: data_stream.dataset
+        type: bool
+        title: Dataset name
+        show_user: true
+        required: false
+        default: false
+owner:
+  github: elastic/ecosystem
+  type: elastic
diff --git a/test/packages/bad_integration_fleet_reserved_vars/LICENSE.txt b/test/packages/bad_integration_fleet_reserved_vars/LICENSE.txt
new file mode 100644
index 000000000..3d74f227e
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/LICENSE.txt
@@ -0,0 +1 @@
+Apache License 2.0
diff --git a/test/packages/bad_integration_fleet_reserved_vars/changelog.yml b/test/packages/bad_integration_fleet_reserved_vars/changelog.yml
new file mode 100644
index 000000000..e00f88133
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/changelog.yml
@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1
diff --git a/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/agent/stream/stream.yml.hbs b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..0967ef424
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{}
diff --git a/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/fields/base-fields.yml b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/fields/base-fields.yml
new file mode 100644
index 000000000..0d1791ffe
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/fields/base-fields.yml
@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: "@timestamp"
+  type: date
+  description: Event timestamp.
diff --git a/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/manifest.yml b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/manifest.yml
new file mode 100644
index 000000000..8a1b4792a
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/data_stream/sample/manifest.yml
@@ -0,0 +1,28 @@
+title: Sample
+type: logs
+streams:
+  - input: otelcol
+    title: Collect via OTel receiver
+    description: Collecting data via OpenTelemetry receiver
+    vars:
+      - name: use_apm
+        type: text
+        title: Enable Elastic APM Enrichment
+        show_user: true
+        required: false
+        default: "false"
+  - input: logfile
+    title: Log files
+    description: Collect log files
+    vars:
+      - name: use_apm
+        type: bool
+        title: Enable Elastic APM Enrichment
+        show_user: true
+        required: false
+        default: false
+      - name: data_stream.dataset
+        type: yaml
+        title: Dataset name
+        show_user: true
+        required: false
diff --git a/test/packages/bad_integration_fleet_reserved_vars/docs/README.md b/test/packages/bad_integration_fleet_reserved_vars/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/bad_integration_fleet_reserved_vars/manifest.yml b/test/packages/bad_integration_fleet_reserved_vars/manifest.yml
new file mode 100644
index 000000000..9787d453e
--- /dev/null
+++ b/test/packages/bad_integration_fleet_reserved_vars/manifest.yml
@@ -0,0 +1,29 @@
+format_version: 3.6.1
+name: bad_integration_fleet_reserved_vars
+title: "Bad integration with incorrectly defined Fleet-reserved variables"
+version: 0.0.1
+source:
+  license: "Apache-2.0"
+description: "Integration package with a data stream whose stream entries define Fleet-reserved variables incorrectly."
+type: integration
+categories:
+  - custom
+conditions:
+  kibana:
+    version: "^9.2.0"
+  elastic:
+    subscription: "basic"
+policy_templates:
+  - name: sample
+    title: Sample
+    description: Collect sample data
+    inputs:
+      - type: otelcol
+        title: Collect via OTel receiver
+        description: Collecting data via OpenTelemetry receiver
+      - type: logfile
+        title: Log files
+        description: Collect log files
+owner:
+  github: elastic/ecosystem
+  type: elastic
diff --git a/test/packages/good_input_fleet_reserved_vars/agent/input/input.yml.hbs b/test/packages/good_input_fleet_reserved_vars/agent/input/input.yml.hbs
new file mode 100644
index 000000000..bd1ec6c14
--- /dev/null
+++ b/test/packages/good_input_fleet_reserved_vars/agent/input/input.yml.hbs
@@ -0,0 +1 @@
+receivers: {}
diff --git a/test/packages/good_input_fleet_reserved_vars/changelog.yml b/test/packages/good_input_fleet_reserved_vars/changelog.yml
new file mode 100644
index 000000000..e00f88133
--- /dev/null
+++ b/test/packages/good_input_fleet_reserved_vars/changelog.yml
@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1
diff --git a/test/packages/good_input_fleet_reserved_vars/docs/README.md b/test/packages/good_input_fleet_reserved_vars/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/good_input_fleet_reserved_vars/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/good_input_fleet_reserved_vars/manifest.yml b/test/packages/good_input_fleet_reserved_vars/manifest.yml
new file mode 100644
index 000000000..f604174fc
--- /dev/null
+++ b/test/packages/good_input_fleet_reserved_vars/manifest.yml
@@ -0,0 +1,38 @@
+format_version: 3.6.1
+name: good_input_fleet_reserved_vars
+title: "Good OTel input with valid Fleet-reserved variable overrides"
+version: 0.0.1
+source:
+  license: "Elastic-2.0"
+description: "Input package that overrides Fleet-reserved variables with correct types."
+type: input
+categories:
+  - custom
+conditions:
+  kibana:
+    version: "^9.2.0"
+  elastic:
+    subscription: "basic"
+policy_templates:
+  - name: sample
+    dynamic_signal_types: true
+    title: Sample OTel metrics
+    description: Collect sample OTel metrics
+    input: otelcol
+    template_path: input.yml.hbs
+    vars:
+      - name: use_apm
+        type: bool
+        title: Enable Elastic APM Enrichment
+        show_user: true
+        required: false
+        default: false
+      - name: data_stream.dataset
+        type: text
+        title: Dataset name
+        show_user: true
+        required: false
+        default: generic
+owner:
+  github: elastic/ecosystem
+  type: elastic
diff --git a/test/packages/good_integration_fleet_reserved_vars/LICENSE.txt b/test/packages/good_integration_fleet_reserved_vars/LICENSE.txt
new file mode 100644
index 000000000..3d74f227e
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/LICENSE.txt
@@ -0,0 +1 @@
+Apache License 2.0
diff --git a/test/packages/good_integration_fleet_reserved_vars/changelog.yml b/test/packages/good_integration_fleet_reserved_vars/changelog.yml
new file mode 100644
index 000000000..e00f88133
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/changelog.yml
@@ -0,0 +1,6 @@
+# newer versions go on top
+- version: "0.0.1"
+  changes:
+    - description: Initial draft of the package
+      type: enhancement
+      link: https://github.com/elastic/integrations/pull/1
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/log/agent/stream/stream.yml.hbs b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..0967ef424
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{}
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/log/fields/base-fields.yml b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/fields/base-fields.yml
new file mode 100644
index 000000000..0d1791ffe
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/fields/base-fields.yml
@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: "@timestamp"
+  type: date
+  description: Event timestamp.
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/log/manifest.yml b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/manifest.yml
new file mode 100644
index 000000000..71fc9e9e3
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/log/manifest.yml
@@ -0,0 +1,13 @@
+title: Logs
+type: logs
+streams:
+  - input: logfile
+    title: Log files
+    description: Collect log files
+    vars:
+      - name: data_stream.dataset
+        type: text
+        title: Dataset name
+        show_user: true
+        required: false
+        default: generic
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/agent/stream/stream.yml.hbs b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..0967ef424
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{}
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/fields/base-fields.yml b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/fields/base-fields.yml
new file mode 100644
index 000000000..0d1791ffe
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/fields/base-fields.yml
@@ -0,0 +1,12 @@
+- name: data_stream.type
+  type: constant_keyword
+  description: Data stream type.
+- name: data_stream.dataset
+  type: constant_keyword
+  description: Data stream dataset.
+- name: data_stream.namespace
+  type: constant_keyword
+  description: Data stream namespace.
+- name: "@timestamp"
+  type: date
+  description: Event timestamp.
diff --git a/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/manifest.yml b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/manifest.yml
new file mode 100644
index 000000000..ee753dd2d
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/data_stream/otel/manifest.yml
@@ -0,0 +1,19 @@
+title: OTel traces
+type: traces
+streams:
+  - input: otelcol
+    title: Collect via OTel receiver
+    description: Collecting data via OpenTelemetry receiver
+    vars:
+      - name: use_apm
+        type: bool
+        title: Enable Elastic APM Enrichment
+        show_user: true
+        required: false
+        default: false
+      - name: data_stream.dataset
+        type: text
+        title: Dataset name
+        show_user: true
+        required: false
+        default: generic
diff --git a/test/packages/good_integration_fleet_reserved_vars/docs/README.md b/test/packages/good_integration_fleet_reserved_vars/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/good_integration_fleet_reserved_vars/manifest.yml b/test/packages/good_integration_fleet_reserved_vars/manifest.yml
new file mode 100644
index 000000000..a127e1fcc
--- /dev/null
+++ b/test/packages/good_integration_fleet_reserved_vars/manifest.yml
@@ -0,0 +1,33 @@
+format_version: 3.6.1
+name: good_integration_fleet_reserved_vars
+title: "Good integration with valid Fleet-reserved variable overrides"
+version: 0.0.1
+source:
+  license: "Apache-2.0"
+description: "Integration package with multiple streams that correctly override Fleet-reserved variables."
+type: integration
+categories:
+  - custom
+conditions:
+  kibana:
+    version: "^9.2.0"
+  elastic:
+    subscription: "basic"
+policy_templates:
+  - name: otel
+    title: OTel logs
+    description: Collect logs via OTel
+    inputs:
+      - type: otelcol
+        title: Collect via OTel receiver
+        description: Collecting data via OpenTelemetry receiver
+  - name: log
+    title: Logs
+    description: Collect logs
+    inputs:
+      - type: logfile
+        title: Log files
+        description: Collect log files
+owner:
+  github: elastic/ecosystem
+  type: elastic