Stop variables export/import from silently corrupting values

airflow-core/src/airflow/cli/commands/variable_command.py

@@ -186,11 +186,16 @@ def variables_export(args):
 
         data = json.JSONDecoder()
         for var in qry:
+            # Emit a form variables_import turns back into var.val: it stores strings verbatim,
+            # JSON-encodes everything else, and unwraps any dict carrying a "value" key.
             try:
                 val = data.decode(var.val)
             except Exception:
                 val = var.val
-            if var.description:
+            else:
+                if isinstance(val, str):
+                    val = var.val
+            if var.description or (isinstance(val, dict) and "value" in val):
                 var_dict[var.key] = {
                     "value": val,
                     "description": var.description,

airflow-core/tests/unit/cli/commands/test_variable_command.py

@@ -360,6 +360,40 @@ def test_variables_export(self):
         """Test variables_export command"""
         variable_command.variables_export(self.parser.parse_args(["variables", "export", os.devnull]))
 
+    @pytest.mark.parametrize(
+        ("stored_value", "expected_export"),
+        [
+            pytest.param(
+                '{"value": "a", "description": "b"}',
+                {"value": {"value": "a", "description": "b"}, "description": None},
+                id="envelope_lookalike",
+            ),
+            pytest.param(
+                '{"value": 1, "other": 2}',
+                {"value": {"value": 1, "other": 2}, "description": None},
+                id="envelope_lookalike_with_extra_keys",
+            ),
+            pytest.param('"hello"', '"hello"', id="json_string"),
+        ],
+    )
+    def test_variables_export_survives_reimport(self, tmp_path, stored_value, expected_export):
+        """Values that collide with the export format must round-trip through export/import."""
+        path = tmp_path / "variables.json"
+        variable_command.variables_set(self.parser.parse_args(["variables", "set", "k", stored_value]))
+        variable_command.variables_export(self.parser.parse_args(["variables", "export", os.fspath(path)]))
+
+        assert json.loads(path.read_text()) == {"k": expected_export}
+
+        variable_command.variables_delete(self.parser.parse_args(["variables", "delete", "k"]))
+        with create_session() as session:
+            variable_command.variables_import(
+                self.parser.parse_args(["variables", "import", os.fspath(path)]), session=session
+            )
+
+        assert Variable.get("k", deserialize_json=True) == json.loads(stored_value)
+        with create_session() as session:
+            assert session.scalar(select(Variable.description).where(Variable.key == "k")) is None
+
     def test_variables_isolation(self, tmp_path):
         """Test isolation of variables"""
         path1 = tmp_path / "testfile1.json"