diff --git a/code/go/internal/validator/semantic/validate_datastream_package_categories.go b/code/go/internal/validator/semantic/validate_datastream_package_categories.go
new file mode 100644
index 000000000..0a4de7638
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_datastream_package_categories.go
@@ -0,0 +1,164 @@
+// 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"
+	"net/http"
+	"path"
+	"slices"
+	"sort"
+	"time"
+
+	"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 packageRegistryCategoriesURL = "https://raw.githubusercontent.com/elastic/package-registry/v1.38.0/categories/categories.yml"
+
+type registryCategories struct {
+	Categories map[string]struct {
+		Title         string `yaml:"title"`
+		Subcategories map[string]struct {
+			Title string `yaml:"title"`
+		} `yaml:"subcategories"`
+	} `yaml:"categories"`
+}
+
+func fetchRegistryCategoryToParentMap() (map[string]string, error) {
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Get(packageRegistryCategoriesURL)
+	if err != nil {
+		return nil, fmt.Errorf("failed to fetch categories from package registry: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected HTTP %d (%s) fetching %s", resp.StatusCode, resp.Status, packageRegistryCategoriesURL)
+	}
+
+	body, err := io.ReadAll(resp.Body)
+	if err != nil {
+		return nil, fmt.Errorf("failed to read categories response from %s: %w", packageRegistryCategoriesURL, err)
+	}
+
+	var rc registryCategories
+	if err := yaml.Unmarshal(body, &rc); err != nil {
+		return nil, fmt.Errorf("failed to parse categories YAML from %s: %w", packageRegistryCategoriesURL, err)
+	}
+
+	if len(rc.Categories) == 0 {
+		return nil, fmt.Errorf("no categories found in response from %s", packageRegistryCategoriesURL)
+	}
+
+	categoryToParent := make(map[string]string)
+	for parentID, cat := range rc.Categories {
+		categoryToParent[parentID] = parentID
+		for subID := range cat.Subcategories {
+			categoryToParent[subID] = parentID
+		}
+	}
+	return categoryToParent, nil
+}
+
+func readPackageManifestTypeAndCategories(fsys fspath.FS) (string, []string, error) {
+	manifest, err := readManifest(fsys)
+	if err != nil {
+		return "", nil, err
+	}
+
+	typeVal, err := manifest.Values("$.type")
+	if err != nil {
+		return "", nil, fmt.Errorf("can't read manifest type: %w", err)
+	}
+	pkgType, ok := typeVal.(string)
+	if !ok {
+		return "", nil, fmt.Errorf("manifest type is not a string")
+	}
+
+	catsVal, err := manifest.Values("$.categories[*]")
+	if err != nil {
+		// categories field may be absent
+		return pkgType, nil, nil
+	}
+	cats, err := toStringSlice(catsVal)
+	if err != nil {
+		return "", nil, fmt.Errorf("can't read manifest categories: %w", err)
+	}
+	return pkgType, cats, nil
+}
+
+// ValidateDatastreamPackageCategories validates that the package manifest
+// categories include all parent-level equivalent categories present in any data stream
+// manifest. Parent categories are determined by fetching the package registry
+// categories.yml. Data stream manifests without a categories field are skipped.
+func ValidateDatastreamPackageCategories(fsys fspath.FS) specerrors.ValidationErrors {
+	manifestPath := "manifest.yml"
+	pkgType, pkgCategories, err := readPackageManifestTypeAndCategories(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), err)}
+	}
+
+	if pkgType != packageTypeIntegration {
+		return nil
+	}
+
+	dsCategories, err := readDataStreamManifestCategories(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), err)}
+	}
+
+	if len(dsCategories) == 0 {
+		return nil
+	}
+
+	categoryToParent, err := fetchRegistryCategoryToParentMap()
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: failed to load registry categories: %w", fsys.Path(manifestPath), err)}
+	}
+
+	var errs specerrors.ValidationErrors
+	for dsName, dsCats := range dsCategories {
+		var missingCats []string
+		for _, dsCat := range dsCats {
+			parent, known := categoryToParent[dsCat]
+			if !known {
+				dsManifestPath := path.Join("data_stream", dsName, "manifest.yml")
+				errs = append(errs, specerrors.NewStructuredErrorf(
+					"file \"%s\" is invalid: data stream \"%s\" has unrecognized category \"%s\" (defined in \"%s\")",
+					fsys.Path(manifestPath),
+					dsName,
+					dsCat,
+					fsys.Path(dsManifestPath),
+				))
+				continue
+			}
+			if !slices.Contains(pkgCategories, parent) && !slices.Contains(missingCats, parent) {
+				missingCats = append(missingCats, parent)
+			}
+		}
+
+		if len(missingCats) > 0 {
+			sort.Strings(missingCats)
+			dsManifestPath := path.Join("data_stream", dsName, "manifest.yml")
+			errs = append(errs, specerrors.NewStructuredErrorf(
+				"file \"%s\" is invalid: package manifest categories %v are missing parent categories %v from data stream \"%s\" (defined in \"%s\")",
+				fsys.Path(manifestPath),
+				pkgCategories,
+				missingCats,
+				dsName,
+				fsys.Path(dsManifestPath),
+			))
+		}
+	}
+
+	return errs
+}
diff --git a/code/go/internal/validator/semantic/validate_datastream_package_categories_test.go b/code/go/internal/validator/semantic/validate_datastream_package_categories_test.go
new file mode 100644
index 000000000..58a957005
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_datastream_package_categories_test.go
@@ -0,0 +1,216 @@
+// 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 (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+)
+
+func TestValidateDatastreamPackageCategories(t *testing.T) {
+	cases := []struct {
+		title        string
+		setup        func(t *testing.T, dir string)
+		expectedErrs []string
+	}{
+		{
+			title: "package includes datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+		{
+			title: "package missing datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+			expectedErrs: []string{
+				`package manifest categories [observability] are missing parent categories [security] from data stream "mylogs"`,
+			},
+		},
+		{
+			title: "datastream subcategory requires its parent in package",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - credential_management
+type: logs
+`)
+			},
+			expectedErrs: []string{
+				`package manifest categories [observability] are missing parent categories [security] from data stream "mylogs"`,
+			},
+		},
+		{
+			title: "datastream subcategory parent already in package",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - credential_management
+type: logs
+`)
+			},
+		},
+		{
+			title: "data stream manifest has no categories field, skipped",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+type: logs
+`)
+			},
+		},
+		{
+			title: "package has no categories but datastream does",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+			expectedErrs: []string{
+				`package manifest categories [] are missing parent categories [security] from data stream "mylogs"`,
+			},
+		},
+		{
+			title: "package has subset, missing one datastream parent category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+  - observability
+type: logs
+`)
+			},
+			expectedErrs: []string{`are missing parent categories [observability] from data stream "mylogs"`},
+		},
+		{
+			title: "non-integration packages are skipped",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: input
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+		{
+			title: "multiple datastreams, one triggers error",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "auditlogs", `
+title: Audit Logs
+categories:
+  - security
+type: logs
+`)
+				writeDataStreamManifest(t, dir, "netlogs", `
+title: Net Logs
+categories:
+  - network
+type: logs
+`)
+			},
+			expectedErrs: []string{`are missing parent categories [network] from data stream "netlogs"`},
+		},
+		{
+			title: "datastream does not need all package categories",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+categories:
+  - security
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+type: logs
+`)
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.title, func(t *testing.T) {
+			dir := t.TempDir()
+			tc.setup(t, dir)
+
+			errs := ValidateDatastreamPackageCategories(fspath.DirFS(dir))
+
+			if len(tc.expectedErrs) == 0 {
+				assert.Empty(t, errs)
+			} else {
+				require.Len(t, errs, 1)
+				for _, expected := range tc.expectedErrs {
+					assert.Contains(t, errs[0].Error(), expected)
+				}
+			}
+		})
+	}
+}
diff --git a/code/go/internal/validator/semantic/validate_policy_template_datastream_categories.go b/code/go/internal/validator/semantic/validate_policy_template_datastream_categories.go
new file mode 100644
index 000000000..dec6f3643
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_policy_template_datastream_categories.go
@@ -0,0 +1,156 @@
+// 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"
+	"path"
+	"slices"
+	"sort"
+
+	"gopkg.in/yaml.v3"
+
+	"github.com/elastic/package-spec/v3/code/go/internal/fspath"
+	"github.com/elastic/package-spec/v3/code/go/internal/pkgpath"
+	"github.com/elastic/package-spec/v3/code/go/pkg/specerrors"
+)
+
+type policyTemplateWithCategories struct {
+	Name        string   `yaml:"name"`
+	DataStreams []string `yaml:"data_streams"`
+	Categories  []string `yaml:"categories"`
+}
+
+type packageManifestWithCategories struct {
+	Type            string                         `yaml:"type"`
+	PolicyTemplates []policyTemplateWithCategories `yaml:"policy_templates"`
+}
+
+func readPackageManifestPolicyTemplates(fsys fspath.FS) (string, []policyTemplateWithCategories, error) {
+	manifest, err := readManifest(fsys)
+	if err != nil {
+		return "", nil, err
+	}
+
+	typeVal, err := manifest.Values("$.type")
+	if err != nil {
+		return "", nil, err
+	}
+	pkgType, ok := typeVal.(string)
+	if !ok {
+		return "", nil, fmt.Errorf("manifest type is not a string: %v", typeVal)
+	}
+
+	data, err := manifest.ReadAll()
+	if err != nil {
+		return "", nil, err
+	}
+	var pkg packageManifestWithCategories
+	if err := yaml.Unmarshal(data, &pkg); err != nil {
+		return "", nil, err
+	}
+	return pkgType, pkg.PolicyTemplates, nil
+}
+
+// ValidatePolicyTemplateDatastreamCategories validates that when a policy template
+// entry in the package manifest.yml defines categories, each referenced data stream's
+// manifest.yml categories include all of the policy template's categories. Data streams
+// may declare additional categories beyond what the policy template specifies.
+// Data stream manifests without a categories field are skipped.
+func ValidatePolicyTemplateDatastreamCategories(fsys fspath.FS) specerrors.ValidationErrors {
+	var errs specerrors.ValidationErrors
+
+	manifestPath := "manifest.yml"
+	pkgType, policyTemplates, err := readPackageManifestPolicyTemplates(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), err)}
+	}
+
+	// only validate integration type packages
+	if pkgType != packageTypeIntegration {
+		return nil
+	}
+
+	dsCategories, err := readDataStreamManifestCategories(fsys)
+	if err != nil {
+		return specerrors.ValidationErrors{
+			specerrors.NewStructuredErrorf("file \"%s\" is invalid: %w", fsys.Path(manifestPath), err)}
+	}
+
+	for _, pt := range policyTemplates {
+		// skip policy templates that don't define both categories and data_streams
+		if len(pt.Categories) == 0 || len(pt.DataStreams) == 0 {
+			continue
+		}
+
+		for _, dsName := range pt.DataStreams {
+			dsCats, hasCats := dsCategories[dsName]
+			if !hasCats {
+				// data stream manifest has no categories field — nothing to validate
+				continue
+			}
+
+			missing := missingCategories(pt.Categories, dsCats)
+			if len(missing) > 0 {
+				dsManifestPath := path.Join("data_stream", dsName, "manifest.yml")
+				errs = append(errs, specerrors.NewStructuredErrorf(
+					"file \"%s\" is invalid: data stream \"%s\" manifest categories %v are missing policy template \"%s\" categories %v (defined in \"%s\")",
+					fsys.Path(manifestPath),
+					dsName,
+					dsCats,
+					pt.Name,
+					missing,
+					fsys.Path(dsManifestPath),
+				))
+			}
+		}
+	}
+
+	return errs
+}
+
+// readDataStreamManifestCategories reads the categories field from every
+// data_stream/*/manifest.yml and returns a map of data stream name to its categories.
+// Data streams without a categories field are omitted from the map.
+func readDataStreamManifestCategories(fsys fspath.FS) (map[string][]string, error) {
+	files, err := pkgpath.Files(fsys, "data_stream/*/manifest.yml")
+	if err != nil {
+		return nil, err
+	}
+
+	result := make(map[string][]string)
+	for _, f := range files {
+		catsVal, err := f.Values("$.categories[*]")
+		if err != nil {
+			continue
+		}
+		cats, err := toStringSlice(catsVal)
+		if err != nil {
+			return nil, fmt.Errorf("can't read categories from %s: %w", f.Path(), err)
+		}
+		if len(cats) == 0 {
+			continue
+		}
+		// path is data_stream/{name}/manifest.yml — extract the name component
+		dsName := path.Base(path.Dir(f.Path()))
+		result[dsName] = cats
+	}
+
+	return result, nil
+}
+
+// missingCategories returns the categories present in want but absent from have.
+// The result is sorted for deterministic error output.
+func missingCategories(want, have []string) []string {
+	var missing []string
+	for _, c := range want {
+		if !slices.Contains(have, c) {
+			missing = append(missing, c)
+		}
+	}
+	sort.Strings(missing)
+	return missing
+}
diff --git a/code/go/internal/validator/semantic/validate_policy_template_datastream_categories_test.go b/code/go/internal/validator/semantic/validate_policy_template_datastream_categories_test.go
new file mode 100644
index 000000000..d60c5ebd0
--- /dev/null
+++ b/code/go/internal/validator/semantic/validate_policy_template_datastream_categories_test.go
@@ -0,0 +1,232 @@
+// 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 writeManifest(t *testing.T, dir, content string) {
+	t.Helper()
+	err := os.WriteFile(filepath.Join(dir, "manifest.yml"), []byte(content), 0o644)
+	require.NoError(t, err)
+}
+
+func writeDataStreamManifest(t *testing.T, dir, dsName, content string) {
+	t.Helper()
+	dsDir := filepath.Join(dir, "data_stream", dsName)
+	err := os.MkdirAll(dsDir, 0o755)
+	require.NoError(t, err)
+	err = os.WriteFile(filepath.Join(dsDir, "manifest.yml"), []byte(content), 0o644)
+	require.NoError(t, err)
+}
+
+func TestValidatePolicyTemplateDatastreamCategories(t *testing.T) {
+	cases := []struct {
+		title        string
+		setup        func(t *testing.T, dir string)
+		expectedErrs []string
+	}{
+		{
+			title: "categories match",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+      - network
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - network
+  - observability
+`)
+			},
+		},
+		{
+			title: "data stream missing policy template category",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+      - network
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - observability
+`)
+			},
+			expectedErrs: []string{`policy template "mytemplate"`, `"mylogs"`, `[network]`},
+		},
+		{
+			title: "data stream has extra categories beyond policy template",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - observability
+  - security
+`)
+			},
+		},
+		{
+			title: "data stream manifest has no categories field",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+type: logs
+streams:
+  - input: logfile
+`)
+			},
+		},
+		{
+			title: "policy template has no categories field",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - observability
+`)
+			},
+		},
+		{
+			title: "multiple data streams — one matches, one does not",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - logs_ok
+      - logs_bad
+    categories:
+      - observability
+`)
+				writeDataStreamManifest(t, dir, "logs_ok", `
+title: OK Logs
+categories:
+  - observability
+`)
+				writeDataStreamManifest(t, dir, "logs_bad", `
+title: Bad Logs
+categories:
+  - security
+`)
+			},
+			expectedErrs: []string{`"logs_bad"`},
+		},
+		{
+			title: "multiple policy templates — one mismatch",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: integration
+policy_templates:
+  - name: template_a
+    data_streams:
+      - ds_a
+    categories:
+      - observability
+  - name: template_b
+    data_streams:
+      - ds_b
+    categories:
+      - network
+`)
+				writeDataStreamManifest(t, dir, "ds_a", `
+title: DS A
+categories:
+  - security
+`)
+				writeDataStreamManifest(t, dir, "ds_b", `
+title: DS B
+categories:
+  - network
+`)
+			},
+			expectedErrs: []string{`"template_a"`, `"ds_a"`},
+		},
+		{
+			title: "non-integration packages are skipped",
+			setup: func(t *testing.T, dir string) {
+				writeManifest(t, dir, `
+type: input
+policy_templates:
+  - name: mytemplate
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+`)
+				writeDataStreamManifest(t, dir, "mylogs", `
+title: My Logs
+categories:
+  - security
+`)
+			},
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.title, func(t *testing.T) {
+			dir := t.TempDir()
+			tc.setup(t, dir)
+
+			errs := ValidatePolicyTemplateDatastreamCategories(fspath.DirFS(dir))
+
+			if len(tc.expectedErrs) == 0 {
+				assert.Empty(t, errs)
+			} else {
+				require.Len(t, errs, 1)
+				for _, expected := range tc.expectedErrs {
+					assert.Contains(t, errs[0].Error(), expected)
+				}
+			}
+		})
+	}
+}
diff --git a/code/go/internal/validator/spec.go b/code/go/internal/validator/spec.go
index 68478a642..884efc095 100644
--- a/code/go/internal/validator/spec.go
+++ b/code/go/internal/validator/spec.go
@@ -234,6 +234,8 @@ func (s Spec) rules(pkgType string, rootSpec spectypes.ItemSpec) validationRules
 		{fn: semantic.ValidateFleetReservedVars, types: []string{"integration", "input"}, since: semver.MustParse("3.6.1")},
 		{fn: semantic.ValidateMinimumAgentVersion},
 		{fn: semantic.ValidateIntegrationPolicyTemplates, types: []string{"integration"}},
+		{fn: semantic.ValidatePolicyTemplateDatastreamCategories, types: []string{"integration"}},
+		{fn: semantic.ValidateDatastreamPackageCategories, types: []string{"integration"}},
 		{fn: semantic.ValidatePipelineTags, types: []string{"integration"}, since: semver.MustParse("3.6.0")},
 		{fn: semantic.ValidateStaticHandlebarsFiles, types: []string{"integration", "input"}},
 		{fn: semantic.ValidateKibanaTagDuplicates},
diff --git a/code/go/pkg/validator/validator_test.go b/code/go/pkg/validator/validator_test.go
index 3201375dd..a0cbd873a 100644
--- a/code/go/pkg/validator/validator_test.go
+++ b/code/go/pkg/validator/validator_test.go
@@ -51,6 +51,8 @@ func TestValidateFile(t *testing.T) {
 		"good_alert_rule_templates":              {},
 		"good_requires":                          {},
 		"good_package_reference_policy_template": {},
+		"good_datastream_categories_match":       {},
+		"good_datastream_package_categories":     {},
 		"deploy_custom_agent":                    {},
 		"deploy_custom_agent_multi_services":     {},
 		"deploy_docker":                          {},
@@ -147,6 +149,18 @@ func TestValidateFile(t *testing.T) {
 				"field (root): Additional property release is not allowed",
 			},
 		},
+		"bad_datastream_categories_mismatch": {
+			"manifest.yml",
+			[]string{
+				fmt.Sprintf(`data stream "mylogs" manifest categories [security] are missing policy template "mytemplate" categories [observability] (defined in "%sbad_datastream_categories_mismatch/data_stream/mylogs/manifest.yml")`, osTestBasePath),
+			},
+		},
+		"bad_datastream_package_categories": {
+			"manifest.yml",
+			[]string{
+				fmt.Sprintf(`package manifest categories [observability] are missing parent categories [security] from data stream "mylogs" (defined in "%sbad_datastream_package_categories/data_stream/mylogs/manifest.yml")`, osTestBasePath),
+			},
+		},
 		"bad_custom_ilm_policy": {
 			"data_stream/test/manifest.yml",
 			[]string{
diff --git a/spec/changelog.yml b/spec/changelog.yml
index 78fd45666..4e5fd37e3 100644
--- a/spec/changelog.yml
+++ b/spec/changelog.yml
@@ -35,6 +35,9 @@
     - description: Add `show_divider` optional boolean to input objects to suppress horizontal dividers in the Fleet UI.
       type: enhancement
       link: https://github.com/elastic/package-spec/pull/1133
+    - description: Add semantic validator to verify data stream manifest categories match policy template categories.
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1095
 - version: 3.6.0
   changes:
     - description: Add pipeline tag validations.
diff --git a/test/packages/bad_datastream_categories_mismatch/changelog.yml b/test/packages/bad_datastream_categories_mismatch/changelog.yml
new file mode 100644
index 000000000..27e62eecd
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/changelog.yml
@@ -0,0 +1,5 @@
+- version: 0.0.1
+  changes:
+    - description: Initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1
diff --git a/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/agent/stream/stream.yml.hbs b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..767dd8f8a
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{{fields "stream"}}
\ No newline at end of file
diff --git a/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/fields/fields.yml b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/fields/fields.yml
new file mode 100644
index 000000000..7c798f453
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/fields/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_datastream_categories_mismatch/data_stream/mylogs/manifest.yml b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/manifest.yml
new file mode 100644
index 000000000..9b8b61a92
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/data_stream/mylogs/manifest.yml
@@ -0,0 +1,8 @@
+title: My logs
+type: logs
+categories:
+  - security
+streams:
+  - input: logfile
+    title: My logs
+    description: Collect my logs
diff --git a/test/packages/bad_datastream_categories_mismatch/docs/README.md b/test/packages/bad_datastream_categories_mismatch/docs/README.md
new file mode 100644
index 000000000..5f19b37d5
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/docs/README.md
@@ -0,0 +1 @@
+# Test package
\ No newline at end of file
diff --git a/test/packages/bad_datastream_categories_mismatch/manifest.yml b/test/packages/bad_datastream_categories_mismatch/manifest.yml
new file mode 100644
index 000000000..289861c1b
--- /dev/null
+++ b/test/packages/bad_datastream_categories_mismatch/manifest.yml
@@ -0,0 +1,28 @@
+format_version: 3.6.0
+name: bad_datastream_categories_mismatch
+title: Bad package with mismatched datastream categories
+description: Policy template categories do not match the data stream manifest categories.
+version: 0.0.1
+type: integration
+categories:
+  - security
+source:
+  license: "Apache-2.0"
+conditions:
+  kibana:
+    version: '^8.0.0'
+policy_templates:
+  - name: mytemplate
+    title: My template
+    description: Collect my logs
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+    inputs:
+      - type: logfile
+        title: Collect my logs
+        description: Collecting logs via log input
+owner:
+  github: elastic/foobar
+  type: community
diff --git a/test/packages/bad_datastream_package_categories/changelog.yml b/test/packages/bad_datastream_package_categories/changelog.yml
new file mode 100644
index 000000000..27e62eecd
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/changelog.yml
@@ -0,0 +1,5 @@
+- version: 0.0.1
+  changes:
+    - description: Initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1
diff --git a/test/packages/bad_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs b/test/packages/bad_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..ad7430eeb
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{{fields "stream"}}
diff --git a/test/packages/bad_datastream_package_categories/data_stream/mylogs/fields/fields.yml b/test/packages/bad_datastream_package_categories/data_stream/mylogs/fields/fields.yml
new file mode 100644
index 000000000..7c798f453
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/data_stream/mylogs/fields/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_datastream_package_categories/data_stream/mylogs/manifest.yml b/test/packages/bad_datastream_package_categories/data_stream/mylogs/manifest.yml
new file mode 100644
index 000000000..9b8b61a92
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/data_stream/mylogs/manifest.yml
@@ -0,0 +1,8 @@
+title: My logs
+type: logs
+categories:
+  - security
+streams:
+  - input: logfile
+    title: My logs
+    description: Collect my logs
diff --git a/test/packages/bad_datastream_package_categories/docs/README.md b/test/packages/bad_datastream_package_categories/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/bad_datastream_package_categories/manifest.yml b/test/packages/bad_datastream_package_categories/manifest.yml
new file mode 100644
index 000000000..87f63b1aa
--- /dev/null
+++ b/test/packages/bad_datastream_package_categories/manifest.yml
@@ -0,0 +1,26 @@
+format_version: 3.6.0
+name: bad_datastream_package_categories
+title: Bad package where package manifest is missing datastream parent categories
+description: Package manifest categories do not include all parent-level categories present in the data stream manifests.
+version: 0.0.1
+type: integration
+categories:
+  - observability
+source:
+  license: "Apache-2.0"
+conditions:
+  kibana:
+    version: '^8.0.0'
+policy_templates:
+  - name: mytemplate
+    title: My template
+    description: Collect my logs
+    data_streams:
+      - mylogs
+    inputs:
+      - type: logfile
+        title: Collect my logs
+        description: Collecting logs via log input
+owner:
+  github: elastic/foobar
+  type: community
diff --git a/test/packages/good_datastream_categories_match/changelog.yml b/test/packages/good_datastream_categories_match/changelog.yml
new file mode 100644
index 000000000..27e62eecd
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/changelog.yml
@@ -0,0 +1,5 @@
+- version: 0.0.1
+  changes:
+    - description: Initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1
diff --git a/test/packages/good_datastream_categories_match/data_stream/mylogs/agent/stream/stream.yml.hbs b/test/packages/good_datastream_categories_match/data_stream/mylogs/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..767dd8f8a
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/data_stream/mylogs/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{{fields "stream"}}
\ No newline at end of file
diff --git a/test/packages/good_datastream_categories_match/data_stream/mylogs/fields/fields.yml b/test/packages/good_datastream_categories_match/data_stream/mylogs/fields/fields.yml
new file mode 100644
index 000000000..7c798f453
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/data_stream/mylogs/fields/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_datastream_categories_match/data_stream/mylogs/manifest.yml b/test/packages/good_datastream_categories_match/data_stream/mylogs/manifest.yml
new file mode 100644
index 000000000..7c9fc4396
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/data_stream/mylogs/manifest.yml
@@ -0,0 +1,8 @@
+title: My logs
+type: logs
+categories:
+  - observability
+streams:
+  - input: logfile
+    title: My logs
+    description: Collect my logs
diff --git a/test/packages/good_datastream_categories_match/docs/README.md b/test/packages/good_datastream_categories_match/docs/README.md
new file mode 100644
index 000000000..5f19b37d5
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/docs/README.md
@@ -0,0 +1 @@
+# Test package
\ No newline at end of file
diff --git a/test/packages/good_datastream_categories_match/manifest.yml b/test/packages/good_datastream_categories_match/manifest.yml
new file mode 100644
index 000000000..b2c6eaf1a
--- /dev/null
+++ b/test/packages/good_datastream_categories_match/manifest.yml
@@ -0,0 +1,28 @@
+format_version: 3.6.0
+name: good_datastream_categories_match
+title: Good package with matching datastream categories
+description: Policy template categories match the data stream manifest categories.
+version: 0.0.1
+type: integration
+categories:
+  - observability
+source:
+  license: "Apache-2.0"
+conditions:
+  kibana:
+    version: '^8.0.0'
+policy_templates:
+  - name: mytemplate
+    title: My template
+    description: Collect my logs
+    data_streams:
+      - mylogs
+    categories:
+      - observability
+    inputs:
+      - type: logfile
+        title: Collect my logs
+        description: Collecting logs via log input
+owner:
+  github: elastic/foobar
+  type: community
diff --git a/test/packages/good_datastream_package_categories/changelog.yml b/test/packages/good_datastream_package_categories/changelog.yml
new file mode 100644
index 000000000..27e62eecd
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/changelog.yml
@@ -0,0 +1,5 @@
+- version: 0.0.1
+  changes:
+    - description: Initial release
+      type: enhancement
+      link: https://github.com/elastic/package-spec/pull/1
diff --git a/test/packages/good_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs b/test/packages/good_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs
new file mode 100644
index 000000000..ad7430eeb
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/data_stream/mylogs/agent/stream/stream.yml.hbs
@@ -0,0 +1 @@
+{{fields "stream"}}
diff --git a/test/packages/good_datastream_package_categories/data_stream/mylogs/fields/fields.yml b/test/packages/good_datastream_package_categories/data_stream/mylogs/fields/fields.yml
new file mode 100644
index 000000000..7c798f453
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/data_stream/mylogs/fields/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_datastream_package_categories/data_stream/mylogs/manifest.yml b/test/packages/good_datastream_package_categories/data_stream/mylogs/manifest.yml
new file mode 100644
index 000000000..7c9fc4396
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/data_stream/mylogs/manifest.yml
@@ -0,0 +1,8 @@
+title: My logs
+type: logs
+categories:
+  - observability
+streams:
+  - input: logfile
+    title: My logs
+    description: Collect my logs
diff --git a/test/packages/good_datastream_package_categories/docs/README.md b/test/packages/good_datastream_package_categories/docs/README.md
new file mode 100644
index 000000000..66173aec4
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/docs/README.md
@@ -0,0 +1 @@
+# Test package
diff --git a/test/packages/good_datastream_package_categories/manifest.yml b/test/packages/good_datastream_package_categories/manifest.yml
new file mode 100644
index 000000000..c38391f07
--- /dev/null
+++ b/test/packages/good_datastream_package_categories/manifest.yml
@@ -0,0 +1,26 @@
+format_version: 3.6.0
+name: good_datastream_package_categories
+title: Good package where package manifest includes all datastream parent categories
+description: Package manifest categories include all parent-level categories present in the data stream manifests.
+version: 0.0.1
+type: integration
+categories:
+  - observability
+source:
+  license: "Apache-2.0"
+conditions:
+  kibana:
+    version: '^8.0.0'
+policy_templates:
+  - name: mytemplate
+    title: My template
+    description: Collect my logs
+    data_streams:
+      - mylogs
+    inputs:
+      - type: logfile
+        title: Collect my logs
+        description: Collecting logs via log input
+owner:
+  github: elastic/foobar
+  type: community