diff --git a/oas_docs/output/kibana.serverless.yaml b/oas_docs/output/kibana.serverless.yaml
index 9e21a5cecbad3..190d4c469e59f 100644
--- a/oas_docs/output/kibana.serverless.yaml
+++ b/oas_docs/output/kibana.serverless.yaml
@@ -24519,6 +24519,8 @@ paths:
                                               type: string
                                             keep_enabled:
                                               type: boolean
+                                            migrate_from:
+                                              type: string
                                             policy_template:
                                               type: string
                                             streams:
@@ -24589,6 +24591,8 @@ paths:
                                                     type: string
                                                   keep_enabled:
                                                     type: boolean
+                                                  migrate_from:
+                                                    type: string
                                                   release:
                                                     enum:
                                                       - ga
@@ -25591,6 +25595,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -25661,6 +25667,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -26445,6 +26453,8 @@ paths:
                                               type: string
                                             keep_enabled:
                                               type: boolean
+                                            migrate_from:
+                                              type: string
                                             policy_template:
                                               type: string
                                             streams:
@@ -26515,6 +26525,8 @@ paths:
                                                     type: string
                                                   keep_enabled:
                                                     type: boolean
+                                                  migrate_from:
+                                                    type: string
                                                   release:
                                                     enum:
                                                       - ga
@@ -27278,6 +27290,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -27348,6 +27362,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -28349,6 +28365,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -28419,6 +28437,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -29295,6 +29315,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -29365,6 +29387,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -31621,6 +31645,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -31691,6 +31717,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -48181,6 +48209,8 @@ paths:
                                     type: string
                                   keep_enabled:
                                     type: boolean
+                                  migrate_from:
+                                    type: string
                                   policy_template:
                                     type: string
                                   streams:
@@ -48251,6 +48281,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         release:
                                           enum:
                                             - ga
@@ -48721,6 +48753,8 @@ paths:
                             type: string
                           keep_enabled:
                             type: boolean
+                          migrate_from:
+                            type: string
                           policy_template:
                             type: string
                           streams:
@@ -48791,6 +48825,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 release:
                                   enum:
                                     - ga
@@ -49283,6 +49319,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -49353,6 +49391,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -49874,6 +49914,8 @@ paths:
                                     type: string
                                   keep_enabled:
                                     type: boolean
+                                  migrate_from:
+                                    type: string
                                   policy_template:
                                     type: string
                                   streams:
@@ -49944,6 +49986,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         release:
                                           enum:
                                             - ga
@@ -50514,6 +50558,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -50584,6 +50630,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -51058,6 +51106,8 @@ paths:
                             type: string
                           keep_enabled:
                             type: boolean
+                          migrate_from:
+                            type: string
                           policy_template:
                             type: string
                           streams:
@@ -51128,6 +51178,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 release:
                                   enum:
                                     - ga
@@ -51616,6 +51668,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -51686,6 +51740,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -52551,6 +52607,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         policy_template:
                                           type: string
                                         streams:
@@ -52621,6 +52679,8 @@ paths:
                                                 type: string
                                               keep_enabled:
                                                 type: boolean
+                                              migrate_from:
+                                                type: string
                                               release:
                                                 enum:
                                                   - ga
@@ -53049,6 +53109,8 @@ paths:
                                       type: string
                                     keep_enabled:
                                       type: boolean
+                                    migrate_from:
+                                      type: string
                                     policy_template:
                                       type: string
                                     streams:
@@ -53119,6 +53181,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           release:
                                             enum:
                                               - ga
diff --git a/oas_docs/output/kibana.yaml b/oas_docs/output/kibana.yaml
index d9a26a03a9d1f..e0d802f276c82 100644
--- a/oas_docs/output/kibana.yaml
+++ b/oas_docs/output/kibana.yaml
@@ -27098,6 +27098,8 @@ paths:
                                               type: string
                                             keep_enabled:
                                               type: boolean
+                                            migrate_from:
+                                              type: string
                                             policy_template:
                                               type: string
                                             streams:
@@ -27168,6 +27170,8 @@ paths:
                                                     type: string
                                                   keep_enabled:
                                                     type: boolean
+                                                  migrate_from:
+                                                    type: string
                                                   release:
                                                     enum:
                                                       - ga
@@ -28170,6 +28174,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -28240,6 +28246,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -29024,6 +29032,8 @@ paths:
                                               type: string
                                             keep_enabled:
                                               type: boolean
+                                            migrate_from:
+                                              type: string
                                             policy_template:
                                               type: string
                                             streams:
@@ -29094,6 +29104,8 @@ paths:
                                                     type: string
                                                   keep_enabled:
                                                     type: boolean
+                                                  migrate_from:
+                                                    type: string
                                                   release:
                                                     enum:
                                                       - ga
@@ -29857,6 +29869,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -29927,6 +29941,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -30928,6 +30944,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -30998,6 +31016,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -31874,6 +31894,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           policy_template:
                                             type: string
                                           streams:
@@ -31944,6 +31966,8 @@ paths:
                                                   type: string
                                                 keep_enabled:
                                                   type: boolean
+                                                migrate_from:
+                                                  type: string
                                                 release:
                                                   enum:
                                                     - ga
@@ -34200,6 +34224,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -34270,6 +34296,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -50760,6 +50788,8 @@ paths:
                                     type: string
                                   keep_enabled:
                                     type: boolean
+                                  migrate_from:
+                                    type: string
                                   policy_template:
                                     type: string
                                   streams:
@@ -50830,6 +50860,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         release:
                                           enum:
                                             - ga
@@ -51300,6 +51332,8 @@ paths:
                             type: string
                           keep_enabled:
                             type: boolean
+                          migrate_from:
+                            type: string
                           policy_template:
                             type: string
                           streams:
@@ -51370,6 +51404,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 release:
                                   enum:
                                     - ga
@@ -51862,6 +51898,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -51932,6 +51970,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -52453,6 +52493,8 @@ paths:
                                     type: string
                                   keep_enabled:
                                     type: boolean
+                                  migrate_from:
+                                    type: string
                                   policy_template:
                                     type: string
                                   streams:
@@ -52523,6 +52565,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         release:
                                           enum:
                                             - ga
@@ -53093,6 +53137,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -53163,6 +53209,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -53637,6 +53685,8 @@ paths:
                             type: string
                           keep_enabled:
                             type: boolean
+                          migrate_from:
+                            type: string
                           policy_template:
                             type: string
                           streams:
@@ -53707,6 +53757,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 release:
                                   enum:
                                     - ga
@@ -54195,6 +54247,8 @@ paths:
                                   type: string
                                 keep_enabled:
                                   type: boolean
+                                migrate_from:
+                                  type: string
                                 policy_template:
                                   type: string
                                 streams:
@@ -54265,6 +54319,8 @@ paths:
                                         type: string
                                       keep_enabled:
                                         type: boolean
+                                      migrate_from:
+                                        type: string
                                       release:
                                         enum:
                                           - ga
@@ -55130,6 +55186,8 @@ paths:
                                           type: string
                                         keep_enabled:
                                           type: boolean
+                                        migrate_from:
+                                          type: string
                                         policy_template:
                                           type: string
                                         streams:
@@ -55200,6 +55258,8 @@ paths:
                                                 type: string
                                               keep_enabled:
                                                 type: boolean
+                                              migrate_from:
+                                                type: string
                                               release:
                                                 enum:
                                                   - ga
@@ -55628,6 +55688,8 @@ paths:
                                       type: string
                                     keep_enabled:
                                       type: boolean
+                                    migrate_from:
+                                      type: string
                                     policy_template:
                                       type: string
                                     streams:
@@ -55698,6 +55760,8 @@ paths:
                                             type: string
                                           keep_enabled:
                                             type: boolean
+                                          migrate_from:
+                                            type: string
                                           release:
                                             enum:
                                               - ga
diff --git a/x-pack/platform/plugins/shared/fleet/common/services/package_to_package_policy.ts b/x-pack/platform/plugins/shared/fleet/common/services/package_to_package_policy.ts
index e96bf92ae4196..3dff0bf9c4a0a 100644
--- a/x-pack/platform/plugins/shared/fleet/common/services/package_to_package_policy.ts
+++ b/x-pack/platform/plugins/shared/fleet/common/services/package_to_package_policy.ts
@@ -141,6 +141,7 @@ export const packageToPackagePolicyInputs = (
         // disable deprecated streams on new installations
         enabled: packageStream.deprecated ? false : packageStream.enabled !== false,
         data_stream: packageStream.data_stream,
+        ...(packageStream.migrate_from ? { migrate_from: packageStream.migrate_from } : {}),
       };
       if (packageStream.vars && packageStream.vars.length) {
         stream.vars = packageStream.vars.reduce(varsReducer, {});
@@ -180,6 +181,7 @@ export const packageToPackagePolicyInputs = (
       policy_template: packageInput.policy_template,
       enabled: enableInput,
       streams: streamsForInput,
+      ...(packageInput.migrate_from ? { migrate_from: packageInput.migrate_from } : {}),
     };
 
     if (Object.keys(varsForInput).length) {
diff --git a/x-pack/platform/plugins/shared/fleet/common/types/models/epm.ts b/x-pack/platform/plugins/shared/fleet/common/types/models/epm.ts
index 2be9db783d517..747e2f130dda1 100644
--- a/x-pack/platform/plugins/shared/fleet/common/types/models/epm.ts
+++ b/x-pack/platform/plugins/shared/fleet/common/types/models/epm.ts
@@ -315,6 +315,7 @@ export enum RegistryInputKeys {
   deployment_modes = 'deployment_modes',
   hide_in_var_group_options = 'hide_in_var_group_options',
   deprecated = 'deprecated',
+  migrate_from = 'migrate_from',
 }
 
 export type RegistryInputGroup = 'logs' | 'metrics';
@@ -331,6 +332,7 @@ export interface RegistryInput {
   [RegistryInputKeys.deployment_modes]?: string[];
   [RegistryInputKeys.hide_in_var_group_options]?: Record<string, string[]>;
   [RegistryInputKeys.deprecated]?: DeprecationInfo;
+  [RegistryInputKeys.migrate_from]?: string;
 }
 
 export enum RegistryStreamKeys {
@@ -344,6 +346,7 @@ export enum RegistryStreamKeys {
   ingestion_method = 'ingestion_method',
   var_groups = 'var_groups',
   deprecated = 'deprecated',
+  migrate_from = 'migrate_from',
 }
 
 export interface RegistryStream {
@@ -357,6 +360,7 @@ export interface RegistryStream {
   [RegistryStreamKeys.ingestion_method]?: string;
   [RegistryStreamKeys.var_groups]?: RegistryVarGroup[];
   [RegistryStreamKeys.deprecated]?: DeprecationInfo;
+  [RegistryStreamKeys.migrate_from]?: string;
 }
 
 export type RegistryStreamWithDataStream = RegistryStream & { data_stream: RegistryDataStream };
diff --git a/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy.ts b/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy.ts
index a19404b6626da..336ded10f5fbd 100644
--- a/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy.ts
+++ b/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy.ts
@@ -51,6 +51,7 @@ export interface NewPackagePolicyInputStream {
   vars?: PackagePolicyConfigRecord;
   var_group_selections?: Record<string, string>;
   config?: PackagePolicyConfigRecord;
+  migrate_from?: string;
 }
 
 export interface PackagePolicyInputStream extends NewPackagePolicyInputStream {
@@ -68,6 +69,7 @@ export interface NewPackagePolicyInput {
   config?: PackagePolicyConfigRecord;
   streams: NewPackagePolicyInputStream[];
   deprecated?: DeprecationInfo;
+  migrate_from?: string;
 }
 
 export interface PackagePolicyInput extends Omit<NewPackagePolicyInput, 'streams'> {
diff --git a/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy_schema.ts b/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy_schema.ts
index 9668d27b37e38..8866e5f0f004d 100644
--- a/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy_schema.ts
+++ b/x-pack/platform/plugins/shared/fleet/common/types/models/package_policy_schema.ts
@@ -90,6 +90,7 @@ const PackagePolicyStreamsSchema = {
   config: schema.maybe(ConfigRecordSchema),
   compiled_stream: schema.maybe(schema.any()),
   deprecated: schema.maybe(DeprecationInfoSchema),
+  migrate_from: schema.maybe(schema.string()),
 };
 
 export const PackagePolicyInputsSchema = {
@@ -102,6 +103,7 @@ export const PackagePolicyInputsSchema = {
   config: schema.maybe(ConfigRecordSchema),
   streams: schema.arrayOf(schema.object(PackagePolicyStreamsSchema), { maxSize: 100 }),
   deprecated: schema.maybe(DeprecationInfoSchema),
+  migrate_from: schema.maybe(schema.string()),
 };
 
 export const ExperimentalDataStreamFeaturesSchema = schema.arrayOf(
diff --git a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_panel.tsx b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_panel.tsx
index 84ec826845812..fe2e5164ef4c7 100644
--- a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_panel.tsx
+++ b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_panel.tsx
@@ -76,6 +76,32 @@ export const shouldShowStreamsByDefault = (
   );
 };
 
+export const MigrationTooltip = ({
+  migrateFrom,
+  isStream = false,
+}: {
+  migrateFrom: string;
+  isStream?: boolean;
+}) => (
+  <EuiFlexItem grow={false}>
+    <EuiIconTip
+      type="info"
+      color="subdued"
+      content={i18n.translate(
+        isStream
+          ? 'xpack.fleet.createPackagePolicy.stepConfigure.streamMigratedTooltip'
+          : 'xpack.fleet.createPackagePolicy.stepConfigure.inputMigratedTooltip',
+        {
+          defaultMessage: isStream
+            ? 'This data stream was automatically migrated from {migrateFrom}.'
+            : 'This input was automatically migrated from {migrateFrom}.',
+          values: { migrateFrom },
+        }
+      )}
+    />
+  </EuiFlexItem>
+);
+
 export const PackagePolicyInputPanel: React.FunctionComponent<{
   packageInput: RegistryInput;
   packageInfo: PackageInfo;
@@ -86,6 +112,7 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
   forceShowErrors?: boolean;
   isSingleInputAndStreams?: boolean;
   isEditPage?: boolean;
+  isUpgrade?: boolean;
   varGroupSelections?: Record<string, string>;
 }> = memo(
   ({
@@ -98,6 +125,7 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
     forceShowErrors,
     isSingleInputAndStreams = false,
     isEditPage = false,
+    isUpgrade = false,
     varGroupSelections = {},
   }) => {
     const theme = useEuiTheme();
@@ -224,12 +252,12 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
     );
 
     const allStreamsDeprecated = useMemo(
-      () => packageInputStreams.length > 0 && packageInputStreams.every((s) => !!s.deprecated),
-      [packageInputStreams]
+      () => inputStreams.length > 0 && inputStreams.every((s) => !!s.packageInputStream.deprecated),
+      [inputStreams]
     );
     const deprecationInfo =
       packagePolicyInput.deprecated ||
-      (allStreamsDeprecated ? packageInputStreams[0].deprecated : undefined);
+      (allStreamsDeprecated ? inputStreams[0].packageInputStream.deprecated : undefined);
     const isDeprecatedInput = !!deprecationInfo;
     const deprecatedInputTooltip = deprecationInfo
       ? deprecationInfo.replaced_by
@@ -262,6 +290,9 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
     if (!isEditPage && (isDeprecatedInput || allStreamsDeprecated)) {
       return null;
     }
+    const migrationTooltip = isUpgrade && packagePolicyInput.migrate_from && !isDeprecatedInput && (
+      <MigrationTooltip migrateFrom={packagePolicyInput.migrate_from} />
+    );
 
     return (
       <>
@@ -269,16 +300,21 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
         <EuiFlexGroup justifyContent="spaceBetween" alignItems="center">
           {isSingleInputAndStreams ? (
             <EuiFlexItem grow={false}>
-              <EuiTitle size="xs">
-                <h3
-                  data-test-subj="PackagePolicy.InputStreamConfig.title"
-                  style={
-                    isDeprecatedInput ? { color: theme.euiTheme.colors.textSubdued } : undefined
-                  }
-                >
-                  {packageInput.title || packageInput.type}
-                </h3>
-              </EuiTitle>
+              <EuiFlexGroup alignItems="center" gutterSize="s">
+                <EuiFlexItem grow={false}>
+                  <EuiTitle size="xs">
+                    <h3
+                      data-test-subj="PackagePolicy.InputStreamConfig.title"
+                      style={
+                        isDeprecatedInput ? { color: theme.euiTheme.colors.textSubdued } : undefined
+                      }
+                    >
+                      {packageInput.title || packageInput.type}
+                    </h3>
+                  </EuiTitle>
+                </EuiFlexItem>
+                {migrationTooltip}
+              </EuiFlexGroup>
               <EuiSpacer size="s" />
               {showTopLevelDescription && topLevelDescription}
             </EuiFlexItem>
@@ -302,6 +338,7 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
                         </h3>
                       </EuiTitle>
                     </EuiFlexItem>
+                    {migrationTooltip}
                     {isDeprecatedInput && (
                       <EuiFlexItem grow={false}>
                         <span data-test-subj="PackagePolicy.InputStreamConfig.deprecatedIcon">
@@ -436,6 +473,7 @@ export const PackagePolicyInputPanel: React.FunctionComponent<{
                   totalStreams={inputStreams.length}
                   packagePolicyInputStream={packagePolicyInputStream!}
                   showDescriptionColumn={!isSingleInputAndStreams}
+                  isUpgrade={isUpgrade}
                   updatePackagePolicyInputStream={(
                     updatedStream: Partial<PackagePolicyInputStream>
                   ) => {
diff --git a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_stream.tsx b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_stream.tsx
index 889b37a40ec67..a7d20d4729aab 100644
--- a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_stream.tsx
+++ b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/components/package_policy_input_stream.tsx
@@ -67,6 +67,7 @@ import { PackagePolicyInputVarField } from './package_policy_input_var_field';
 import { useDataStreamId, useVarGroupSelections } from './hooks';
 import { sortDatastreamsByDataset } from './sort_datastreams';
 import { VarGroupSelector } from './var_group_selector';
+import { MigrationTooltip } from './package_policy_input_panel';
 
 const ScrollAnchor = styled.div`
   display: none;
@@ -81,6 +82,7 @@ interface Props {
   inputStreamValidationResults: PackagePolicyConfigValidationResults;
   forceShowErrors?: boolean;
   isEditPage?: boolean;
+  isUpgrade?: boolean;
   totalStreams?: number;
   showDescriptionColumn?: boolean;
   varGroupSelections?: Record<string, string>;
@@ -95,6 +97,7 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
     inputStreamValidationResults,
     forceShowErrors,
     isEditPage,
+    isUpgrade,
     totalStreams,
     showDescriptionColumn = true,
     varGroupSelections = {},
@@ -308,6 +311,14 @@ export const PackagePolicyInputStreamConfig = memo<Props>(
                             </span>
                           </EuiFlexItem>
                         )}
+                        {isUpgrade &&
+                          packagePolicyInputStream.migrate_from &&
+                          !showStreamDeprecationIcon && (
+                            <MigrationTooltip
+                              migrateFrom={packagePolicyInputStream.migrate_from}
+                              isStream
+                            />
+                          )}
                       </EuiFlexGroup>
                     </EuiFlexItem>
                   )}
diff --git a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/step_configure_package.tsx b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/step_configure_package.tsx
index 9ade6bf797ca0..14f8c95b4be8d 100644
--- a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/step_configure_package.tsx
+++ b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/create_package_policy_page/components/steps/step_configure_package.tsx
@@ -48,6 +48,7 @@ export const StepConfigurePackagePolicy: React.FunctionComponent<{
   submitAttempted: boolean;
   noTopRule?: boolean;
   isEditPage?: boolean;
+  isUpgrade?: boolean;
   isAgentlessSelected?: boolean;
   varGroupSelections?: VarGroupSelection;
 }> = ({
@@ -59,6 +60,7 @@ export const StepConfigurePackagePolicy: React.FunctionComponent<{
   submitAttempted,
   noTopRule = false,
   isEditPage = false,
+  isUpgrade = false,
   isAgentlessSelected = false,
   varGroupSelections = {},
 }) => {
@@ -199,6 +201,7 @@ export const StepConfigurePackagePolicy: React.FunctionComponent<{
                         }
                         forceShowErrors={submitAttempted}
                         isEditPage={isEditPage}
+                        isUpgrade={isUpgrade}
                         varGroupSelections={varGroupSelections}
                       />
                       <EuiHorizontalRule margin="m" />
diff --git a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/components/upgrade.tsx b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/components/upgrade.tsx
index 850662d30ecf4..a65ca9171d64b 100644
--- a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/components/upgrade.tsx
+++ b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/components/upgrade.tsx
@@ -150,6 +150,21 @@ const ReadyToUpgradeCallOut = ({
   );
 };
 
+const InputMigrationCallout = () => (
+  <EuiCallOut
+    title={i18n.translate('xpack.fleet.upgradePackagePolicy.statusCallOut.inputMigrationTitle', {
+      defaultMessage: 'Input type migration',
+    })}
+    color="primary"
+    iconType="info"
+  >
+    <FormattedMessage
+      id="xpack.fleet.upgradePackagePolicy.statusCallout.inputMigrationContent"
+      defaultMessage="This upgrade replaces the input used for data collection. Your existing configuration has been automatically carried over."
+    />
+  </EuiCallOut>
+);
+
 export const UpgradeStatusCallout: React.FunctionComponent<{
   dryRunData: UpgradePackagePolicyDryRunResponse;
   newSecrets: RegistryVarsEntry[];
@@ -164,6 +179,10 @@ export const UpgradeStatusCallout: React.FunctionComponent<{
   const [currentPackagePolicy, proposedUpgradePackagePolicy] = dryRunData[0].diff || [];
   const isReadyForUpgrade = currentPackagePolicy && !dryRunData[0].hasErrors;
 
+  const hasMigratedInputs = (proposedUpgradePackagePolicy?.inputs ?? []).some(
+    (input) => !!input.migrate_from
+  );
+
   return (
     <>
       {isPreviousVersionFlyoutOpen && currentPackagePolicy && (
@@ -171,6 +190,7 @@ export const UpgradeStatusCallout: React.FunctionComponent<{
           <EuiFlyout
             onClose={() => setIsPreviousVersionFlyoutOpen(false)}
             maxWidth={MAX_FLYOUT_WIDTH}
+            aria-label="Previous version configuration flyout"
           >
             <EuiFlyoutHeader hasBorder>
               <EuiTitle size="m">
@@ -210,6 +230,12 @@ export const UpgradeStatusCallout: React.FunctionComponent<{
           <HasNewSecretsCallOut newSecrets={newSecrets} />
         </>
       )}
+      {hasMigratedInputs && (
+        <>
+          <EuiSpacer size="m" />
+          <InputMigrationCallout />
+        </>
+      )}
     </>
   );
 };
diff --git a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/index.tsx b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/index.tsx
index b9a7d508240a8..85a50f9450b30 100644
--- a/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/index.tsx
+++ b/x-pack/platform/plugins/shared/fleet/public/applications/fleet/sections/agent_policy/edit_package_policy_page/index.tsx
@@ -474,6 +474,7 @@ export const EditPackagePolicyForm = memo<{
               validationResults={validationResults}
               submitAttempted={formState === 'INVALID'}
               isEditPage={true}
+              isUpgrade={isUpgrade}
               isAgentlessSelected={hasAgentlessAgentPolicy}
               varGroupSelections={varGroupSelections}
             />
@@ -513,6 +514,7 @@ export const EditPackagePolicyForm = memo<{
       selectedTab,
       tabsViews,
       updatePackagePolicy,
+      isUpgrade,
       validationResults,
       varGroupSelections,
     ]
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/package_policy.test.ts b/x-pack/platform/plugins/shared/fleet/server/services/package_policy.test.ts
index fa31fdfd6f89c..1f3d2ed3885e1 100644
--- a/x-pack/platform/plugins/shared/fleet/server/services/package_policy.test.ts
+++ b/x-pack/platform/plugins/shared/fleet/server/services/package_policy.test.ts
@@ -8047,9 +8047,1688 @@ describe('Package policy service', () => {
         expect(disabledInputs[0]?.policy_template).toBe('gmail');
       });
     });
+
+    describe('when input has migrate_from', () => {
+      const makeBasePolicy = (overrides?: Partial<NewPackagePolicyInput>): NewPackagePolicy => ({
+        name: 'base-package-policy',
+        description: 'Base Package Policy',
+        namespace: 'default',
+        enabled: true,
+        policy_id: 'xxxx',
+        policy_ids: ['xxxx'],
+        package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+        inputs: [
+          {
+            type: 'httpjson',
+            policy_template: 'template_1',
+            enabled: true,
+            vars: {
+              url: { type: 'text', value: 'http://example.com' },
+              interval: { type: 'text', value: '10s' },
+            },
+            streams: [
+              {
+                enabled: true,
+                data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                vars: {
+                  tags: { type: 'text', value: 'httpjson-tag' },
+                  stale_var: { type: 'text', value: 'should-be-removed' },
+                },
+              },
+            ],
+            ...overrides,
+          },
+        ],
+      });
+
+      const makeCelPackageInfo = (extraInputProps?: Record<string, unknown>): PackageInfo =>
+        ({
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '0.0.2',
+          latestVersion: '0.0.2',
+          release: 'experimental',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  migrate_from: 'httpjson',
+                  vars: [
+                    { name: 'url', type: 'text' },
+                    { name: 'interval', type: 'text' },
+                  ],
+                  ...extraInputProps,
+                },
+              ],
+            },
+          ],
+          assets: {},
+        } as unknown as PackageInfo);
+
+      const makeCelInputsOverride = (extraProps?: Record<string, unknown>): InputsOverride[] => [
+        {
+          type: 'cel',
+          policy_template: 'template_1',
+          enabled: false,
+          migrate_from: 'httpjson',
+          vars: {
+            url: { type: 'text', value: 'http://new-default.com' },
+            interval: { type: 'text', value: '30s' },
+          },
+          streams: [
+            {
+              enabled: true,
+              data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+              vars: {
+                tags: { type: 'text', value: 'cel-default-tag' },
+              },
+            },
+          ],
+          ...extraProps,
+        } as unknown as InputsOverride,
+      ];
+
+      it('carries input-level vars from the old input type to the new one', () => {
+        const result = updatePackageInputs(
+          makeBasePolicy(),
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput).toBeDefined();
+        // User-configured value from httpjson should override the new package default
+        expect(celInput?.vars?.url.value).toBe('http://example.com');
+        expect(celInput?.vars?.interval.value).toBe('10s');
+      });
+
+      it('enables the new input when migration succeeds and the old input was enabled', () => {
+        const result = updatePackageInputs(
+          makeBasePolicy(), // httpjson input has enabled: true
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput?.enabled).toBe(true);
+      });
+
+      it('keeps the new input disabled when migration succeeds but the old input was disabled', () => {
+        const result = updatePackageInputs(
+          makeBasePolicy({ enabled: false }), // httpjson input disabled by the user
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput?.enabled).toBe(false);
+      });
+
+      it('removes the old input type from the policy after migration', () => {
+        const result = updatePackageInputs(
+          makeBasePolicy(),
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const httpjsonInput = result.inputs.find((i) => i.type === 'httpjson');
+        expect(httpjsonInput).toBeUndefined();
+      });
+
+      it('carries stream-level vars by position from old streams to new streams', () => {
+        const result = updatePackageInputs(
+          makeBasePolicy(),
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        // The new stream's dataset should be from the new package
+        expect(celInput?.streams[0]?.data_stream.dataset).toBe('test_package.cel_log');
+        // But the var value should come from the old httpjson stream
+        expect(celInput?.streams[0]?.vars?.tags?.value).toBe('httpjson-tag');
+        // Vars not in the new stream template should be removed
+        expect(celInput?.streams[0]?.vars?.stale_var).toBeUndefined();
+      });
+
+      it('carries stream enabled state from old streams even when new package defaults to disabled', () => {
+        // The new CEL stream override starts with enabled: false (package default disabled)
+        const celOverrideWithDisabledStream: InputsOverride[] = [
+          {
+            ...makeCelInputsOverride()[0],
+            streams: [
+              {
+                ...makeCelInputsOverride()[0].streams![0],
+                enabled: false,
+              },
+            ],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(
+          makeBasePolicy(), // old httpjson stream has enabled: true
+          makeCelPackageInfo(),
+          celOverrideWithDisabledStream,
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        // Should carry over enabled: true from the old httpjson stream
+        expect(celInput?.streams[0]?.enabled).toBe(true);
+      });
+
+      it('falls back to new input defaults when the old input type is not found', () => {
+        const policyWithoutHttpjson: NewPackagePolicy = {
+          ...makeBasePolicy(),
+          inputs: [
+            {
+              type: 'logfile',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {},
+              streams: [],
+            },
+          ],
+        };
+
+        const result = updatePackageInputs(
+          policyWithoutHttpjson,
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput).toBeDefined();
+        // No old input found, so the new package defaults are used
+        expect(celInput?.vars?.url.value).toBe('http://new-default.com');
+        // New input should NOT be auto-enabled when no migration source found
+        expect(celInput?.enabled).toBe(false);
+      });
+
+      it('does not migrate vars or enable the new input when it is deprecated (input-level migrate_from)', () => {
+        const deprecationInfo = { description: 'Use cel instead', replaced_by: { type: 'cel' } };
+        const result = updatePackageInputs(
+          makeBasePolicy(), // httpjson input with user-configured vars
+          makeCelPackageInfo(),
+          // cel input is deprecated — migration should be skipped entirely
+          makeCelInputsOverride({ deprecated: deprecationInfo }),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput).toBeDefined();
+        // Vars should NOT be carried over from the old httpjson input
+        expect(celInput?.vars?.url.value).toBe('http://new-default.com');
+        // The new input should not have been enabled by the migration logic
+        expect(celInput?.enabled).toBe(false);
+      });
+
+      it('does not enable the new input for limited packages even when migration succeeds', () => {
+        const limitedPackageInfo = makeCelPackageInfo();
+        // Make it a limited (single-policy) package
+        (limitedPackageInfo as any).policy_templates![0].multiple = false;
+        (limitedPackageInfo as any).type = 'logrt';
+
+        const result = updatePackageInputs(
+          makeBasePolicy(),
+          limitedPackageInfo,
+          makeCelInputsOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        // Limited packages should have inputs disabled regardless of migration
+        expect(celInput?.enabled).toBe(false);
+      });
+
+      it('removes the old input from inputs even when it has no policy_template set', () => {
+        // Old input without policy_template — the initial filter would normally keep it
+        const policyWithNoPolicyTemplate = makeBasePolicy({ policy_template: undefined });
+
+        const result = updatePackageInputs(
+          policyWithNoPolicyTemplate,
+          makeCelPackageInfo(),
+          makeCelInputsOverride(),
+          false
+        );
+
+        const httpjsonInput = result.inputs.find((i) => i.type === 'httpjson');
+        expect(httpjsonInput).toBeUndefined();
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        expect(celInput).toBeDefined();
+        expect(celInput?.vars?.url.value).toBe('http://example.com');
+      });
+
+      it('migrates each cel input from its own policy_template httpjson input in a multi-template package', () => {
+        // Two policy templates each having an httpjson input that migrates to cel.
+        // The cel input for template_1 must pick up template_1's httpjson vars, and
+        // the cel input for template_2 must pick up template_2's httpjson vars.
+        const basePolicy: NewPackagePolicy = {
+          name: 'base-package-policy',
+          description: 'Base Package Policy',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                url: { type: 'text', value: 'http://template1.example.com' },
+                interval: { type: 'text', value: '10s' },
+              },
+              streams: [],
+            },
+            {
+              type: 'httpjson',
+              policy_template: 'template_2',
+              enabled: false,
+              vars: {
+                url: { type: 'text', value: 'http://template2.example.com' },
+                interval: { type: 'text', value: '30s' },
+              },
+              streams: [],
+            },
+          ],
+        };
+
+        const packageInfo: PackageInfo = {
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '0.0.2',
+          latestVersion: '0.0.2',
+          release: 'experimental',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  migrate_from: 'httpjson',
+                  vars: [
+                    { name: 'url', type: 'text' },
+                    { name: 'interval', type: 'text' },
+                  ],
+                },
+              ],
+            },
+            {
+              name: 'template_2',
+              title: 'Template 2',
+              description: 'Template 2',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  migrate_from: 'httpjson',
+                  vars: [
+                    { name: 'url', type: 'text' },
+                    { name: 'interval', type: 'text' },
+                  ],
+                },
+              ],
+            },
+          ],
+          assets: {},
+        } as unknown as PackageInfo;
+
+        const inputsOverride: InputsOverride[] = [
+          {
+            type: 'cel',
+            policy_template: 'template_1',
+            enabled: false,
+            migrate_from: 'httpjson',
+            vars: {
+              url: { type: 'text', value: 'http://new-default.com' },
+              interval: { type: 'text', value: '60s' },
+            },
+            streams: [],
+          } as unknown as InputsOverride,
+          {
+            type: 'cel',
+            policy_template: 'template_2',
+            enabled: false,
+            migrate_from: 'httpjson',
+            vars: {
+              url: { type: 'text', value: 'http://new-default.com' },
+              interval: { type: 'text', value: '60s' },
+            },
+            streams: [],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(basePolicy, packageInfo, inputsOverride, false);
+
+        // Both old httpjson inputs should have been removed
+        expect(result.inputs.filter((i) => i.type === 'httpjson')).toHaveLength(0);
+
+        const celT1 = result.inputs.find(
+          (i) => i.type === 'cel' && i.policy_template === 'template_1'
+        );
+        const celT2 = result.inputs.find(
+          (i) => i.type === 'cel' && i.policy_template === 'template_2'
+        );
+
+        // Each cel input must carry vars and enabled state from its OWN template's httpjson input
+        expect(celT1?.vars?.url.value).toBe('http://template1.example.com');
+        expect(celT1?.vars?.interval.value).toBe('10s');
+        expect(celT1?.enabled).toBe(true);
+
+        expect(celT2?.vars?.url.value).toBe('http://template2.example.com');
+        expect(celT2?.vars?.interval.value).toBe('30s');
+        expect(celT2?.enabled).toBe(false);
+      });
+
+      describe('null-value variable migration priority', () => {
+        // Build a base policy where the old httpjson input has a mix of:
+        //  - a value the user explicitly set (url)
+        //  - a bool var that was never configured (null)
+        //  - a non-bool var that was never configured (null)
+        const makeBasePolicyWithNullVars = (): NewPackagePolicy => ({
+          name: 'base-package-policy',
+          description: 'Base Package Policy',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                url: { type: 'text', value: 'http://user-set.com' },
+                enable_tracer: { type: 'bool', value: null }, // never configured
+                proxy_url: { type: 'text', value: null }, // never configured
+              },
+              streams: [],
+            },
+          ],
+        });
+
+        // New CEL package defines url, enable_tracer (bool, default false), proxy_url (no default).
+        const makeCelPackageWithNullDefaults = (): PackageInfo =>
+          ({
+            name: 'test-package',
+            description: 'Test Package',
+            title: 'Test Package',
+            version: '0.0.2',
+            latestVersion: '0.0.2',
+            release: 'experimental',
+            format_version: '1.0.0',
+            owner: { github: 'elastic/fleet' },
+            policy_templates: [
+              {
+                name: 'template_1',
+                title: 'Template 1',
+                description: 'Template 1',
+                inputs: [
+                  {
+                    type: 'cel',
+                    title: 'CEL',
+                    description: 'CEL Input',
+                    migrate_from: 'httpjson',
+                    vars: [
+                      { name: 'url', type: 'text' },
+                      { name: 'enable_tracer', type: 'bool' },
+                      { name: 'proxy_url', type: 'text' },
+                    ],
+                  },
+                ],
+              },
+            ],
+            assets: {},
+          } as unknown as PackageInfo);
+
+        const makeCelOverrideWithDefaults = (): InputsOverride[] => [
+          {
+            type: 'cel',
+            policy_template: 'template_1',
+            enabled: false,
+            migrate_from: 'httpjson',
+            vars: {
+              url: { type: 'text', value: 'http://new-default.com' },
+              enable_tracer: { type: 'bool', value: false }, // package default = false
+              proxy_url: { type: 'text', value: null }, // package has no default
+            },
+            streams: [],
+          } as unknown as InputsOverride,
+        ];
+
+        it('priority 1: preserves old value when it was explicitly set (non-null)', () => {
+          const result = updatePackageInputs(
+            makeBasePolicyWithNullVars(),
+            makeCelPackageWithNullDefaults(),
+            makeCelOverrideWithDefaults(),
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          expect(celInput?.vars?.url.value).toBe('http://user-set.com');
+        });
+
+        it('priority 2: uses new package default when old value was null', () => {
+          const result = updatePackageInputs(
+            makeBasePolicyWithNullVars(),
+            makeCelPackageWithNullDefaults(),
+            makeCelOverrideWithDefaults(),
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          // Old value was null → falls through to new package default (false)
+          expect(celInput?.vars?.enable_tracer.value).toBe(false);
+        });
+
+        it('priority 3: falls back to false for bool vars when both old and new default are null', () => {
+          // Override has null default for enable_tracer (no package default defined)
+          const overrideWithNullDefault: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              migrate_from: 'httpjson',
+              vars: {
+                url: { type: 'text', value: 'http://new-default.com' },
+                enable_tracer: { type: 'bool', value: null }, // no package default either
+                proxy_url: { type: 'text', value: null },
+              },
+              streams: [],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            makeBasePolicyWithNullVars(),
+            makeCelPackageWithNullDefaults(),
+            overrideWithNullDefault,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          // Both old and new default are null → sanitizeMigratedVars forces false for bool
+          expect(celInput?.vars?.enable_tracer.value).toBe(false);
+        });
+
+        it('leaves non-bool vars as null when neither old nor new has a value', () => {
+          const result = updatePackageInputs(
+            makeBasePolicyWithNullVars(),
+            makeCelPackageWithNullDefaults(),
+            makeCelOverrideWithDefaults(),
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          // proxy_url: old=null, new default=null, type=text → stays null (no bool fallback)
+          expect(celInput?.vars?.proxy_url.value).toBeNull();
+        });
+
+        it('applies same null-value priority to stream-level vars during migration', () => {
+          const baseWithNullStreamVars: NewPackagePolicy = {
+            name: 'base-package-policy',
+            description: 'Base Package Policy',
+            namespace: 'default',
+            enabled: true,
+            policy_id: 'xxxx',
+            policy_ids: ['xxxx'],
+            package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+            inputs: [
+              {
+                type: 'httpjson',
+                policy_template: 'template_1',
+                enabled: true,
+                vars: {},
+                streams: [
+                  {
+                    enabled: true,
+                    data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                    vars: {
+                      site_ids: { type: 'text', value: '1234' }, // explicitly set
+                      enable_tracer: { type: 'bool', value: null }, // never set
+                    },
+                  },
+                ],
+              },
+            ],
+          };
+
+          const overrideWithStreamMigrateFrom: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              migrate_from: 'httpjson',
+              vars: {},
+              streams: [
+                {
+                  enabled: false,
+                  data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                  vars: {
+                    site_ids: { type: 'text', value: null }, // new default null
+                    enable_tracer: { type: 'bool', value: false }, // new default false
+                  },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            baseWithNullStreamVars,
+            makeCelPackageWithNullDefaults(),
+            overrideWithStreamMigrateFrom,
+            false
+          );
+
+          const celStream = result.inputs.find((i) => i.type === 'cel')?.streams[0];
+          // site_ids was explicitly set → preserved
+          expect(celStream?.vars?.site_ids.value).toBe('1234');
+          // enable_tracer old=null, new default=false → uses new default (false)
+          expect(celStream?.vars?.enable_tracer.value).toBe(false);
+        });
+      });
+
+      describe('when individual streams have migrate_from inside the datastream', () => {
+        it('should support stream-level migrate_from an input-level migration', () => {
+          const overrideWithBothLevels: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              migrate_from: 'httpjson',
+              vars: { url: { type: 'text', value: 'http://new-default.com' } },
+              streams: [
+                {
+                  enabled: true,
+                  migrate_from: 'httpjson',
+                  data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                  vars: { tags: { type: 'text', value: 'cel-default-tag' } },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            makeBasePolicy(), // has httpjson with tags: 'httpjson-tag'
+            makeCelPackageInfo(),
+            overrideWithBothLevels,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          expect(celInput?.streams[0]?.vars?.tags?.value).toBe('httpjson-tag');
+        });
+
+        const makeStreamOnlyMigrationFixtures = (oldInputEnabled: boolean) => {
+          const policyWithHttpjsonOnly: NewPackagePolicy = {
+            name: 'stream-only-migration-policy',
+            description: '',
+            namespace: 'default',
+            enabled: true,
+            policy_id: 'xxxx',
+            policy_ids: ['xxxx'],
+            package: { name: 'test-package', title: 'Test Package', version: '1.0.0' },
+            inputs: [
+              {
+                type: 'httpjson',
+                policy_template: 'template_1',
+                enabled: oldInputEnabled,
+                vars: {},
+                streams: [
+                  {
+                    enabled: true,
+                    data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                    vars: { paths: { type: 'text', value: '/var/log/app.log' } },
+                  },
+                ],
+              },
+            ],
+          };
+
+          const celOverrideStreamOnlyMigration: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              vars: {},
+              streams: [
+                {
+                  enabled: true,
+                  migrate_from: 'httpjson',
+                  data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                  vars: { paths: { type: 'text', value: '/default/path.log' } },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const celPackageInfoNoInputMigration = {
+            ...makeCelPackageInfo(),
+            policy_templates: [
+              {
+                name: 'template_1',
+                title: 'Template 1',
+                description: 'Template 1',
+                inputs: [
+                  {
+                    type: 'cel',
+                    title: 'CEL',
+                    description: 'CEL Input',
+                    // No input-level migrate_from — only the stream declares it
+                    vars: [],
+                  },
+                ],
+              },
+            ],
+          } as unknown as PackageInfo;
+
+          return {
+            policyWithHttpjsonOnly,
+            celOverrideStreamOnlyMigration,
+            celPackageInfoNoInputMigration,
+          };
+        };
+
+        it('should enable a new input when stream-level migrate_from is declared and the old input was enabled', () => {
+          const {
+            policyWithHttpjsonOnly,
+            celOverrideStreamOnlyMigration,
+            celPackageInfoNoInputMigration,
+          } = makeStreamOnlyMigrationFixtures(true);
+
+          const result = updatePackageInputs(
+            policyWithHttpjsonOnly,
+            celPackageInfoNoInputMigration,
+            celOverrideStreamOnlyMigration,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          expect(celInput).toBeDefined();
+
+          // Old httpjson input was enabled → new cel input should be enabled too
+          expect(celInput?.enabled).toBe(true);
+
+          // Stream vars and enabled state should be carried over from the old httpjson stream
+          expect(celInput?.streams[0]?.vars?.paths?.value).toBe('/var/log/app.log');
+          expect(celInput?.streams[0]?.enabled).toBe(true);
+        });
+
+        it('should keep the new input disabled when stream-level migrate_from is declared but the old input was disabled', () => {
+          const {
+            policyWithHttpjsonOnly,
+            celOverrideStreamOnlyMigration,
+            celPackageInfoNoInputMigration,
+          } = makeStreamOnlyMigrationFixtures(false);
+
+          const result = updatePackageInputs(
+            policyWithHttpjsonOnly,
+            celPackageInfoNoInputMigration,
+            celOverrideStreamOnlyMigration,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          expect(celInput).toBeDefined();
+
+          // Old httpjson input was disabled → new cel input should remain disabled
+          expect(celInput?.enabled).toBe(false);
+
+          // Stream vars should still be migrated even though the input is disabled
+          expect(celInput?.streams[0]?.vars?.paths?.value).toBe('/var/log/app.log');
+        });
+
+        it('should not migrate stream vars or enable the input when the new input is deprecated', () => {
+          const { policyWithHttpjsonOnly, celPackageInfoNoInputMigration } =
+            makeStreamOnlyMigrationFixtures(true);
+
+          // Mark the new cel input as deprecated
+          const deprecatedCelOverride: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              deprecated: { description: 'Use filestream instead' },
+              vars: {},
+              streams: [
+                {
+                  enabled: true,
+                  migrate_from: 'httpjson',
+                  data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                  vars: { paths: { type: 'text', value: '/default/path.log' } },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            policyWithHttpjsonOnly,
+            celPackageInfoNoInputMigration,
+            deprecatedCelOverride,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          expect(celInput).toBeDefined();
+
+          // Deprecated input should not be enabled by the migration logic
+          expect(celInput?.enabled).toBe(false);
+
+          // Stream vars should NOT be carried over from the old httpjson stream
+          expect(celInput?.streams[0]?.vars?.paths?.value).toBe('/default/path.log');
+        });
+      });
+
+      describe('when vars move from input-level in the old input to stream-level in the new input', () => {
+        // Mirrors the real-world httpjson→CEL scenario where site_ids and
+        // enable_request_tracer live at input-level in httpjson but at stream-level in CEL.
+        const makePolicyWithInputLevelVars = (): NewPackagePolicy => ({
+          name: 'input-level-vars-policy',
+          description: '',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '1.0.0' },
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                url: { type: 'text', value: 'http://example.com' },
+                site_ids: { type: 'text', value: '1392053568582758390' },
+                enable_request_tracer: { type: 'bool', value: true },
+              },
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                  vars: { interval: { type: 'text', value: '1m' } },
+                },
+              ],
+            },
+          ],
+        });
+
+        // CEL input: only `url` at input level; `site_ids` and `enable_request_tracer`
+        // moved to stream-level in the new package version.
+        const makeCelOverrideWithStreamLevelVars = (
+          extraStreamProps?: Record<string, unknown>
+        ): InputsOverride[] => [
+          {
+            type: 'cel',
+            policy_template: 'template_1',
+            enabled: false,
+            migrate_from: 'httpjson',
+            vars: { url: { type: 'text', value: 'http://new-default.com' } },
+            streams: [
+              {
+                enabled: true,
+                data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                vars: {
+                  interval: { type: 'text', value: '30s' },
+                  site_ids: { type: 'text', value: '' },
+                  enable_request_tracer: { type: 'bool', value: false },
+                },
+                ...extraStreamProps,
+              },
+            ],
+          } as unknown as InputsOverride,
+        ];
+
+        it('carries old input-level var values into the new stream-level vars', () => {
+          const result = updatePackageInputs(
+            makePolicyWithInputLevelVars(),
+            makeCelPackageInfo(),
+            makeCelOverrideWithStreamLevelVars(),
+            false
+          );
+
+          const celStream = result.inputs.find((i) => i.type === 'cel')?.streams[0];
+          expect(celStream?.vars?.site_ids?.value).toBe('1392053568582758390');
+          expect(celStream?.vars?.enable_request_tracer?.value).toBe(true);
+        });
+
+        it('gives old stream-level vars priority over old input-level vars when both define the same key', () => {
+          // interval exists at input-level in httpjson (with a different value) AND
+          // at stream-level in the old httpjson stream. The stream value should win.
+          const policyWithCollision: NewPackagePolicy = {
+            ...makePolicyWithInputLevelVars(),
+            inputs: [
+              {
+                type: 'httpjson',
+                policy_template: 'template_1',
+                enabled: true,
+                vars: {
+                  url: { type: 'text', value: 'http://example.com' },
+                  interval: { type: 'text', value: 'input-level-value' },
+                },
+                streams: [
+                  {
+                    enabled: true,
+                    data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                    vars: { interval: { type: 'text', value: 'stream-level-value' } },
+                  },
+                ],
+              },
+            ],
+          };
+
+          const celOverride: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              migrate_from: 'httpjson',
+              vars: { url: { type: 'text', value: 'http://new-default.com' } },
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.cel_log', type: 'logs' },
+                  vars: { interval: { type: 'text', value: 'new-default' } },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            policyWithCollision,
+            makeCelPackageInfo(),
+            celOverride,
+            false
+          );
+
+          const celStream = result.inputs.find((i) => i.type === 'cel')?.streams[0];
+          // Old stream-level value must win over old input-level value
+          expect(celStream?.vars?.interval?.value).toBe('stream-level-value');
+        });
+
+        it('does not introduce vars from the old input that are absent from the new stream schema', () => {
+          // orphaned_var exists in the old httpjson input but is not defined in any cel stream
+          const policyWithOrphanedVar: NewPackagePolicy = {
+            ...makePolicyWithInputLevelVars(),
+            inputs: [
+              {
+                type: 'httpjson',
+                policy_template: 'template_1',
+                enabled: true,
+                vars: {
+                  url: { type: 'text', value: 'http://example.com' },
+                  orphaned_var: { type: 'text', value: 'should-not-appear' },
+                },
+                streams: [
+                  {
+                    enabled: true,
+                    data_stream: { dataset: 'test_package.httpjson_log', type: 'logs' },
+                    vars: {},
+                  },
+                ],
+              },
+            ],
+          };
+
+          const result = updatePackageInputs(
+            policyWithOrphanedVar,
+            makeCelPackageInfo(),
+            makeCelOverrideWithStreamLevelVars(),
+            false
+          );
+
+          const celStream = result.inputs.find((i) => i.type === 'cel')?.streams[0];
+          // removeStaleVars must have stripped the key not present in the new stream schema
+          expect(celStream?.vars?.orphaned_var).toBeUndefined();
+        });
+
+        it('seeds old input-level vars into a new stream that has no positional old stream', () => {
+          // New CEL input has 2 streams; old httpjson only had 1.
+          // The second CEL stream has no old counterpart, but old input-level vars should
+          // still be seeded where the new stream schema defines them.
+          const celOverrideTwoStreams: InputsOverride[] = [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: false,
+              migrate_from: 'httpjson',
+              vars: { url: { type: 'text', value: 'http://new-default.com' } },
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.cel_log_a', type: 'logs' },
+                  vars: {
+                    interval: { type: 'text', value: '30s' },
+                    site_ids: { type: 'text', value: '' },
+                  },
+                },
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.cel_log_b', type: 'logs' },
+                  vars: {
+                    interval: { type: 'text', value: '60s' },
+                    site_ids: { type: 'text', value: '' },
+                  },
+                },
+              ],
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            makePolicyWithInputLevelVars(),
+            makeCelPackageInfo(),
+            celOverrideTwoStreams,
+            false
+          );
+
+          const celInput = result.inputs.find((i) => i.type === 'cel');
+          // First stream: has an old positional counterpart — gets both old stream + old input vars
+          expect(celInput?.streams[0]?.vars?.site_ids?.value).toBe('1392053568582758390');
+          // Second stream: no old counterpart stream, but old input-level vars are still seeded
+          expect(celInput?.streams[1]?.vars?.site_ids?.value).toBe('1392053568582758390');
+        });
+
+        it('does not carry old input-level vars to stream level when the new input is deprecated', () => {
+          const deprecatedCelOverride: InputsOverride[] = [
+            {
+              ...makeCelOverrideWithStreamLevelVars()[0],
+              deprecated: { description: 'Use filestream instead' },
+            } as unknown as InputsOverride,
+          ];
+
+          const result = updatePackageInputs(
+            makePolicyWithInputLevelVars(),
+            makeCelPackageInfo(),
+            deprecatedCelOverride,
+            false
+          );
+
+          const celStream = result.inputs.find((i) => i.type === 'cel')?.streams[0];
+          // Values should remain at new-package defaults, not the old user-configured values
+          expect(celStream?.vars?.site_ids?.value).toBe('');
+          expect(celStream?.vars?.enable_request_tracer?.value).toBe(false);
+        });
+      });
+    });
+
+    describe('when new input already exists alongside old input (partial migration)', () => {
+      // Mirrors real-world integrations like sentinel_one / cisco_duo where BOTH httpjson AND
+      // cel existed in the old policy. In the new package httpjson is removed and its streams
+      // are transferred to cel via stream-level migrate_from.
+      //
+      // The input-level migrate_from path (originalInput === undefined) is NOT triggered here
+      // because cel already exists in the old policy. Instead, the normal merge path runs but
+      // must handle new streams that carry migrate_from declarations.
+
+      const makePartialMigrationBasePolicy = (): NewPackagePolicy => ({
+        name: 'partial-migration-policy',
+        description: '',
+        namespace: 'default',
+        enabled: true,
+        policy_id: 'xxxx',
+        policy_ids: ['xxxx'],
+        package: { name: 'test-package', title: 'Test Package', version: '1.0.0' },
+        inputs: [
+          {
+            // Old httpjson input — will be removed in the new package version
+            type: 'httpjson',
+            policy_template: 'template_1',
+            enabled: true,
+            vars: {
+              url: { type: 'text', value: 'http://user-configured.com' },
+              api_token: { type: 'password', value: 'secret-token' },
+              // These two live at input-level in httpjson but at stream-level in cel
+              site_ids: { type: 'text', value: '1392053568582758390' },
+              enable_request_tracer: { type: 'bool', value: true },
+            },
+            streams: [
+              {
+                enabled: true,
+                data_stream: { dataset: 'test_package.activity', type: 'logs' },
+                vars: {
+                  interval: { type: 'text', value: '5m' },
+                  tags: { type: 'text', value: 'custom-tag' },
+                },
+              },
+              {
+                // User explicitly disabled this stream
+                enabled: false,
+                data_stream: { dataset: 'test_package.agent', type: 'logs' },
+                vars: { interval: { type: 'text', value: '10m' } },
+              },
+            ],
+          },
+          {
+            // Pre-existing cel input — stays in policy and gets new streams via migrate_from
+            type: 'cel',
+            policy_template: 'template_1',
+            enabled: true,
+            vars: {
+              url: { type: 'text', value: 'http://cel-configured.com' },
+              api_token: { type: 'password', value: 'cel-secret' },
+            },
+            streams: [
+              {
+                // This stream already existed in cel — must NOT be reset
+                enabled: true,
+                data_stream: { dataset: 'test_package.application', type: 'logs' },
+                vars: {
+                  batch_size: { type: 'text', value: '500' }, // user changed from default 1000
+                  interval: { type: 'text', value: '2m' },
+                },
+              },
+            ],
+          },
+        ],
+      });
+
+      const makePartialMigrationPackageInfo = (): PackageInfo =>
+        ({
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '1.0.2',
+          latestVersion: '1.0.2',
+          release: 'experimental',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  // httpjson is gone — only cel remains
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  vars: [
+                    { name: 'url', type: 'text' },
+                    { name: 'api_token', type: 'password' },
+                  ],
+                },
+              ],
+            },
+          ],
+          assets: {},
+        } as unknown as PackageInfo);
+
+      // The cel InputsOverride with 3 streams: 2 new (migrating from httpjson) + 1 existing
+      const makePartialMigrationOverride = (): InputsOverride[] => [
+        {
+          type: 'cel',
+          policy_template: 'template_1',
+          enabled: false,
+          // No input-level migrate_from — streams declare it individually
+          vars: {
+            url: { type: 'text', value: 'http://new-package-default.com' },
+            api_token: { type: 'password', value: '' },
+          },
+          streams: [
+            {
+              // New stream that migrates from httpjson
+              enabled: true,
+              migrate_from: 'httpjson',
+              data_stream: { dataset: 'test_package.activity', type: 'logs' },
+              vars: {
+                interval: { type: 'text', value: '30s' },
+                tags: { type: 'text', value: 'default-tag' },
+                site_ids: { type: 'text', value: '' },
+                enable_request_tracer: { type: 'bool', value: false },
+              },
+            },
+            {
+              // New stream that migrates from httpjson
+              enabled: true,
+              migrate_from: 'httpjson',
+              data_stream: { dataset: 'test_package.agent', type: 'logs' },
+              vars: {
+                interval: { type: 'text', value: '30s' },
+                tags: { type: 'text', value: 'default-tag' },
+                site_ids: { type: 'text', value: '' },
+                enable_request_tracer: { type: 'bool', value: false },
+              },
+            },
+            {
+              // Existing cel stream — matched by dataset, goes through normal merge
+              enabled: true,
+              data_stream: { dataset: 'test_package.application', type: 'logs' },
+              vars: {
+                batch_size: { type: 'text', value: '1000' },
+                interval: { type: 'text', value: '60s' },
+              },
+            },
+          ],
+        } as unknown as InputsOverride,
+      ];
+
+      it('carries stream-level vars from old httpjson stream to the new migrating cel stream', () => {
+        const result = updatePackageInputs(
+          makePartialMigrationBasePolicy(),
+          makePartialMigrationPackageInfo(),
+          makePartialMigrationOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const activityStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.activity'
+        );
+        // Old httpjson activity stream had interval: '5m' and tags: 'custom-tag'
+        expect(activityStream?.vars?.interval?.value).toBe('5m');
+        expect(activityStream?.vars?.tags?.value).toBe('custom-tag');
+      });
+
+      it('seeds old input-level vars into stream-level vars of the migrating stream', () => {
+        const result = updatePackageInputs(
+          makePartialMigrationBasePolicy(),
+          makePartialMigrationPackageInfo(),
+          makePartialMigrationOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const activityStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.activity'
+        );
+        // site_ids and enable_request_tracer were at input-level in httpjson but stream-level in cel
+        expect(activityStream?.vars?.site_ids?.value).toBe('1392053568582758390');
+        expect(activityStream?.vars?.enable_request_tracer?.value).toBe(true);
+      });
+
+      it('preserves the enabled state from the old httpjson stream on the new cel stream', () => {
+        const result = updatePackageInputs(
+          makePartialMigrationBasePolicy(),
+          makePartialMigrationPackageInfo(),
+          makePartialMigrationOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const agentStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.agent'
+        );
+        // Old httpjson agent stream was disabled by the user → new cel stream must stay disabled
+        expect(agentStream?.enabled).toBe(false);
+      });
+
+      it('leaves existing cel streams completely untouched during partial migration', () => {
+        const result = updatePackageInputs(
+          makePartialMigrationBasePolicy(),
+          makePartialMigrationPackageInfo(),
+          makePartialMigrationOverride(),
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const appStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.application'
+        );
+        // User had changed batch_size from 1000 → 500; it must not be reset to the package default
+        expect(appStream?.vars?.batch_size?.value).toBe('500');
+        expect(appStream?.vars?.interval?.value).toBe('2m');
+      });
+
+      it('falls back to package defaults when migrate_from points to a non-existent input type', () => {
+        const overrideWithBadMigrateFrom: InputsOverride[] = [
+          {
+            ...makePartialMigrationOverride()[0],
+            streams: [
+              {
+                enabled: true,
+                migrate_from: 'nonexistent',
+                data_stream: { dataset: 'test_package.activity', type: 'logs' },
+                vars: {
+                  interval: { type: 'text', value: 'package-default' },
+                  site_ids: { type: 'text', value: '' },
+                },
+              },
+            ],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(
+          makePartialMigrationBasePolicy(),
+          makePartialMigrationPackageInfo(),
+          overrideWithBadMigrateFrom,
+          false
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const activityStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.activity'
+        );
+        // Old input type not found → stream gets package defaults, not the httpjson values
+        expect(activityStream?.vars?.interval?.value).toBe('package-default');
+        expect(activityStream?.vars?.site_ids?.value).toBe('');
+      });
+
+      it('does not touch a third input type during partial migration', () => {
+        const policyWithThreeInputs: NewPackagePolicy = {
+          ...makePartialMigrationBasePolicy(),
+          inputs: [
+            ...makePartialMigrationBasePolicy().inputs,
+            {
+              type: 'azure-eventhub',
+              policy_template: 'template_1',
+              enabled: false,
+              vars: {},
+              streams: [
+                {
+                  enabled: false,
+                  data_stream: { dataset: 'test_package.event', type: 'logs' },
+                  vars: {
+                    connection_string: { type: 'password', value: 'user-connection-string' },
+                    consumer_group: { type: 'text', value: 'my-group' },
+                  },
+                },
+              ],
+            },
+          ],
+        };
+
+        const packageInfoWithThreeInputs: PackageInfo = {
+          ...makePartialMigrationPackageInfo(),
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  vars: [{ name: 'url', type: 'text' }],
+                },
+                {
+                  type: 'azure-eventhub',
+                  title: 'Azure Event Hub',
+                  description: 'Azure Event Hub Input',
+                  vars: [],
+                },
+              ],
+            },
+          ],
+        } as unknown as PackageInfo;
+
+        const overrideWithAzure: InputsOverride[] = [
+          ...makePartialMigrationOverride(),
+          {
+            type: 'azure-eventhub',
+            policy_template: 'template_1',
+            enabled: false,
+            vars: {},
+            streams: [
+              {
+                enabled: false,
+                data_stream: { dataset: 'test_package.event', type: 'logs' },
+                vars: {
+                  connection_string: { type: 'password', value: '' },
+                  consumer_group: { type: 'text', value: '$default' },
+                },
+              },
+            ],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(
+          policyWithThreeInputs,
+          packageInfoWithThreeInputs,
+          overrideWithAzure,
+          false
+        );
+
+        const azureInput = result.inputs.find((i) => i.type === 'azure-eventhub');
+        const eventStream = azureInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.event'
+        );
+        // azure-eventhub input and streams must be completely unaffected by the httpjson→cel migration
+        expect(eventStream?.vars?.connection_string?.value).toBe('user-connection-string');
+        expect(eventStream?.vars?.consumer_group?.value).toBe('my-group');
+      });
+
+      it('migrates old httpjson input-level vars to new cel stream-level vars when going through packageToPackagePolicyInputs end-to-end', () => {
+        // 2.5.0 packageInfo: only cel input remains; activity stream declares migrate_from: httpjson
+        const packageInfo250: PackageInfo = {
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '2.5.0',
+          latestVersion: '2.5.0',
+          release: 'ga',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL input',
+                  vars: [
+                    { name: 'url', type: 'url' },
+                    { name: 'api_token', type: 'password' },
+                  ],
+                },
+              ],
+            },
+          ],
+          data_streams: [
+            {
+              type: 'logs',
+              dataset: 'test_package.activity',
+              path: 'activity',
+              title: 'Activity',
+              release: 'ga',
+              ingest_pipeline: 'default',
+              package: 'test-package',
+              streams: [
+                {
+                  input: 'cel',
+                  migrate_from: 'httpjson',
+                  title: 'Activity',
+                  vars: [
+                    { name: 'interval', type: 'text', default: '30s' },
+                    { name: 'site_ids', type: 'text' },
+                    { name: 'enable_request_tracer', type: 'bool', default: false },
+                  ],
+                },
+              ],
+            },
+            {
+              type: 'logs',
+              dataset: 'test_package.application',
+              path: 'application',
+              title: 'Application',
+              release: 'ga',
+              ingest_pipeline: 'default',
+              package: 'test-package',
+              streams: [
+                {
+                  input: 'cel',
+                  title: 'Application',
+                  vars: [{ name: 'batch_size', type: 'text', default: '1000' }],
+                },
+              ],
+            },
+          ],
+          assets: {},
+        } as unknown as PackageInfo;
+
+        // 2.4.1 base policy: old cel (application only) + old httpjson (activity + more)
+        // The httpjson input stores site_ids and enable_request_tracer at input level.
+        const basePolicy241: NewPackagePolicy = {
+          name: 'sentinel-one-policy',
+          description: '',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '2.4.1' },
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                url: { type: 'url', value: 'http://sentinelone.example.com' },
+                api_token: { type: 'password', value: 'user-token' },
+                site_ids: { type: 'text', value: '1111,2222' },
+                enable_request_tracer: { type: 'bool', value: true },
+              },
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.activity', type: 'logs' },
+                  vars: { interval: { type: 'text', value: '1m' } },
+                },
+              ],
+            },
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                url: { type: 'url', value: 'http://sentinelone.example.com' },
+                api_token: { type: 'password', value: 'user-token' },
+              },
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.application', type: 'logs' },
+                  vars: { batch_size: { type: 'text', value: '500' } },
+                },
+              ],
+            },
+          ],
+        };
+
+        const result = updatePackageInputs(
+          basePolicy241,
+          packageInfo250,
+          packageToPackagePolicyInputs(packageInfo250) as InputsOverride[]
+        );
+
+        const celInput = result.inputs.find((i) => i.type === 'cel');
+        const activityStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.activity'
+        );
+
+        // site_ids and enable_request_tracer lived at httpjson INPUT level in 2.4.1,
+        // and at cel STREAM level in 2.5.0 — they must be carried over by the migration.
+        expect(activityStream?.vars?.site_ids?.value).toBe('1111,2222');
+        expect(activityStream?.vars?.enable_request_tracer?.value).toBe(true);
+
+        // Stream-level vars from the old httpjson stream are also migrated.
+        expect(activityStream?.vars?.interval?.value).toBe('1m');
+
+        // Existing application stream must keep its user-configured value.
+        const appStream = celInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.application'
+        );
+        expect(appStream?.vars?.batch_size?.value).toBe('500');
+      });
+    });
+
+    describe('when re-upgrading to a package version that removes deprecated/migrate_from', () => {
+      it('clears deprecated and migrate_from on an existing input when new package no longer declares them', () => {
+        const basePackagePolicy: NewPackagePolicy = {
+          name: 'base-package-policy',
+          description: 'Base Package Policy',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+          inputs: [
+            {
+              type: 'logs',
+              policy_template: 'template_1',
+              enabled: true,
+              // Simulates a policy stored after a previous upgrade that added these fields
+              deprecated: { description: 'Use cel input instead' },
+              migrate_from: 'httpjson',
+              vars: { path: { type: 'text', value: '/var/log/logfile.log' } },
+              streams: [],
+            },
+          ],
+        };
+
+        const packageInfo: PackageInfo = {
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '0.0.2',
+          latestVersion: '0.0.2',
+          release: 'experimental',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'logs',
+                  title: 'Log',
+                  description: 'Log Input',
+                  // New package version no longer marks the input as deprecated or migrate_from
+                  vars: [{ name: 'path', type: 'text' }],
+                },
+              ],
+            },
+          ],
+          // @ts-ignore
+          assets: {},
+        };
+
+        const inputsOverride: InputsOverride[] = [
+          {
+            type: 'logs',
+            policy_template: 'template_1',
+            enabled: true,
+            // No deprecated, no migrate_from in the new package definition
+            vars: { path: { type: 'text', value: '/var/log/new-default.log' } },
+            streams: [],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(basePackagePolicy, packageInfo, inputsOverride, false);
+
+        const logsInput = result.inputs.find((i) => i.type === 'logs');
+        expect(logsInput).toBeDefined();
+        expect(logsInput?.deprecated).toBeUndefined();
+        expect(logsInput?.migrate_from).toBeUndefined();
+        // User-configured var should still be preserved
+        expect(logsInput?.vars?.path?.value).toBe('/var/log/logfile.log');
+      });
+
+      it('clears migrate_from on an existing stream when new package no longer declares it', () => {
+        const basePackagePolicy: NewPackagePolicy = {
+          name: 'base-package-policy',
+          description: 'Base Package Policy',
+          namespace: 'default',
+          enabled: true,
+          policy_id: 'xxxx',
+          policy_ids: ['xxxx'],
+          package: { name: 'test-package', title: 'Test Package', version: '0.0.1' },
+          inputs: [
+            {
+              type: 'logs',
+              policy_template: 'template_1',
+              enabled: true,
+              streams: [
+                {
+                  enabled: true,
+                  data_stream: { dataset: 'test_package.logs', type: 'logs' },
+                  // Simulates a stream stored with migrate_from from a previous upgrade
+                  migrate_from: 'httpjson',
+                  vars: { tags: { type: 'text', value: 'user-tag' } },
+                },
+              ],
+            },
+          ],
+        };
+
+        const packageInfo: PackageInfo = {
+          name: 'test-package',
+          description: 'Test Package',
+          title: 'Test Package',
+          version: '0.0.2',
+          latestVersion: '0.0.2',
+          release: 'experimental',
+          format_version: '1.0.0',
+          owner: { github: 'elastic/fleet' },
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [{ type: 'logs', title: 'Log', description: 'Log Input', vars: [] }],
+            },
+          ],
+          data_streams: [
+            {
+              dataset: 'test_package.logs',
+              type: 'logs',
+              title: 'Logs',
+              release: 'experimental' as any,
+              package: 'test-package',
+              path: 'logs',
+              streams: [
+                {
+                  input: 'logs',
+                  title: 'Logs',
+                  vars: [{ name: 'tags', type: 'text' }],
+                  template_path: 'agent.yml',
+                },
+              ],
+            },
+          ],
+          // @ts-ignore
+          assets: {},
+        };
+
+        const inputsOverride: InputsOverride[] = [
+          {
+            type: 'logs',
+            policy_template: 'template_1',
+            enabled: true,
+            streams: [
+              {
+                enabled: true,
+                data_stream: { dataset: 'test_package.logs', type: 'logs' },
+                // New package version stream no longer declares migrate_from
+                vars: { tags: { type: 'text', value: 'default-tag' } },
+              },
+            ],
+          } as unknown as InputsOverride,
+        ];
+
+        const result = updatePackageInputs(basePackagePolicy, packageInfo, inputsOverride, false);
+
+        const logsInput = result.inputs.find((i) => i.type === 'logs');
+        const logsStream = logsInput?.streams.find(
+          (s) => s.data_stream.dataset === 'test_package.logs'
+        );
+        expect(logsStream?.migrate_from).toBeUndefined();
+        // User-configured var should still be preserved
+        expect(logsStream?.vars?.tags?.value).toBe('user-tag');
+      });
+    });
   });
 
-  describe('enrich package policy on create', () => {
+  describe('Enrich package policy on create', () => {
     beforeEach(() => {
       (packageToPackagePolicy as jest.Mock).mockReturnValue({
         package: { name: 'apache', title: 'Apache', version: '1.0.0' },
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts b/x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts
index 0da00dacba0f7..9d0758198e76d 100644
--- a/x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts
+++ b/x-pack/platform/plugins/shared/fleet/server/services/package_policy.ts
@@ -50,7 +50,6 @@ import {
   getNormalizedInputs,
   isRootPrivilegesRequired,
   checkIntegrationFipsLooseCompatibility,
-  varsReducer,
   hasMultipleEnabledPolicyTemplates,
 } from '../../common/services';
 import {
@@ -84,7 +83,6 @@ import type {
   CloudConnectorVars,
   CloudConnectorSecretVar,
   AwsCloudConnectorVars,
-  PackagePolicyConfigRecord,
   ArchiveEntry,
 } from '../../common/types';
 import type {
@@ -225,6 +223,15 @@ import {
   hasAgentVersionConditionInInputTemplate,
 } from './utils/version_specific_policies';
 import { recompileInputsWithAgentVersion } from './agent_policies/package_policies_to_agent_inputs';
+import {
+  findInputForMigration,
+  applyInputLevelMigration,
+  migrateStreamVars,
+  applyStreamLevelMigration,
+  deepMergeVars,
+  getUpdatedGlobalVars,
+  removeStaleVars,
+} from './package_policy_migration_helpers';
 
 export type InputsOverride = Partial<NewPackagePolicyInput> & {
   vars?: Array<NewPackagePolicyInput['vars'] & { name: string }>;
@@ -2581,7 +2588,6 @@ class PackagePolicyClientImpl implements PackagePolicyClient {
               i.type === input.type &&
               (!input.policy_template || input.policy_template === i.policy_template)
           );
-
           return {
             ...defaultInput,
             enabled: input.enabled,
@@ -3907,7 +3913,6 @@ export function updatePackageInputs(
         : policyTemplate.inputs?.some(
             (policyTemplateInput) => policyTemplateInput.type === input.type
           ) ?? false;
-
       return policyTemplateStillIncludesInput;
     }),
   ];
@@ -3938,6 +3943,13 @@ export function updatePackageInputs(
     // take the override value from the new package as-is. This case typically
     // occurs when inputs or package policy templates are added/removed between versions.
     if (originalInput === undefined) {
+      const originalInputToMigrate = applyInputLevelMigration(
+        update,
+        basePackagePolicy.inputs,
+        inputs
+      );
+      applyStreamLevelMigration(update, originalInputToMigrate, basePackagePolicy.inputs);
+
       // Do not enable new inputs for limited packages
       if (limitedPackage) {
         update.enabled = false;
@@ -3963,6 +3975,12 @@ export function updatePackageInputs(
       originalInput.policy_template = update.policy_template;
     }
 
+    // `deprecated` and `migrate_from` are package schema fields, not user-configured state.
+    // They must be unconditionally synced from the new package definition so they are cleared
+    // when the new package version no longer declares them.
+    originalInput.deprecated = update.deprecated;
+    originalInput.migrate_from = update.migrate_from;
+
     if (update.vars || originalInput.vars) {
       const indexOfInput = inputs.indexOf(originalInput);
       const mergedVars = deepMergeVars(originalInput, update, true) as NewPackagePolicyInput;
@@ -3976,6 +3994,9 @@ export function updatePackageInputs(
         packageInfo.type === 'input' &&
         update.streams.length === 1 &&
         originalInput?.streams.length === 1;
+      // Per-source-type counters for positional stream matching when a stream declares
+      // migrate_from. Shared across iterations so each source stream is consumed once.
+      const streamMigrateFromCounters: Record<string, number> = {};
       for (const stream of update.streams) {
         let originalStream = originalInput?.streams.find(
           (s) => s.data_stream.dataset === stream.data_stream.dataset
@@ -3992,6 +4013,26 @@ export function updatePackageInputs(
         }
 
         if (originalStream === undefined) {
+          // When a new stream declares migrate_from, carry vars and enabled state over from
+          // the positionally-matched old stream instead of pushing with package defaults.
+          // This handles partial-migration integrations where
+          // the new input type already exists in the old policy alongside the old input type.
+          if (stream.migrate_from) {
+            const counter = streamMigrateFromCounters[stream.migrate_from] ?? 0;
+            streamMigrateFromCounters[stream.migrate_from] = counter + 1;
+            const oldInputForStream = findInputForMigration(
+              basePackagePolicy.inputs,
+              stream.migrate_from,
+              update.policy_template
+            );
+            const oldStream = oldInputForStream?.streams[counter];
+            if (oldStream) {
+              originalInput.streams.push(
+                migrateStreamVars(stream as InputsOverride, oldStream, oldInputForStream?.vars)
+              );
+              continue;
+            }
+          }
           originalInput.streams.push(stream);
           continue;
         }
@@ -4000,6 +4041,9 @@ export function updatePackageInputs(
           originalStream.enabled = stream.enabled;
         }
 
+        // Sync migrate_from from the new package schema — package-owned field, not user-configured.
+        originalStream.migrate_from = stream.migrate_from;
+
         if (stream.vars || originalStream.vars) {
           // streams wont match for input pkgs
           const indexOfStream = isInputPkgUpdate
@@ -4012,7 +4056,6 @@ export function updatePackageInputs(
         }
       }
     }
-
     // Filter all stream that have been removed from the input
     originalInput.streams = originalInput.streams.filter((originalStream) => {
       return (
@@ -4285,83 +4328,6 @@ export function sendUpdatePackagePolicyTelemetryEvent(
   });
 }
 
-function deepMergeVars(original: any, override: any, keepOriginalValue = false): any {
-  if (!override.vars) {
-    return original;
-  }
-  if (!original.vars) {
-    original.vars = { ...override.vars };
-  }
-
-  const result = { ...original };
-
-  const overrideVars = Array.isArray(override.vars)
-    ? override.vars
-    : Object.entries(override.vars!).map(([key, rest]) => ({
-        name: key,
-        ...(rest as any),
-      }));
-
-  for (const { name, ...overrideVal } of overrideVars) {
-    const originalVar = original.vars[name];
-
-    result.vars[name] = { ...originalVar, ...overrideVal };
-
-    // Ensure that any value from the original object is persisted on the newly merged resulting object,
-    // even if we merge other data about the given variable
-    if (keepOriginalValue && originalVar?.value !== undefined) {
-      result.vars[name].value = originalVar.value;
-    }
-  }
-
-  return result;
-}
-
-function getUpdatedGlobalVars(packageInfo: PackageInfo, packagePolicy: NewPackagePolicy) {
-  if (!packageInfo.vars) {
-    return undefined;
-  }
-
-  const packageInfoVars = packageInfo.vars.reduce(varsReducer, {});
-  const result = deepMergeVars(packagePolicy, { vars: packageInfoVars }, true);
-  return removeStaleVars(result, { vars: packageInfoVars }).vars;
-}
-
-interface SupportsVars {
-  vars?: PackagePolicyConfigRecord;
-}
-
-function removeStaleVars<T extends SupportsVars>(
-  currentWithVars: T,
-  expectedVars: SupportsVars
-): T {
-  if (!currentWithVars.vars) {
-    return currentWithVars;
-  }
-
-  if (!expectedVars.vars) {
-    return {
-      ...currentWithVars,
-      vars: {},
-    };
-  }
-
-  const filteredVars = Object.entries(currentWithVars.vars).reduce<PackagePolicyConfigRecord>(
-    (acc, [key, val]) => {
-      if (key in expectedVars.vars!) {
-        acc[key] = val;
-      }
-      return acc;
-    },
-    {}
-  );
-
-  return {
-    ...currentWithVars,
-    vars: filteredVars,
-  };
-}
-
 async function requireUniqueName(
   soClient: SavedObjectsClientContract,
   packagePolicy: UpdatePackagePolicy | NewPackagePolicy,
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/package_policy_migration_helpers.ts b/x-pack/platform/plugins/shared/fleet/server/services/package_policy_migration_helpers.ts
new file mode 100644
index 0000000000000..39950b984c2bf
--- /dev/null
+++ b/x-pack/platform/plugins/shared/fleet/server/services/package_policy_migration_helpers.ts
@@ -0,0 +1,297 @@
+/*
+ * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
+ * or more contributor license agreements. Licensed under the Elastic License
+ * 2.0; you may not use this file except in compliance with the Elastic License
+ * 2.0.
+ */
+
+import type {
+  NewPackagePolicyInput,
+  PackagePolicyConfigRecord,
+  NewPackagePolicyInputStream,
+  PackageInfo,
+  InputsOverride,
+} from '../../common/types';
+import type { NewPackagePolicy } from '../types';
+import { varsReducer } from '../../common/services';
+
+/**
+ * Finds an input in `inputs` matching `type`, preferring a match on `policyTemplate`. Falls back
+ * to an entry with no `policy_template` set to handle older stored policies where that field was
+ * not reliably populated. Returns undefined when no match is found.
+ */
+export function findInputForMigration(
+  inputs: NewPackagePolicyInput[],
+  type: string,
+  policyTemplate: string | undefined
+): NewPackagePolicyInput | undefined {
+  if (policyTemplate) {
+    return (
+      inputs.find((i) => i.type === type && i.policy_template === policyTemplate) ??
+      inputs.find((i) => i.type === type && !i.policy_template)
+    );
+  }
+  return inputs.find((i) => i.type === type);
+}
+
+/**
+ * Applies input-level `migrate_from` migration when a new input type explicitly replaces an old
+ * one.
+ *
+ * Mutates `update` in-place (merges vars, preserves the old input's enabled state) and removes the
+ * now-stale old input from the mutable `inputs` array when it was not already pruned by the
+ * policy-template filter above.
+ *
+ * Returns the source input that was migrated from, or undefined when no migration applies.
+ */
+export function applyInputLevelMigration(
+  update: InputsOverride,
+  allBaseInputs: NewPackagePolicyInput[],
+  inputs: NewPackagePolicyInput[]
+): NewPackagePolicyInput | undefined {
+  if (update.migrate_from === undefined || update.deprecated) {
+    return undefined;
+  }
+
+  const originalInputToMigrate = findInputForMigration(
+    allBaseInputs,
+    update.migrate_from,
+    update.policy_template
+  );
+  if (!originalInputToMigrate) {
+    return undefined;
+  }
+
+  // Ensure the old input doesn't linger in inputs when it has no policy_template
+  // (inputs with a policy_template are already removed by the filter above)
+  const foundStale = findInputForMigration(inputs, update.migrate_from, update.policy_template);
+  const staleIdx = foundStale ? inputs.indexOf(foundStale) : -1;
+  if (staleIdx !== -1) inputs.splice(staleIdx, 1);
+
+  // Merge old input vars into the new input: seed with old values, keep new schema as the
+  // authoritative list of vars, then strip any keys not present in the new schema.
+  // deepMergeVars iterates over `update` (new schema) and restores non-null old values, so
+  // null old values fall through to the new package defaults (see keepOriginalValue logic).
+  const mergedInput = deepMergeVars(
+    { ...update, vars: originalInputToMigrate.vars },
+    update,
+    true
+  ) as InputsOverride;
+  update.vars = sanitizeMigratedVars(removeStaleVars(mergedInput, update)).vars;
+  // Preserve the enabled state of the old input rather than unconditionally enabling.
+  update.enabled = originalInputToMigrate.enabled;
+
+  return originalInputToMigrate;
+}
+
+/**
+ * Merges vars from an old stream (and optionally its parent input's vars) into a new stream,
+ * preserving the old stream's `enabled` state.
+ *
+ * Old input-level vars and old stream-level vars are combined before merging so that vars which
+ * moved from input-level in the old input to stream-level in the new input are also carried over.
+ * Stream-level vars take priority over input-level vars on collision.
+ * `removeStaleVars` then discards any key not defined in the new stream schema.
+ *
+ * `oldStream` may be `undefined` when only input-level vars are available (e.g. the new input
+ * has more streams than the old one). In that case `enabled` is left at the new stream's default.
+ */
+export function migrateStreamVars(
+  newStream: InputsOverride,
+  oldStream: NewPackagePolicyInputStream | undefined,
+  oldInputVars: PackagePolicyConfigRecord | undefined
+): NewPackagePolicyInputStream {
+  // Combine old input-level vars with old stream-level vars. Stream-level vars take
+  // priority over input-level vars for the same key (more specific value wins).
+  // removeStaleVars below will then discard any key not defined in the new stream schema.
+  const combinedOldVars: PackagePolicyConfigRecord = {
+    ...(oldInputVars ?? {}),
+    ...(oldStream?.vars ?? {}),
+  };
+
+  const merged = deepMergeVars(
+    { ...newStream, vars: combinedOldVars },
+    newStream as InputsOverride,
+    true
+  );
+  // deepMergeVars only handles vars; explicitly carry the enabled state from
+  // the old stream so the user's enable/disable choice is preserved.
+  return sanitizeMigratedVars({
+    ...removeStaleVars(merged, newStream),
+    ...(oldStream ? { enabled: oldStream.enabled } : {}),
+  });
+}
+
+/**
+ * Applies stream-level `migrate_from` migration for a new input that has no matching existing
+ * input in the current policy.
+ *
+ * Two sources of old streams are considered (in priority order):
+ *  1. `originalInputToMigrate.streams` — when the parent input declared `migrate_from`, streams
+ *     are matched positionally to the corresponding old input's streams.
+ *  2. `newStream.migrate_from` — each stream can independently declare which old input type it
+ *     migrates from, also matched positionally per source type.
+ *
+ * `update.streams` is replaced with the merged streams.
+ * `update.enabled` is set from the old input's enabled state when stream-level migration occurred
+ * without a corresponding input-level `migrate_from`.
+ */
+export function applyStreamLevelMigration(
+  update: InputsOverride,
+  originalInputToMigrate: NewPackagePolicyInput | undefined,
+  allBaseInputs: NewPackagePolicyInput[]
+): void {
+  if (!update.streams || update.deprecated) {
+    return;
+  }
+
+  const streamMigrateFromCounters: Record<string, number> = {};
+  let streamMigrationOccurred = false;
+  // Track the first old input encountered during stream-level migration so we can
+  // carry its enabled state over to the new input.
+  let oldInputForStreamMigration: NewPackagePolicyInput | undefined;
+
+  update.streams = update.streams.map((newStream, idx) => {
+    let oldStream: NewPackagePolicyInputStream | undefined;
+    let oldInputVars: PackagePolicyConfigRecord | undefined;
+
+    // Migrate stream-level vars by position since datasets differ between input types.
+    // Use the new stream as the structural base (preserving data_stream identity) and seed
+    // its vars with values from the old stream so user configuration is carried over.
+    if (originalInputToMigrate && originalInputToMigrate.streams.length > 0) {
+      oldStream = originalInputToMigrate.streams[idx];
+      // Capture old input-level vars so that vars which moved from input-level in the
+      // old input to stream-level in the new input are also carried over.
+      oldInputVars = originalInputToMigrate.vars;
+    } else if (newStream.migrate_from) {
+      // When streams have migrate_from:
+      // each stream with migrate_from is matched positionally to the corresponding old stream in the specified input type.
+      const counter = streamMigrateFromCounters[newStream.migrate_from] ?? 0;
+      streamMigrateFromCounters[newStream.migrate_from] = counter + 1;
+      const oldInputForStream = findInputForMigration(
+        allBaseInputs,
+        newStream.migrate_from,
+        update.policy_template
+      );
+      oldStream = oldInputForStream?.streams[counter];
+      if (oldStream) {
+        streamMigrationOccurred = true;
+        // Capture the old input so we can preserve its enabled state below.
+        if (!oldInputForStreamMigration) {
+          oldInputForStreamMigration = oldInputForStream;
+        }
+      }
+      // Capture old input-level vars even when no positional old stream exists.
+      oldInputVars = oldInputForStream?.vars;
+    }
+
+    if (!oldStream && !oldInputVars) return newStream;
+
+    return migrateStreamVars(newStream as InputsOverride, oldStream, oldInputVars);
+  });
+
+  // If stream-level migration succeeded without an input-level migrate_from, carry the
+  // old input's enabled state over instead of unconditionally enabling the new input.
+  if (streamMigrationOccurred && update.migrate_from === undefined) {
+    update.enabled = oldInputForStreamMigration?.enabled ?? update.enabled;
+  }
+}
+
+export function deepMergeVars(original: any, override: any, keepOriginalValue = false): any {
+  if (!override.vars) {
+    return original;
+  }
+  if (!original.vars) {
+    original.vars = { ...override.vars };
+  }
+
+  const result = { ...original };
+
+  const overrideVars = Array.isArray(override.vars)
+    ? override.vars
+    : Object.entries(override.vars!).map(([key, rest]) => ({
+        name: key,
+        ...(rest as any),
+      }));
+
+  for (const { name, ...overrideVal } of overrideVars) {
+    const originalVar = original.vars[name];
+
+    result.vars[name] = { ...originalVar, ...overrideVal };
+
+    // Persist the original value only when it was explicitly set (not null / undefined).
+    // A null original value is treated as "not configured" so the new package default wins.
+    if (keepOriginalValue && originalVar?.value != null) {
+      result.vars[name].value = originalVar.value;
+    }
+  }
+
+  return result;
+}
+
+export function getUpdatedGlobalVars(packageInfo: PackageInfo, packagePolicy: NewPackagePolicy) {
+  if (!packageInfo.vars) {
+    return undefined;
+  }
+
+  const packageInfoVars = packageInfo.vars.reduce(varsReducer, {});
+  const result = deepMergeVars(packagePolicy, { vars: packageInfoVars }, true);
+  return removeStaleVars(result, { vars: packageInfoVars }).vars;
+}
+
+export interface SupportsVars {
+  vars?: PackagePolicyConfigRecord;
+}
+
+export function removeStaleVars<T extends SupportsVars>(
+  currentWithVars: T,
+  expectedVars: SupportsVars
+): T {
+  if (!currentWithVars.vars) {
+    return currentWithVars;
+  }
+
+  if (!expectedVars.vars) {
+    return {
+      ...currentWithVars,
+      vars: {},
+    };
+  }
+
+  const filteredVars = Object.entries(currentWithVars.vars).reduce<PackagePolicyConfigRecord>(
+    (acc, [key, val]) => {
+      if (key in expectedVars.vars!) {
+        acc[key] = val;
+      }
+      return acc;
+    },
+    {}
+  );
+
+  return {
+    ...currentWithVars,
+    vars: filteredVars,
+  };
+}
+
+/**
+ * Replaces null/undefined values on bool-typed vars with `false` after migration.
+ *
+ * Priority (from highest to lowest):
+ *  1. Old variable value (carried over by deepMergeVars when non-null)
+ *  2. New package default (used by deepMergeVars when old value is null)
+ *  3. `false` — guaranteed fallback for bool vars so the compiled agent YAML never
+ *     contains an explicit `null` for a boolean field.
+ */
+export function sanitizeMigratedVars<T extends SupportsVars>(obj: T): T {
+  if (!obj.vars) return obj;
+  return {
+    ...obj,
+    vars: Object.fromEntries(
+      Object.entries(obj.vars).map(([k, v]) => [
+        k,
+        v.type === 'bool' && v.value == null ? { ...v, value: false } : v,
+      ])
+    ),
+  };
+}
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.test.ts b/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.test.ts
index be4cef459e3ec..a4af2fd8d47c0 100644
--- a/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.test.ts
+++ b/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.test.ts
@@ -1758,5 +1758,222 @@ describe('Package policy secrets', () => {
         expect(result.secretsToDelete).toHaveLength(6);
       });
     });
+
+    describe('when secret vars are migrated from an input type with migrate_from to another', () => {
+      // New package: only `cel` input, with a secret var `api_key`.
+      // The old `httpjson` input is gone, so getPolicySecretPaths on the old policy
+      // returns [] because the package no longer declares httpjson secrets.
+      // The migrated cel policy carries the old secret reference in `api_key`.
+      const newCelPackageInfo = {
+        name: 'test-package',
+        title: 'Test Package',
+        version: '2.0.0',
+        description: 'description',
+        type: 'integration',
+        status: 'not_installed',
+        data_streams: [
+          {
+            dataset: 'test_package.cel_log',
+            streams: [
+              {
+                input: 'cel',
+                title: 'CEL',
+                vars: [{ name: 'tags', type: 'text', secret: false }],
+              },
+            ],
+          },
+        ],
+        policy_templates: [
+          {
+            name: 'template_1',
+            title: 'Template 1',
+            description: 'Template 1',
+            inputs: [
+              {
+                type: 'cel',
+                title: 'CEL',
+                description: 'CEL Input',
+                vars: [{ name: 'api_key', type: 'text', secret: true }],
+              },
+            ],
+          },
+        ],
+      } as unknown as PackageInfo;
+
+      it('does not create a new secret when the migrated value is already a secret reference', async () => {
+        const oldHttpjsonPolicy = {
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { id: 'httpjson-secret-id', isSecretRef: true } },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as PackagePolicy;
+
+        const migratedCelPolicy = {
+          inputs: [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { id: 'httpjson-secret-id', isSecretRef: true } },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as UpdatePackagePolicy;
+
+        const result = await extractAndUpdateSecrets({
+          oldPackagePolicy: oldHttpjsonPolicy,
+          packagePolicyUpdate: migratedCelPolicy,
+          packageInfo: newCelPackageInfo,
+          esClient: esClientMock,
+        });
+
+        // No new Elasticsearch secret should be created — the existing one is reused.
+        expect(esClientMock.transport.request).not.toHaveBeenCalled();
+
+        // The original secret reference must be tracked so the policy retains it.
+        expect(result.secretReferences).toEqual([{ id: 'httpjson-secret-id' }]);
+
+        // Nothing to delete — the old secret is still in use by the new policy.
+        expect(result.secretsToDelete).toHaveLength(0);
+
+        // The migrated policy's api_key value should be unchanged.
+        expect((result.packagePolicyUpdate.inputs[0].vars as any).api_key.value).toEqual({
+          id: 'httpjson-secret-id',
+          isSecretRef: true,
+        });
+      });
+
+      it('handles multi-value migrated secret references without creating new secrets', async () => {
+        const oldHttpjsonPolicy = {
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { ids: ['secret-id-1', 'secret-id-2'], isSecretRef: true } },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as PackagePolicy;
+
+        const migratedCelPolicy = {
+          inputs: [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { ids: ['secret-id-1', 'secret-id-2'], isSecretRef: true } },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as UpdatePackagePolicy;
+
+        const result = await extractAndUpdateSecrets({
+          oldPackagePolicy: oldHttpjsonPolicy,
+          packagePolicyUpdate: migratedCelPolicy,
+          packageInfo: newCelPackageInfo,
+          esClient: esClientMock,
+        });
+
+        expect(esClientMock.transport.request).not.toHaveBeenCalled();
+        expect(result.secretReferences).toEqual([{ id: 'secret-id-1' }, { id: 'secret-id-2' }]);
+        expect(result.secretsToDelete).toHaveLength(0);
+      });
+
+      it('handles a mix: migrated secret reference alongside a new plaintext secret', async () => {
+        // New package has two secret vars on the cel input.
+        const newCelPackageInfoTwoSecrets = {
+          ...newCelPackageInfo,
+          policy_templates: [
+            {
+              name: 'template_1',
+              title: 'Template 1',
+              description: 'Template 1',
+              inputs: [
+                {
+                  type: 'cel',
+                  title: 'CEL',
+                  description: 'CEL Input',
+                  vars: [
+                    { name: 'api_key', type: 'text', secret: true },
+                    { name: 'token', type: 'text', secret: true },
+                  ],
+                },
+              ],
+            },
+          ],
+        } as unknown as PackageInfo;
+
+        const oldHttpjsonPolicy = {
+          inputs: [
+            {
+              type: 'httpjson',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { id: 'httpjson-secret-id', isSecretRef: true } },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as PackagePolicy;
+
+        // api_key is migrated (already a reference); token is a new plaintext value.
+        const migratedCelPolicy = {
+          inputs: [
+            {
+              type: 'cel',
+              policy_template: 'template_1',
+              enabled: true,
+              vars: {
+                api_key: { value: { id: 'httpjson-secret-id', isSecretRef: true } },
+                token: { value: 'my-new-token' },
+              },
+              streams: [],
+            },
+          ],
+        } as unknown as UpdatePackagePolicy;
+
+        const result = await extractAndUpdateSecrets({
+          oldPackagePolicy: oldHttpjsonPolicy,
+          packagePolicyUpdate: migratedCelPolicy,
+          packageInfo: newCelPackageInfoTwoSecrets,
+          esClient: esClientMock,
+        });
+
+        // Only the new plaintext `token` should trigger a secret creation.
+        expect(esClientMock.transport.request).toHaveBeenCalledTimes(1);
+
+        // Both the migrated reference and the newly created token should be tracked.
+        expect(result.secretReferences).toHaveLength(2);
+        expect(result.secretReferences).toContainEqual({ id: 'httpjson-secret-id' });
+
+        // The new token var should have been replaced with a secret reference.
+        expect((result.packagePolicyUpdate.inputs[0].vars as any).token.value.isSecretRef).toBe(
+          true
+        );
+
+        // The api_key should still hold the original reference, unchanged.
+        expect((result.packagePolicyUpdate.inputs[0].vars as any).api_key.value).toEqual({
+          id: 'httpjson-secret-id',
+          isSecretRef: true,
+        });
+
+        expect(result.secretsToDelete).toHaveLength(0);
+      });
+    });
   });
 });
diff --git a/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.ts b/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.ts
index 75373e739cbee..03e59698af77b 100644
--- a/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.ts
+++ b/x-pack/platform/plugins/shared/fleet/server/services/secrets/package_policies.ts
@@ -115,7 +115,14 @@ export async function extractAndUpdateSecrets(opts: {
 
   const { toCreate, toDelete, noChange } = diffSecretPaths(oldSecretPaths, updatedSecretPaths);
 
-  const secretsToCreate = toCreate.filter((secretPath) => !!secretPath.value.value);
+  // Handle the case of a secret being migrated from a different input type:
+  // the old input no longer exists in the new package, so `oldSecretPaths` is empty and `diffSecretPaths` puts
+  // the path in `toCreate`. We shouldn't create a new secret from an object value,
+  // instead, preserve the existing reference exactly as-is.
+  const secretsToCreate = toCreate.filter(
+    (secretPath) => !!secretPath.value.value && !secretPath.value.value?.isSecretRef
+  );
+  const existingSecretRefs = toCreate.filter((secretPath) => !!secretPath.value.value?.isSecretRef);
 
   const createdSecrets = await createSecrets({
     esClient,
@@ -141,6 +148,14 @@ export async function extractAndUpdateSecrets(opts: {
       }
       return [...acc, { id: secret.id }];
     }, []),
+    // Migrated secret references (isSecretRef already set): carry them forward
+    // as tracked references without re-creating the underlying Elasticsearch secret.
+    ...existingSecretRefs.reduce((acc: SecretReference[], secretPath) => {
+      if (secretPath.value.value.ids) {
+        return [...acc, ...secretPath.value.value.ids.map((id: string) => ({ id }))];
+      }
+      return [...acc, { id: secretPath.value.value.id }];
+    }, []),
   ];
 
   const secretsToDelete: SecretReference[] = [];