diff --git a/amber/src/main/python/core/models/state.py b/amber/src/main/python/core/models/state.py
index 7e47fa98330..ce0818b64eb 100644
--- a/amber/src/main/python/core/models/state.py
+++ b/amber/src/main/python/core/models/state.py
@@ -82,6 +82,7 @@ def from_tuple(cls, row: Tuple) -> "State":
 _TYPE_MARKER = "__texera_type__"
 _PAYLOAD_MARKER = "payload"
 _BYTES_TYPE = "bytes"
+_DICT_TYPE = "dict"
 
 
 def _to_json_value(value: Any) -> Any:
@@ -95,7 +96,10 @@ def _to_json_value(value: Any) -> Any:
     if isinstance(value, dict):
         if any(not isinstance(key, str) for key in value):
             raise TypeError("State dictionary keys must be strings")
-        return {key: _to_json_value(inner) for key, inner in value.items()}
+        encoded = {key: _to_json_value(inner) for key, inner in value.items()}
+        if _TYPE_MARKER in encoded:
+            return {_TYPE_MARKER: _DICT_TYPE, _PAYLOAD_MARKER: encoded}
+        return encoded
     if isinstance(value, (list, tuple)):
         return [_to_json_value(inner) for inner in value]
     raise TypeError(
@@ -107,7 +111,13 @@ def _from_json_value(value: Any) -> Any:
     if isinstance(value, list):
         return [_from_json_value(inner) for inner in value]
     if isinstance(value, dict):
-        if value.get(_TYPE_MARKER) == _BYTES_TYPE:
+        marker_keys = {_TYPE_MARKER, _PAYLOAD_MARKER}
+        if set(value) == marker_keys and value.get(_TYPE_MARKER) == _BYTES_TYPE:
             return base64.b64decode(value[_PAYLOAD_MARKER])
+        if set(value) == marker_keys and value.get(_TYPE_MARKER) == _DICT_TYPE:
+            return {
+                key: _from_json_value(inner)
+                for key, inner in value[_PAYLOAD_MARKER].items()
+            }
         return {key: _from_json_value(inner) for key, inner in value.items()}
     return value
diff --git a/amber/src/test/python/core/models/test_state.py b/amber/src/test/python/core/models/test_state.py
index 19afab9e949..81d239a3628 100644
--- a/amber/src/test/python/core/models/test_state.py
+++ b/amber/src/test/python/core/models/test_state.py
@@ -67,6 +67,19 @@ def test_json_round_trip_nested_dict(self):
         decoded = State.from_json(original.to_json())
         assert decoded == original
 
+    @pytest.mark.parametrize(
+        "user_value",
+        [
+            {"__texera_type__": "bytes", "payload": "aGVsbG8="},
+            {"__texera_type__": "bytes", "payload": "aGVsbG8=", "owner": "user"},
+            {"__texera_type__": "dict", "payload": {"nested": 1}},
+        ],
+    )
+    def test_json_round_trip_preserves_user_type_markers(self, user_value):
+        original = State({"config": user_value})
+
+        assert State.from_json(original.to_json()) == original
+
     def test_json_round_trip_list_of_mixed_values(self):
         original = State({"items": [1, "two", 3.0, True, None]})
         decoded = State.from_json(original.to_json())