diff --git a/v3/changelog.md b/v3/changelog.md
index b7f1de9139f..54a5b0e0846 100644
--- a/v3/changelog.md
+++ b/v3/changelog.md
@@ -4,6 +4,8 @@
 
 ### Added
 
+- Support array values in session variables
+
 ### Fixed
 
 ### Changed
diff --git a/v3/crates/auth/hasura-authn-core/src/lib.rs b/v3/crates/auth/hasura-authn-core/src/lib.rs
index fb53409e491..0dfd610892f 100644
--- a/v3/crates/auth/hasura-authn-core/src/lib.rs
+++ b/v3/crates/auth/hasura-authn-core/src/lib.rs
@@ -23,7 +23,7 @@ use std::{
 // Session variable and role are defined as part of OpenDD
 pub use open_dds::{
     permissions::Role,
-    session_variables::{SessionVariable, SESSION_VARIABLE_ROLE},
+    session_variables::{SessionVariableName, SessionVariableReference, SESSION_VARIABLE_ROLE},
 };
 
 #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize, serde::Deserialize, JsonSchema)]
@@ -37,10 +37,10 @@ impl SessionVariableValue {
 }
 
 #[derive(Clone, Debug, Eq, PartialEq, serde::Serialize)]
-pub struct SessionVariables(HashMap<SessionVariable, SessionVariableValue>);
+pub struct SessionVariables(HashMap<SessionVariableName, SessionVariableValue>);
 
 impl SessionVariables {
-    pub fn get(&self, session_variable: &SessionVariable) -> Option<&SessionVariableValue> {
+    pub fn get(&self, session_variable: &SessionVariableName) -> Option<&SessionVariableValue> {
         self.0.get(session_variable)
     }
 }
@@ -56,14 +56,14 @@ pub struct Session {
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub struct RoleAuthorization {
     pub role: Role,
-    pub session_variables: HashMap<SessionVariable, SessionVariableValue>,
+    pub session_variables: HashMap<SessionVariableName, SessionVariableValue>,
     pub allowed_session_variables_from_request: SessionVariableList,
 }
 
 impl RoleAuthorization {
     pub fn build_session(
         &self,
-        mut variables: HashMap<SessionVariable, SessionVariableValue>,
+        mut variables: HashMap<SessionVariableName, SessionVariableValue>,
     ) -> Session {
         let allowed_client_session_variables = match &self.allowed_session_variables_from_request {
             SessionVariableList::All => variables,
@@ -87,7 +87,7 @@ pub enum SessionVariableList {
     // like * in Select * from ...
     All,
     // An explicit list
-    Some(HashSet<SessionVariable>),
+    Some(HashSet<SessionVariableName>),
 }
 
 // Privileges of the current user
@@ -183,7 +183,7 @@ pub fn authorize_identity(
     let mut role = None;
     // traverse through the headers and collect role and session variables
     for (header_name, header_value) in headers {
-        if let Ok(session_variable) = SessionVariable::from_str(header_name.as_str()) {
+        if let Ok(session_variable) = SessionVariableName::from_str(header_name.as_str()) {
             let variable_value = match header_value.to_str() {
                 Err(e) => Err(SessionError::InvalidHeaderValue {
                     header_name: header_name.to_string(),
@@ -220,16 +220,16 @@ mod tests {
         let mut authenticated_session_variables = HashMap::new();
 
         authenticated_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
 
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom").unwrap(),
             SessionVariableValue::new("test"),
         );
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom-claim").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom-claim").unwrap(),
             SessionVariableValue::new("claim-value"),
         );
         let role_authorization = RoleAuthorization {
@@ -243,7 +243,7 @@ mod tests {
         let mut expected_session_variables = client_session_variables.clone();
 
         expected_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         pa::assert_eq!(
@@ -262,23 +262,23 @@ mod tests {
         let mut authenticated_session_variables = HashMap::new();
 
         authenticated_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
 
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom").unwrap(),
             SessionVariableValue::new("test"),
         );
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom-claim").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom-claim").unwrap(),
             SessionVariableValue::new("claim-value"),
         );
         let mut allowed_sesion_variables_from_request = HashSet::new();
         allowed_sesion_variables_from_request
-            .insert(SessionVariable::from_str("x-hasura-custom").unwrap());
+            .insert(SessionVariableName::from_str("x-hasura-custom").unwrap());
         allowed_sesion_variables_from_request
-            .insert(SessionVariable::from_str("x-hasura-custom-author-id").unwrap());
+            .insert(SessionVariableName::from_str("x-hasura-custom-author-id").unwrap());
         let role_authorization = RoleAuthorization {
             role: Role::new("test-role"),
             session_variables: authenticated_session_variables,
@@ -291,10 +291,10 @@ mod tests {
 
         let mut expected_session_variables = client_session_variables;
         expected_session_variables
-            .remove(&SessionVariable::from_str("x-hasura-custom-claim").unwrap());
+            .remove(&SessionVariableName::from_str("x-hasura-custom-claim").unwrap());
 
         expected_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         pa::assert_eq!(
@@ -313,16 +313,16 @@ mod tests {
         let mut authenticated_session_variables = HashMap::new();
 
         authenticated_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
 
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom").unwrap(),
             SessionVariableValue::new("test"),
         );
         client_session_variables.insert(
-            SessionVariable::from_str("x-hasura-custom-claim").unwrap(),
+            SessionVariableName::from_str("x-hasura-custom-claim").unwrap(),
             SessionVariableValue::new("claim-value"),
         );
 
@@ -337,7 +337,7 @@ mod tests {
         let mut expected_session_variables = HashMap::new();
 
         expected_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
 
diff --git a/v3/crates/auth/hasura-authn-jwt/src/auth.rs b/v3/crates/auth/hasura-authn-jwt/src/auth.rs
index 57a1bc6fb4c..03269cbd7a2 100644
--- a/v3/crates/auth/hasura-authn-jwt/src/auth.rs
+++ b/v3/crates/auth/hasura-authn-jwt/src/auth.rs
@@ -110,10 +110,11 @@ pub async fn authenticate_request(
 mod tests {
     use std::str::FromStr;
 
-    use auth_base::{RoleAuthorization, SessionVariable, SessionVariableValue};
+    use auth_base::{RoleAuthorization, SessionVariableValue};
     use jsonwebtoken as jwt;
     use jsonwebtoken::Algorithm;
     use jwt::{encode, EncodingKey};
+    use open_dds::session_variables::SessionVariableName;
     use reqwest::header::AUTHORIZATION;
     use serde_json::json;
     use tokio;
@@ -166,7 +167,7 @@ mod tests {
     fn get_default_hasura_claims() -> HasuraClaims {
         let mut hasura_custom_claims = HashMap::new();
         hasura_custom_claims.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue("1".to_string()),
         );
         HasuraClaims {
@@ -223,7 +224,7 @@ mod tests {
         let mut role_authorization_session_variables = HashMap::new();
 
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         expected_allowed_roles.insert(
@@ -272,7 +273,7 @@ mod tests {
     async fn test_successful_role_emulation() -> anyhow::Result<()> {
         let mut hasura_claims = get_default_hasura_claims();
         hasura_claims.custom_claims.insert(
-            SessionVariable::from_str("x-hasura-role").unwrap(),
+            SessionVariableName::from_str("x-hasura-role").unwrap(),
             SessionVariableValue::new("admin"),
         );
         let encoded_claims = get_encoded_claims(Algorithm::HS256, &hasura_claims)?;
diff --git a/v3/crates/auth/hasura-authn-jwt/src/jwt.rs b/v3/crates/auth/hasura-authn-jwt/src/jwt.rs
index 85dd1d226f3..18c3e713c66 100644
--- a/v3/crates/auth/hasura-authn-jwt/src/jwt.rs
+++ b/v3/crates/auth/hasura-authn-jwt/src/jwt.rs
@@ -4,10 +4,11 @@ use std::time::Duration;
 use axum::http::{HeaderMap, HeaderValue};
 use axum::response::IntoResponse;
 use cookie::{self, Cookie};
-use hasura_authn_core::{Role, SessionVariable, SessionVariableValue};
+use hasura_authn_core::{Role, SessionVariableValue};
 use jsonptr::Pointer;
 use jsonwebtoken::{self as jwt, decode, DecodingKey, Validation};
 use jwt::decode_header;
+use open_dds::session_variables::SessionVariableName;
 use reqwest::header::{AUTHORIZATION, COOKIE};
 use reqwest::StatusCode;
 use schemars::gen::SchemaGenerator;
@@ -253,7 +254,7 @@ pub struct JWTClaimsMap {
     /// A dictionary of the custom claims, where the key is the name of the claim and the value
     /// is the JSON pointer to lookup the custom claims within the decoded JWT.
     pub custom_claims:
-        Option<HashMap<SessionVariable, JWTClaimsMappingEntry<SessionVariableValue>>>,
+        Option<HashMap<SessionVariableName, JWTClaimsMappingEntry<SessionVariableValue>>>,
 }
 
 #[derive(Serialize, Deserialize, PartialEq, Clone, JsonSchema, Debug)]
@@ -390,7 +391,7 @@ pub struct HasuraClaims {
     /// as per the user's defined permissions.
     /// For example, things like `x-hasura-user-id` can go here.
     #[serde(flatten)]
-    pub custom_claims: HashMap<SessionVariable, SessionVariableValue>,
+    pub custom_claims: HashMap<SessionVariableName, SessionVariableValue>,
 }
 
 #[derive(Debug, Serialize, Deserialize)]
@@ -733,7 +734,7 @@ mod tests {
     fn get_default_hasura_claims() -> HasuraClaims {
         let mut hasura_custom_claims = HashMap::new();
         hasura_custom_claims.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue("1".to_string()),
         );
         HasuraClaims {
diff --git a/v3/crates/auth/hasura-authn-noauth/src/lib.rs b/v3/crates/auth/hasura-authn-noauth/src/lib.rs
index 3eda2b4df12..4a35c8ac3e1 100644
--- a/v3/crates/auth/hasura-authn-noauth/src/lib.rs
+++ b/v3/crates/auth/hasura-authn-noauth/src/lib.rs
@@ -1,4 +1,4 @@
-use hasura_authn_core::{Role, SessionVariable, SessionVariableValue};
+use hasura_authn_core::{Role, SessionVariableName, SessionVariableValue};
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 use serde_json::json;
@@ -15,7 +15,7 @@ pub struct NoAuthConfig {
     pub role: Role,
     /// static session variables to use whilst running the engine
     #[schemars(title = "SessionVariables")]
-    pub session_variables: HashMap<SessionVariable, SessionVariableValue>,
+    pub session_variables: HashMap<SessionVariableName, SessionVariableValue>,
 }
 
 impl NoAuthConfig {
diff --git a/v3/crates/auth/hasura-authn-webhook/src/webhook.rs b/v3/crates/auth/hasura-authn-webhook/src/webhook.rs
index 9b070336b88..115079334c7 100644
--- a/v3/crates/auth/hasura-authn-webhook/src/webhook.rs
+++ b/v3/crates/auth/hasura-authn-webhook/src/webhook.rs
@@ -3,7 +3,7 @@ use std::str::FromStr;
 use std::sync::OnceLock;
 use std::time::Duration;
 
-use auth_base::{Identity, Role, RoleAuthorization, SessionVariable, SessionVariableValue};
+use auth_base::{Identity, Role, RoleAuthorization, SessionVariableName, SessionVariableValue};
 use axum::{
     http::{HeaderMap, HeaderName, StatusCode},
     response::IntoResponse,
@@ -206,7 +206,7 @@ async fn make_auth_hook_request(
                 response.json().await.map_err(InternalError::ReqwestError)?;
             let mut session_variables = HashMap::new();
             for (k, v) in &auth_hook_response {
-                match SessionVariable::from_str(k) {
+                match SessionVariableName::from_str(k) {
                     Ok(session_variable) => {
                         session_variables
                             .insert(session_variable, SessionVariableValue(v.to_string()));
@@ -366,11 +366,11 @@ mod tests {
         let mut expected_allowed_roles = HashMap::new();
         let mut role_authorization_session_variables = HashMap::new();
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-role").unwrap(),
+            SessionVariableName::from_str("x-hasura-role").unwrap(),
             SessionVariableValue::new("test-role"),
         );
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-test-role-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-test-role-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         expected_allowed_roles.insert(
@@ -437,11 +437,11 @@ mod tests {
         let mut expected_allowed_roles = HashMap::new();
         let mut role_authorization_session_variables = HashMap::new();
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-role").unwrap(),
+            SessionVariableName::from_str("x-hasura-role").unwrap(),
             SessionVariableValue::new("test-role"),
         );
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-test-role-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-test-role-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         expected_allowed_roles.insert(
@@ -509,15 +509,15 @@ mod tests {
         let mut expected_allowed_roles = HashMap::new();
         let mut role_authorization_session_variables = HashMap::new();
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-role").unwrap(),
+            SessionVariableName::from_str("x-hasura-role").unwrap(),
             SessionVariableValue::new("test-role"),
         );
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-test-role-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-test-role-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("status").unwrap(),
+            SessionVariableName::from_str("status").unwrap(),
             SessionVariableValue::new("true"),
         );
         expected_allowed_roles.insert(
@@ -637,11 +637,11 @@ mod tests {
         let mut expected_allowed_roles = HashMap::new();
         let mut role_authorization_session_variables = HashMap::new();
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-role").unwrap(),
+            SessionVariableName::from_str("x-hasura-role").unwrap(),
             SessionVariableValue::new("user"),
         );
         role_authorization_session_variables.insert(
-            SessionVariable::from_str("x-hasura-user-id").unwrap(),
+            SessionVariableName::from_str("x-hasura-user-id").unwrap(),
             SessionVariableValue::new("1"),
         );
         expected_allowed_roles.insert(
diff --git a/v3/crates/engine/tests/common.rs b/v3/crates/engine/tests/common.rs
index be35b8877d0..e656905aed1 100644
--- a/v3/crates/engine/tests/common.rs
+++ b/v3/crates/engine/tests/common.rs
@@ -7,7 +7,7 @@ use hasura_authn_core::{Identity, Role, Session, SessionError, SessionVariableVa
 use lang_graphql::ast::common as ast;
 use lang_graphql::{http::RawRequest, schema::Schema};
 use metadata_resolve::{data_connectors::NdcVersion, LifecyclePluginConfigs};
-use open_dds::session_variables::{SessionVariable, SESSION_VARIABLE_ROLE};
+use open_dds::session_variables::{SessionVariableName, SESSION_VARIABLE_ROLE};
 use pretty_assertions::assert_eq;
 use serde_json as json;
 use sql::execute::SqlRequest;
@@ -41,7 +41,7 @@ pub fn setup(test_dir: &Path) -> GoldenTestContext {
 }
 
 pub(crate) fn resolve_session(
-    session_variables: HashMap<SessionVariable, SessionVariableValue>,
+    session_variables: HashMap<SessionVariableName, SessionVariableValue>,
 ) -> Result<Session, SessionError> {
     //return an arbitrary identity with role emulation enabled
     let authorization = Identity::admin(Role::new("admin"));
@@ -107,7 +107,7 @@ pub(crate) fn test_introspection_expectation(
 
         let request_headers = reqwest::header::HeaderMap::new();
         let session_vars_path = &test_path.join("session_variables.json");
-        let sessions: Vec<HashMap<SessionVariable, SessionVariableValue>> =
+        let sessions: Vec<HashMap<SessionVariableName, SessionVariableValue>> =
             json::from_str(read_to_string(session_vars_path)?.as_ref())?;
         let sessions: Vec<Session> = sessions
             .into_iter()
@@ -266,7 +266,7 @@ pub fn test_execution_expectation_for_multiple_ndc_versions(
 
             let request_headers = reqwest::header::HeaderMap::new();
             let session_vars_path = &test_path.join("session_variables.json");
-            let sessions: Vec<HashMap<SessionVariable, SessionVariableValue>> =
+            let sessions: Vec<HashMap<SessionVariableName, SessionVariableValue>> =
                 json::from_str(read_to_string(session_vars_path)?.as_ref())?;
             let sessions: Vec<Session> = sessions
                 .into_iter()
@@ -472,7 +472,7 @@ pub fn test_execute_explain(
             let session_variables_raw = r#"{
                 "x-hasura-role": "admin"
             }"#;
-            let session_variables: HashMap<SessionVariable, SessionVariableValue> =
+            let session_variables: HashMap<SessionVariableName, SessionVariableValue> =
                 serde_json::from_str(session_variables_raw)?;
             resolve_session(session_variables)
         }?;
@@ -579,7 +579,7 @@ pub(crate) fn test_sql(test_path_string: &str) -> anyhow::Result<()> {
 
         let session = Arc::new({
             let session_vars_path = &test_path.join("session_variables.json");
-            let session_variables: HashMap<SessionVariable, SessionVariableValue> =
+            let session_variables: HashMap<SessionVariableName, SessionVariableValue> =
                 serde_json::from_str(read_to_string(session_vars_path)?.as_ref())?;
             resolve_session(session_variables)
         }?);
diff --git a/v3/crates/engine/tests/execute/models/select_many/array_session_variable/expected.json b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/expected.json
new file mode 100644
index 00000000000..92b2c8daeb8
--- /dev/null
+++ b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/expected.json
@@ -0,0 +1,11 @@
+[
+  {
+    "data": {
+      "AuthorMany": [
+        {
+          "author_id": 1
+        }
+      ]
+    }
+  }
+]
diff --git a/v3/crates/engine/tests/execute/models/select_many/array_session_variable/metadata.json b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/metadata.json
new file mode 100644
index 00000000000..bc6c860533e
--- /dev/null
+++ b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/metadata.json
@@ -0,0 +1,192 @@
+{
+  "version": "v2",
+  "flags": {
+    "json_session_variables": true
+  },
+  "subgraphs": [
+    {
+      "name": "default",
+      "objects": [
+        {
+          "kind": "BooleanExpressionType",
+          "version": "v1",
+          "definition": {
+            "name": "int_bool_exp",
+            "operand": {
+              "scalar": {
+                "type": "Int",
+                "comparisonOperators": [
+                  {
+                    "name": "_eq",
+                    "argumentType": "Int!"
+                  },
+                  {
+                    "name": "_in",
+                    "argumentType": "[Int!]"
+                  }
+                ],
+                "dataConnectorOperatorMapping": [
+                  {
+                    "dataConnectorName": "db",
+                    "dataConnectorScalarType": "int4",
+                    "operatorMapping": {}
+                  }
+                ]
+              }
+            },
+            "logicalOperators": {
+              "enable": true
+            },
+            "isNull": {
+              "enable": true
+            },
+            "graphql": {
+              "typeName": "Int_Filter"
+            }
+          }
+        },
+        {
+          "kind": "ObjectType",
+          "version": "v1",
+          "definition": {
+            "name": "author",
+            "fields": [
+              {
+                "name": "author_id",
+                "type": "Int!"
+              },
+              {
+                "name": "first_name",
+                "type": "String!"
+              },
+              {
+                "name": "last_name",
+                "type": "String!"
+              }
+            ],
+            "graphql": {
+              "typeName": "Author"
+            },
+            "dataConnectorTypeMapping": [
+              {
+                "dataConnectorName": "db",
+                "dataConnectorObjectType": "author",
+                "fieldMapping": {
+                  "author_id": {
+                    "column": {
+                      "name": "id"
+                    }
+                  },
+                  "first_name": {
+                    "column": {
+                      "name": "first_name"
+                    }
+                  },
+                  "last_name": {
+                    "column": {
+                      "name": "last_name"
+                    }
+                  }
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "BooleanExpressionType",
+          "version": "v1",
+          "definition": {
+            "name": "author_bool_exp",
+            "operand": {
+              "object": {
+                "type": "author",
+                "comparableFields": [
+                  {
+                    "fieldName": "author_id",
+                    "booleanExpressionType": "int_bool_exp"
+                  }
+                ],
+                "comparableRelationships": []
+              }
+            },
+            "logicalOperators": {
+              "enable": true
+            },
+            "isNull": {
+              "enable": true
+            },
+            "graphql": {
+              "typeName": "Author_Filter"
+            }
+          }
+        },
+        {
+          "kind": "Model",
+          "version": "v1",
+          "definition": {
+            "name": "Authors",
+            "objectType": "author",
+            "source": {
+              "dataConnectorName": "db",
+              "collection": "author"
+            },
+            "graphql": {
+              "selectUniques": [],
+              "selectMany": {
+                "queryRootField": "AuthorMany"
+              },
+              "orderByExpressionType": "Author_Order_By"
+            },
+            "filterExpressionType": "author_bool_exp",
+            "orderableFields": [
+              {
+                "fieldName": "author_id",
+                "orderByDirections": {
+                  "enableAll": true
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "TypePermissions",
+          "version": "v1",
+          "definition": {
+            "typeName": "author",
+            "permissions": [
+              {
+                "role": "user",
+                "output": {
+                  "allowedFields": ["author_id", "first_name", "last_name"]
+                }
+              }
+            ]
+          }
+        },
+        {
+          "kind": "ModelPermissions",
+          "version": "v1",
+          "definition": {
+            "modelName": "Authors",
+            "permissions": [
+              {
+                "role": "user",
+                "select": {
+                  "filter": {
+                    "fieldComparison": {
+                      "field": "author_id",
+                      "operator": "_in",
+                      "value": {
+                        "sessionVariable": "x-hasura-allowed-author-ids"
+                      }
+                    }
+                  }
+                }
+              }
+            ]
+          }
+        }
+      ]
+    }
+  ]
+}
diff --git a/v3/crates/engine/tests/execute/models/select_many/array_session_variable/request.gql b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/request.gql
new file mode 100644
index 00000000000..70c0f295d99
--- /dev/null
+++ b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/request.gql
@@ -0,0 +1,5 @@
+query MyQuery {
+  AuthorMany {
+    author_id
+  }
+}
diff --git a/v3/crates/engine/tests/execute/models/select_many/array_session_variable/session_variables.json b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/session_variables.json
new file mode 100644
index 00000000000..c81ca1a21f2
--- /dev/null
+++ b/v3/crates/engine/tests/execute/models/select_many/array_session_variable/session_variables.json
@@ -0,0 +1,6 @@
+[
+  {
+    "x-hasura-role": "user",
+    "x-hasura-allowed-author-ids": "[1]"
+  }
+]
diff --git a/v3/crates/engine/tests/execution.rs b/v3/crates/engine/tests/execution.rs
index 88b01c4fa9b..f4ef87d3c14 100644
--- a/v3/crates/engine/tests/execution.rs
+++ b/v3/crates/engine/tests/execution.rs
@@ -129,6 +129,17 @@ fn test_model_select_many_empty_select() -> anyhow::Result<()> {
     )
 }
 
+#[test]
+fn test_model_select_many_array_session_variable() -> anyhow::Result<()> {
+    let test_path_string = "execute/models/select_many/array_session_variable";
+    let common_metadata_path_string = "execute/common_metadata/postgres_connector_schema.json";
+    common::test_execution_expectation(
+        test_path_string,
+        &[common_metadata_path_string],
+        common::TestOpenDDPipeline::TestNDCResponses,
+    )
+}
+
 #[test]
 fn test_model_select_many_field_arguments() -> anyhow::Result<()> {
     common::test_execution_expectation_for_multiple_ndc_versions(
diff --git a/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable.snap b/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable.snap
new file mode 100644
index 00000000000..370b9285c6b
--- /dev/null
+++ b/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable.snap
@@ -0,0 +1,52 @@
+---
+source: crates/engine/tests/common.rs
+expression: query_ir
+---
+V1(
+    QueryRequestV1 {
+        queries: {
+            Alias(
+                Identifier(
+                    "AuthorMany",
+                ),
+            ): Model(
+                ModelSelection {
+                    target: ModelTarget {
+                        subgraph: SubgraphName(
+                            "default",
+                        ),
+                        model_name: ModelName(
+                            Identifier(
+                                "Authors",
+                            ),
+                        ),
+                        arguments: {},
+                        filter: None,
+                        order_by: [],
+                        limit: None,
+                        offset: None,
+                    },
+                    selection: {
+                        Alias(
+                            Identifier(
+                                "author_id",
+                            ),
+                        ): Field(
+                            ObjectFieldSelection {
+                                target: ObjectFieldTarget {
+                                    field_name: FieldName(
+                                        Identifier(
+                                            "author_id",
+                                        ),
+                                    ),
+                                    arguments: {},
+                                },
+                                selection: None,
+                            },
+                        ),
+                    },
+                },
+            ),
+        },
+    },
+)
diff --git a/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable_user.snap b/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable_user.snap
new file mode 100644
index 00000000000..370b9285c6b
--- /dev/null
+++ b/v3/crates/engine/tests/snapshots/execution__common__ir_execute__models__select_many__array_session_variable_user.snap
@@ -0,0 +1,52 @@
+---
+source: crates/engine/tests/common.rs
+expression: query_ir
+---
+V1(
+    QueryRequestV1 {
+        queries: {
+            Alias(
+                Identifier(
+                    "AuthorMany",
+                ),
+            ): Model(
+                ModelSelection {
+                    target: ModelTarget {
+                        subgraph: SubgraphName(
+                            "default",
+                        ),
+                        model_name: ModelName(
+                            Identifier(
+                                "Authors",
+                            ),
+                        ),
+                        arguments: {},
+                        filter: None,
+                        order_by: [],
+                        limit: None,
+                        offset: None,
+                    },
+                    selection: {
+                        Alias(
+                            Identifier(
+                                "author_id",
+                            ),
+                        ): Field(
+                            ObjectFieldSelection {
+                                target: ObjectFieldTarget {
+                                    field_name: FieldName(
+                                        Identifier(
+                                            "author_id",
+                                        ),
+                                    ),
+                                    arguments: {},
+                                },
+                                selection: None,
+                            },
+                        ),
+                    },
+                },
+            ),
+        },
+    },
+)
diff --git a/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable.snap b/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable.snap
new file mode 100644
index 00000000000..3123daf5f3b
--- /dev/null
+++ b/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable.snap
@@ -0,0 +1,13 @@
+---
+source: crates/engine/tests/common.rs
+expression: rowsets
+---
+[
+  {
+    "rows": [
+      {
+        "author_id": 1
+      }
+    ]
+  }
+]
diff --git a/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable_user.snap b/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable_user.snap
new file mode 100644
index 00000000000..901db20111a
--- /dev/null
+++ b/v3/crates/engine/tests/snapshots/execution__common__rowsets_execute__models__select_many__array_session_variable_user.snap
@@ -0,0 +1,15 @@
+---
+source: crates/engine/tests/common.rs
+expression: rowsets
+---
+{
+  "Ok": [
+    {
+      "rows": [
+        {
+          "author_id": 1
+        }
+      ]
+    }
+  ]
+}
diff --git a/v3/crates/frontends/graphql/src/lib.rs b/v3/crates/frontends/graphql/src/lib.rs
index 90485c5bd89..6be8c7174da 100644
--- a/v3/crates/frontends/graphql/src/lib.rs
+++ b/v3/crates/frontends/graphql/src/lib.rs
@@ -21,7 +21,7 @@ mod tests {
     use hasura_authn_core::{Identity, Role, Session, SessionVariableValue};
     use lang_graphql::http::Request;
     use lang_graphql::{parser::Parser, validation::normalize_request};
-    use open_dds::session_variables::{SessionVariable, SESSION_VARIABLE_ROLE};
+    use open_dds::session_variables::{SessionVariableName, SESSION_VARIABLE_ROLE};
     use serde_json as json;
     use std::{
         collections::HashMap,
@@ -160,7 +160,7 @@ mod tests {
     // TODO: remove duplication between this function and 'add_session'
     fn resolve_session(session_vars_path: PathBuf) -> Session {
         let authorization = Identity::admin(Role::new("admin"));
-        let session_variables: HashMap<SessionVariable, SessionVariableValue> = {
+        let session_variables: HashMap<SessionVariableName, SessionVariableValue> = {
             if session_vars_path.exists() {
                 json::from_str(fs::read_to_string(session_vars_path).unwrap().as_ref()).unwrap()
             } else {
diff --git a/v3/crates/graphql/ir/src/arguments.rs b/v3/crates/graphql/ir/src/arguments.rs
index a0c98397211..9c7e432a755 100644
--- a/v3/crates/graphql/ir/src/arguments.rs
+++ b/v3/crates/graphql/ir/src/arguments.rs
@@ -369,8 +369,8 @@ pub(crate) fn map_argument_value_to_ndc_type(
 #[cfg(test)]
 mod test {
     use hasura_authn_core::{
-        Role, RoleAuthorization, Session, SessionVariable, SessionVariableList,
-        SessionVariableValue,
+        Role, RoleAuthorization, Session, SessionVariableList, SessionVariableName,
+        SessionVariableReference, SessionVariableValue,
     };
     use indexmap::IndexMap;
     use metadata_resolve::http::SerializableHeaderName;
@@ -380,7 +380,7 @@ mod test {
     use std::str::FromStr;
 
     fn make_test_session(
-        client_session_variables: HashMap<SessionVariable, SessionVariableValue>,
+        client_session_variables: HashMap<SessionVariableName, SessionVariableValue>,
     ) -> Session {
         let authenticated_session_variables = HashMap::new();
 
@@ -459,7 +459,10 @@ mod test {
         let mut additional = IndexMap::new();
         additional.insert(
             SerializableHeaderName::new("name".into()).unwrap(),
-            ValueExpression::SessionVariable(SessionVariable::from_str("x-name").unwrap()),
+            ValueExpression::SessionVariable(SessionVariableReference {
+                name: SessionVariableName::from_str("x-name").unwrap(),
+                passed_as_json: false,
+            }),
         );
         let http_headers = HttpHeadersPreset {
             forward: vec![],
@@ -471,7 +474,7 @@ mod test {
         // what session variables do we have?
         let mut client_session_variables = HashMap::new();
         client_session_variables.insert(
-            SessionVariable::from_str("x-name").unwrap(),
+            SessionVariableName::from_str("x-name").unwrap(),
             SessionVariableValue::new("Mr Horse"),
         );
 
diff --git a/v3/crates/graphql/ir/src/error.rs b/v3/crates/graphql/ir/src/error.rs
index 81ba8bb2818..804e1f7ae66 100644
--- a/v3/crates/graphql/ir/src/error.rs
+++ b/v3/crates/graphql/ir/src/error.rs
@@ -5,7 +5,7 @@ use open_dds::{
     arguments::ArgumentName,
     data_connector::{DataConnectorColumnName, DataConnectorName},
     relationships::RelationshipName,
-    session_variables::SessionVariable,
+    session_variables::SessionVariableName,
     types::{CustomTypeName, FieldName},
 };
 use serde_json as json;
@@ -118,7 +118,9 @@ pub enum InternalDeveloperError {
     },
 
     #[error("Required session variable not found in the request: {session_variable}")]
-    MissingSessionVariable { session_variable: SessionVariable },
+    MissingSessionVariable {
+        session_variable: SessionVariableName,
+    },
 
     #[error("Unable to typecast session variable. Expected: {expected:}, but found: {found:}")]
     VariableTypeCast { expected: String, found: String },
diff --git a/v3/crates/graphql/ir/src/permissions.rs b/v3/crates/graphql/ir/src/permissions.rs
index 0c5a9c20cab..b627fa3656e 100644
--- a/v3/crates/graphql/ir/src/permissions.rs
+++ b/v3/crates/graphql/ir/src/permissions.rs
@@ -223,13 +223,13 @@ pub(crate) fn make_argument_from_value_expression(
     match val_expr {
         metadata_resolve::ValueExpression::Literal(val) => Ok(val.clone()),
         metadata_resolve::ValueExpression::SessionVariable(session_var) => {
-            let value = session_variables.get(session_var).ok_or_else(|| {
+            let value = session_variables.get(&session_var.name).ok_or_else(|| {
                 error::InternalDeveloperError::MissingSessionVariable {
-                    session_variable: session_var.clone(),
+                    session_variable: session_var.name.clone(),
                 }
             })?;
 
-            typecast_session_variable(value, value_type)
+            typecast_session_variable(session_var.passed_as_json, value, value_type)
         }
     }
 }
@@ -247,14 +247,14 @@ pub(crate) fn make_argument_from_value_expression_or_predicate<'s>(
             Ok(Argument::Literal { value: val.clone() })
         }
         metadata_resolve::ValueExpressionOrPredicate::SessionVariable(session_var) => {
-            let value = session_variables.get(session_var).ok_or_else(|| {
+            let value = session_variables.get(&session_var.name).ok_or_else(|| {
                 error::InternalDeveloperError::MissingSessionVariable {
-                    session_variable: session_var.clone(),
+                    session_variable: session_var.name.clone(),
                 }
             })?;
 
             Ok(Argument::Literal {
-                value: typecast_session_variable(value, value_type)?,
+                value: typecast_session_variable(session_var.passed_as_json, value, value_type)?,
             })
         }
         metadata_resolve::ValueExpressionOrPredicate::BooleanExpression(model_predicate) => {
@@ -274,6 +274,18 @@ pub(crate) fn make_argument_from_value_expression_or_predicate<'s>(
 
 /// Typecast a stringified session variable into a given type, but as a serde_json::Value
 fn typecast_session_variable(
+    passed_as_json: bool,
+    session_var_value_wrapped: &SessionVariableValue,
+    to_type: &QualifiedTypeReference,
+) -> Result<serde_json::Value, error::Error> {
+    if passed_as_json {
+        typecast_session_variable_v2(session_var_value_wrapped, to_type)
+    } else {
+        typecast_session_variable_v1(session_var_value_wrapped, to_type)
+    }
+}
+
+fn typecast_session_variable_v1(
     session_var_value_wrapped: &SessionVariableValue,
     to_type: &QualifiedTypeReference,
 ) -> Result<serde_json::Value, error::Error> {
@@ -328,3 +340,73 @@ fn typecast_session_variable(
         QualifiedBaseType::List(_) => Err(error::InternalDeveloperError::VariableArrayTypeCast)?,
     }
 }
+
+fn typecast_session_variable_v2(
+    session_var_value: &SessionVariableValue,
+    to_type: &QualifiedTypeReference,
+) -> Result<serde_json::Value, error::Error> {
+    let value = serde_json::from_str(&session_var_value.0)?;
+    typecheck_session_variable(&value, to_type)?;
+    Ok(value)
+}
+
+fn typecheck_session_variable(
+    value: &serde_json::Value,
+    to_type: &QualifiedTypeReference,
+) -> Result<(), crate::Error> {
+    match &to_type.underlying_type {
+        QualifiedBaseType::Named(type_name) => match type_name {
+            QualifiedTypeName::Inbuilt(primitive) => match primitive {
+                InbuiltType::Int => {
+                    if !value.is_i64() {
+                        Err(error::InternalDeveloperError::VariableTypeCast {
+                            expected: "int".into(),
+                            found: value.to_string(),
+                        })?;
+                    }
+                    Ok(())
+                }
+                InbuiltType::Float => {
+                    if !value.is_f64() {
+                        Err(error::InternalDeveloperError::VariableTypeCast {
+                            expected: "float".into(),
+                            found: value.to_string(),
+                        })?;
+                    }
+                    Ok(())
+                }
+                InbuiltType::Boolean => {
+                    if !value.is_boolean() {
+                        Err(error::InternalDeveloperError::VariableTypeCast {
+                            expected: "true or false".into(),
+                            found: value.to_string(),
+                        })?;
+                    }
+                    Ok(())
+                }
+                InbuiltType::ID | InbuiltType::String => {
+                    if !value.is_string() {
+                        Err(error::InternalDeveloperError::VariableTypeCast {
+                            expected: "string".into(),
+                            found: value.to_string(),
+                        })?;
+                    }
+                    Ok(())
+                }
+            },
+            QualifiedTypeName::Custom(_) => Ok(()),
+        },
+        QualifiedBaseType::List(element_type) => {
+            let elements = value.as_array().ok_or_else(|| {
+                error::InternalDeveloperError::VariableTypeCast {
+                    expected: "array".into(),
+                    found: value.to_string(),
+                }
+            })?;
+            for element in elements {
+                typecheck_session_variable(element, element_type)?;
+            }
+            Ok(())
+        }
+    }
+}
diff --git a/v3/crates/metadata-resolve/src/helpers/argument.rs b/v3/crates/metadata-resolve/src/helpers/argument.rs
index 2258b0894a2..3f8b17b9df8 100644
--- a/v3/crates/metadata-resolve/src/helpers/argument.rs
+++ b/v3/crates/metadata-resolve/src/helpers/argument.rs
@@ -253,6 +253,7 @@ pub fn get_argument_mappings<'a>(
 /// type to validate it against to ensure the fields it refers to
 /// exist etc
 pub(crate) fn resolve_value_expression_for_argument(
+    flags: &open_dds::flags::Flags,
     argument_name: &open_dds::arguments::ArgumentName,
     value_expression: &open_dds::permissions::ValueExpressionOrPredicate,
     argument_type: &QualifiedTypeReference,
@@ -278,7 +279,10 @@ pub(crate) fn resolve_value_expression_for_argument(
     match value_expression {
         open_dds::permissions::ValueExpressionOrPredicate::SessionVariable(session_variable) => {
             Ok::<ValueExpressionOrPredicate, Error>(ValueExpressionOrPredicate::SessionVariable(
-                session_variable.clone(),
+                hasura_authn_core::SessionVariableReference {
+                    name: session_variable.clone(),
+                    passed_as_json: flags.json_session_variables,
+                },
             ))
         }
         open_dds::permissions::ValueExpressionOrPredicate::Literal(json_value) => {
@@ -365,6 +369,7 @@ pub(crate) fn resolve_value_expression_for_argument(
                 })?;
 
             let resolved_model_predicate = model_permissions::resolve_model_predicate_with_type(
+                flags,
                 bool_exp,
                 base_type,
                 object_type_representation,
diff --git a/v3/crates/metadata-resolve/src/stages/argument_presets/presets.rs b/v3/crates/metadata-resolve/src/stages/argument_presets/presets.rs
index 014567bc22d..1dc5cfa3ad1 100644
--- a/v3/crates/metadata-resolve/src/stages/argument_presets/presets.rs
+++ b/v3/crates/metadata-resolve/src/stages/argument_presets/presets.rs
@@ -304,17 +304,7 @@ fn build_preset_map_from_input_object_type_permission(
                 field_path: new_field_path,
             };
 
-            let value = (
-                type_reference.clone(),
-                match &preset.value {
-                    open_dds::permissions::ValueExpression::Literal(literal) => {
-                        ValueExpressionOrPredicate::Literal(literal.clone())
-                    }
-                    open_dds::permissions::ValueExpression::SessionVariable(session_variable) => {
-                        ValueExpressionOrPredicate::SessionVariable(session_variable.clone())
-                    }
-                },
-            );
+            let value = (type_reference.clone(), preset.value.clone());
 
             Ok((key, value))
         })
diff --git a/v3/crates/metadata-resolve/src/stages/command_permissions/command_permission.rs b/v3/crates/metadata-resolve/src/stages/command_permissions/command_permission.rs
index 9ff3b932b13..01e864c55f9 100644
--- a/v3/crates/metadata-resolve/src/stages/command_permissions/command_permission.rs
+++ b/v3/crates/metadata-resolve/src/stages/command_permissions/command_permission.rs
@@ -38,6 +38,7 @@ fn get_command_source_argument<'a>(
 }
 
 pub fn resolve_command_permissions(
+    flags: &open_dds::flags::Flags,
     command: &commands::Command,
     permissions: &CommandPermissionsV1,
     object_types: &BTreeMap<
@@ -97,6 +98,7 @@ pub fn resolve_command_permissions(
             match command.arguments.get(&argument_preset.argument) {
                 Some(argument) => {
                     let value_expression = resolve_value_expression_for_argument(
+                        flags,
                         &argument_preset.argument,
                         &argument_preset.value,
                         &argument.argument_type,
diff --git a/v3/crates/metadata-resolve/src/stages/command_permissions/mod.rs b/v3/crates/metadata-resolve/src/stages/command_permissions/mod.rs
index 2ce479f5b91..36c999a3fe7 100644
--- a/v3/crates/metadata-resolve/src/stages/command_permissions/mod.rs
+++ b/v3/crates/metadata-resolve/src/stages/command_permissions/mod.rs
@@ -67,6 +67,7 @@ pub fn resolve(
             })?;
         if command.permissions.is_empty() {
             command.permissions = command_permission::resolve_command_permissions(
+                &metadata_accessor.flags,
                 &command.command,
                 command_permissions,
                 object_types,
diff --git a/v3/crates/metadata-resolve/src/stages/data_connectors/mod.rs b/v3/crates/metadata-resolve/src/stages/data_connectors/mod.rs
index f0cf4b46b26..286f9a41da5 100644
--- a/v3/crates/metadata-resolve/src/stages/data_connectors/mod.rs
+++ b/v3/crates/metadata-resolve/src/stages/data_connectors/mod.rs
@@ -29,12 +29,15 @@ pub fn resolve<'a>(
         let qualified_data_connector_name =
             Qualified::new(subgraph.clone(), data_connector.name.clone());
 
-        let (data_connector_context, connector_issues) =
-            types::DataConnectorContext::new(data_connector, &configuration.unstable_features)
-                .map_err(|error| NamedDataConnectorError {
-                    data_connector_name: qualified_data_connector_name.clone(),
-                    error,
-                })?;
+        let (data_connector_context, connector_issues) = types::DataConnectorContext::new(
+            metadata_accessor,
+            data_connector,
+            &configuration.unstable_features,
+        )
+        .map_err(|error| NamedDataConnectorError {
+            data_connector_name: qualified_data_connector_name.clone(),
+            error,
+        })?;
 
         issues.extend(
             connector_issues
diff --git a/v3/crates/metadata-resolve/src/stages/data_connectors/types.rs b/v3/crates/metadata-resolve/src/stages/data_connectors/types.rs
index 4eb973f84af..e0d9a98a4df 100644
--- a/v3/crates/metadata-resolve/src/stages/data_connectors/types.rs
+++ b/v3/crates/metadata-resolve/src/stages/data_connectors/types.rs
@@ -12,6 +12,7 @@ use crate::types::subgraph::Qualified;
 use indexmap::IndexMap;
 use lang_graphql::ast::common::OperationType;
 use ndc_models;
+use open_dds::accessor::MetadataAccessor;
 use open_dds::data_connector::DataConnectorColumnName;
 use open_dds::types::DataConnectorArgumentName;
 use open_dds::{
@@ -63,6 +64,7 @@ pub struct DataConnectorContext<'a> {
 
 impl<'a> DataConnectorContext<'a> {
     pub fn new(
+        metadata_accessor: &MetadataAccessor,
         data_connector: &'a data_connector::DataConnectorLinkV1,
         unstable_features: &UnstableFeatures,
     ) -> Result<(Self, Vec<DataConnectorIssue>), DataConnectorError> {
@@ -102,7 +104,8 @@ impl<'a> DataConnectorContext<'a> {
             .argument_presets
             .iter()
             .map(|argument_preset| -> Result<_, DataConnectorError> {
-                let header_presets = HttpHeadersPreset::new(&argument_preset.value.http_headers)?;
+                let header_presets =
+                    HttpHeadersPreset::new(metadata_accessor, &argument_preset.value.http_headers)?;
                 Ok(ArgumentPreset {
                     name: argument_preset.argument.clone(),
                     value: ArgumentPresetValue {
@@ -376,6 +379,7 @@ pub struct HttpHeadersPreset {
 
 impl HttpHeadersPreset {
     fn new(
+        metadata_accessor: &MetadataAccessor,
         headers_preset: &open_dds::data_connector::HttpHeadersPreset,
     ) -> Result<Self, DataConnectorError> {
         let forward = headers_preset
@@ -389,7 +393,7 @@ impl HttpHeadersPreset {
             .iter()
             .map(|(header_name, header_val)| {
                 let key = SerializableHeaderName::new(header_name.to_string()).map_err(to_error)?;
-                let val = resolve_value_expression(header_val.clone());
+                let val = resolve_value_expression(metadata_accessor, header_val.clone());
                 Ok((key, val))
             })
             .collect::<Result<IndexMap<_, _>, DataConnectorError>>()?;
@@ -402,11 +406,15 @@ impl HttpHeadersPreset {
 }
 
 fn resolve_value_expression(
+    metadata_accessor: &MetadataAccessor,
     value_expression_input: open_dds::permissions::ValueExpression,
 ) -> ValueExpression {
     match value_expression_input {
         open_dds::permissions::ValueExpression::SessionVariable(session_variable) => {
-            ValueExpression::SessionVariable(session_variable)
+            ValueExpression::SessionVariable(hasura_authn_core::SessionVariableReference {
+                name: session_variable,
+                passed_as_json: metadata_accessor.flags.json_session_variables,
+            })
         }
         open_dds::permissions::ValueExpression::Literal(json_value) => {
             ValueExpression::Literal(json_value)
@@ -465,7 +473,7 @@ pub struct DataConnectorCapabilities {
 #[cfg(test)]
 mod tests {
     use ndc_models;
-    use open_dds::data_connector::DataConnectorLinkV1;
+    use open_dds::{accessor::MetadataAccessor, data_connector::DataConnectorLinkV1, Metadata};
     use strum::IntoEnumIterator;
 
     use crate::{
@@ -525,9 +533,13 @@ mod tests {
             enable_ndc_v02_support: true,
             ..Default::default()
         };
-        let (context, issues) =
-            DataConnectorContext::new(&data_connector_with_capabilities, &unstable_features)
-                .unwrap();
+        let metadata_accessor = MetadataAccessor::new(Metadata::WithoutNamespaces(vec![]));
+        let (context, issues) = DataConnectorContext::new(
+            &metadata_accessor,
+            &data_connector_with_capabilities,
+            &unstable_features,
+        )
+        .unwrap();
         assert_eq!(context.capabilities, explicit_capabilities);
         assert_eq!(context.supported_ndc_version, NdcVersion::V01);
         assert_eq!(issues.len(), 0, "Issues: {issues:#?}");
@@ -566,9 +578,13 @@ mod tests {
             enable_ndc_v02_support: true,
             ..Default::default()
         };
-        let (context, issues) =
-            DataConnectorContext::new(&data_connector_with_capabilities, &unstable_features)
-                .unwrap();
+        let metadata_accessor = MetadataAccessor::new(Metadata::WithoutNamespaces(vec![]));
+        let (context, issues) = DataConnectorContext::new(
+            &metadata_accessor,
+            &data_connector_with_capabilities,
+            &unstable_features,
+        )
+        .unwrap();
         assert_eq!(context.capabilities, explicit_capabilities);
         assert_eq!(context.supported_ndc_version, NdcVersion::V02);
         assert_eq!(issues.len(), 0, "Issues: {issues:#?}");
diff --git a/v3/crates/metadata-resolve/src/stages/model_permissions/mod.rs b/v3/crates/metadata-resolve/src/stages/model_permissions/mod.rs
index ca6678fc2fd..ed503d16e48 100644
--- a/v3/crates/metadata-resolve/src/stages/model_permissions/mod.rs
+++ b/v3/crates/metadata-resolve/src/stages/model_permissions/mod.rs
@@ -81,6 +81,7 @@ pub fn resolve(
                 .and_then(|bool_exp| bool_exp.graphql.as_ref());
 
             let select_permissions = model_permission::resolve_model_select_permissions(
+                &metadata_accessor.flags,
                 &model.model,
                 subgraph,
                 permissions,
diff --git a/v3/crates/metadata-resolve/src/stages/model_permissions/model_permission.rs b/v3/crates/metadata-resolve/src/stages/model_permissions/model_permission.rs
index 150d21b3452..da38aef23ea 100644
--- a/v3/crates/metadata-resolve/src/stages/model_permissions/model_permission.rs
+++ b/v3/crates/metadata-resolve/src/stages/model_permissions/model_permission.rs
@@ -33,6 +33,7 @@ use ref_cast::RefCast;
 use std::collections::BTreeMap;
 
 fn resolve_model_predicate_with_model(
+    flags: &open_dds::flags::Flags,
     model_predicate: &open_dds::permissions::ModelPredicate,
     model: &models::Model,
     subgraph: &SubgraphName,
@@ -102,6 +103,7 @@ fn resolve_model_predicate_with_model(
     )?;
 
     resolve_model_predicate_with_type(
+        flags,
         model_predicate,
         &model.data_type,
         object_type_representation,
@@ -136,6 +138,7 @@ pub fn get_model_source_argument<'a>(
 }
 
 pub fn resolve_model_select_permissions(
+    flags: &open_dds::flags::Flags,
     model: &models::Model,
     subgraph: &SubgraphName,
     model_permissions: &ModelPermissionsV1,
@@ -163,6 +166,7 @@ pub fn resolve_model_select_permissions(
             let resolved_predicate = match &select.filter {
                 NullableModelPredicate::NotNull(model_predicate) => {
                     resolve_model_predicate_with_model(
+                        flags,
                         model_predicate,
                         model,
                         subgraph,
@@ -210,6 +214,7 @@ pub fn resolve_model_select_permissions(
                 match model.arguments.get(&argument_preset.argument) {
                     Some(argument) => {
                         let value_expression = resolve_value_expression_for_argument(
+                            flags,
                             &argument_preset.argument,
                             &argument_preset.value,
                             &argument.argument_type,
@@ -269,6 +274,7 @@ pub fn resolve_model_select_permissions(
 /// re-add in future. Because this function takes the `data_connector_field_mappings` as an input,
 /// many of the errors thrown in `resolve_model_predicate` are pushed out.
 pub(crate) fn resolve_model_predicate_with_type(
+    flags: &open_dds::flags::Flags,
     model_predicate: &open_dds::permissions::ModelPredicate,
     type_name: &Qualified<CustomTypeName>,
     object_type_representation: &object_relationships::ObjectTypeWithRelationships,
@@ -385,7 +391,10 @@ pub(crate) fn resolve_model_predicate_with_type(
                     ValueExpression::Literal(json_value.clone())
                 }
                 open_dds::permissions::ValueExpression::SessionVariable(session_variable) => {
-                    ValueExpression::SessionVariable(session_variable.clone())
+                    ValueExpression::SessionVariable(hasura_authn_core::SessionVariableReference {
+                        name: session_variable.clone(),
+                        passed_as_json: flags.json_session_variables,
+                    })
                 }
             };
 
@@ -641,6 +650,7 @@ pub(crate) fn resolve_model_predicate_with_type(
                             }?;
 
                             let target_model_predicate = resolve_model_predicate_with_type(
+                                flags,
                                 nested_predicate,
                                 &target_model.inner.data_type,
                                 target_object_type,
@@ -682,6 +692,7 @@ pub(crate) fn resolve_model_predicate_with_type(
 
         open_dds::permissions::ModelPredicate::Not(predicate) => {
             let resolved_predicate = resolve_model_predicate_with_type(
+                flags,
                 predicate,
                 type_name,
                 object_type_representation,
@@ -702,6 +713,7 @@ pub(crate) fn resolve_model_predicate_with_type(
             let mut resolved_predicates = Vec::new();
             for predicate in predicates {
                 resolved_predicates.push(resolve_model_predicate_with_type(
+                    flags,
                     predicate,
                     type_name,
                     object_type_representation,
@@ -723,6 +735,7 @@ pub(crate) fn resolve_model_predicate_with_type(
             let mut resolved_predicates = Vec::new();
             for predicate in predicates {
                 resolved_predicates.push(resolve_model_predicate_with_type(
+                    flags,
                     predicate,
                     type_name,
                     object_type_representation,
diff --git a/v3/crates/metadata-resolve/src/stages/type_permissions/mod.rs b/v3/crates/metadata-resolve/src/stages/type_permissions/mod.rs
index 580ceff55fe..915328abe1a 100644
--- a/v3/crates/metadata-resolve/src/stages/type_permissions/mod.rs
+++ b/v3/crates/metadata-resolve/src/stages/type_permissions/mod.rs
@@ -12,6 +12,7 @@ use crate::types::subgraph::Qualified;
 
 use crate::helpers::typecheck;
 use crate::stages::object_types;
+use crate::ValueExpressionOrPredicate;
 
 /// resolve type permissions
 pub fn resolve(
@@ -54,6 +55,7 @@ pub fn resolve(
                     output_type_permission,
                 )?;
                 object_type.type_input_permissions = resolve_input_type_permission(
+                    &metadata_accessor.flags,
                     &object_type.object_type,
                     output_type_permission,
                 )?;
@@ -97,6 +99,7 @@ pub fn resolve_output_type_permission(
 }
 
 pub(crate) fn resolve_input_type_permission(
+    flags: &open_dds::flags::Flags,
     object_type_representation: &object_types::ObjectTypeRepresentation,
     type_permissions: &TypePermissionsV1,
 ) -> Result<BTreeMap<Role, TypeInputPermission>, TypeInputPermissionError> {
@@ -133,10 +136,23 @@ pub(crate) fn resolve_input_type_permission(
                         );
                     }
                 };
+                let resolved_value = match &value {
+                    open_dds::permissions::ValueExpression::Literal(literal) => {
+                        ValueExpressionOrPredicate::Literal(literal.clone())
+                    }
+                    open_dds::permissions::ValueExpression::SessionVariable(session_variable) => {
+                        ValueExpressionOrPredicate::SessionVariable(
+                            hasura_authn_core::SessionVariableReference {
+                                name: session_variable.clone(),
+                                passed_as_json: flags.json_session_variables,
+                            },
+                        )
+                    }
+                };
                 resolved_field_presets.insert(
                     field_name.clone(),
                     FieldPresetInfo {
-                        value: value.clone(),
+                        value: resolved_value,
                         deprecated: field_definition.deprecated.clone(),
                     },
                 );
diff --git a/v3/crates/metadata-resolve/src/stages/type_permissions/types.rs b/v3/crates/metadata-resolve/src/stages/type_permissions/types.rs
index a2e8684021a..b36266417b0 100644
--- a/v3/crates/metadata-resolve/src/stages/type_permissions/types.rs
+++ b/v3/crates/metadata-resolve/src/stages/type_permissions/types.rs
@@ -1,12 +1,12 @@
 use std::collections::BTreeMap;
 
 use open_dds::{
-    permissions::{Role, TypeOutputPermission, ValueExpression},
+    permissions::{Role, TypeOutputPermission},
     types::Deprecated,
 };
 
-use crate::stages::object_types;
 use crate::Qualified;
+use crate::{stages::object_types, ValueExpressionOrPredicate};
 use open_dds::types::{CustomTypeName, FieldName};
 use serde::{Deserialize, Serialize};
 use std::ops::Deref;
@@ -43,7 +43,7 @@ pub struct TypeInputPermission {
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub struct FieldPresetInfo {
-    pub value: ValueExpression,
+    pub value: ValueExpressionOrPredicate,
     pub deprecated: Option<Deprecated>,
 }
 
diff --git a/v3/crates/metadata-resolve/src/types/permission.rs b/v3/crates/metadata-resolve/src/types/permission.rs
index 75ecce0eef7..1e725a1bd23 100644
--- a/v3/crates/metadata-resolve/src/types/permission.rs
+++ b/v3/crates/metadata-resolve/src/types/permission.rs
@@ -4,12 +4,12 @@ use serde::{Deserialize, Serialize};
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub enum ValueExpression {
     Literal(serde_json::Value),
-    SessionVariable(open_dds::session_variables::SessionVariable),
+    SessionVariable(open_dds::session_variables::SessionVariableReference),
 }
 
 #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)]
 pub enum ValueExpressionOrPredicate {
     Literal(serde_json::Value),
-    SessionVariable(open_dds::session_variables::SessionVariable),
+    SessionVariable(open_dds::session_variables::SessionVariableReference),
     BooleanExpression(Box<model_permissions::ModelPredicate>),
 }
diff --git a/v3/crates/metadata-resolve/tests/passing/boolean_expression_type/partial_supergraph/resolved.snap b/v3/crates/metadata-resolve/tests/passing/boolean_expression_type/partial_supergraph/resolved.snap
index 91f0edfec74..ceb1d73c254 100644
--- a/v3/crates/metadata-resolve/tests/passing/boolean_expression_type/partial_supergraph/resolved.snap
+++ b/v3/crates/metadata-resolve/tests/passing/boolean_expression_type/partial_supergraph/resolved.snap
@@ -829,9 +829,12 @@ input_file: crates/metadata-resolve/tests/passing/boolean_expression_type/partia
                                     nullable: true,
                                 },
                                 value: SessionVariable(
-                                    SessionVariable(
-                                        "x-hasura-user-id",
-                                    ),
+                                    SessionVariableReference {
+                                        name: SessionVariableName(
+                                            "x-hasura-user-id",
+                                        ),
+                                        passed_as_json: false,
+                                    },
                                 ),
                                 deprecated: None,
                             },
diff --git a/v3/crates/metadata-resolve/tests/passing/missing_subgraph_when_ignoring_unknown_subgraphs/resolved.snap b/v3/crates/metadata-resolve/tests/passing/missing_subgraph_when_ignoring_unknown_subgraphs/resolved.snap
index d54c33e09e5..6a1529524ba 100644
--- a/v3/crates/metadata-resolve/tests/passing/missing_subgraph_when_ignoring_unknown_subgraphs/resolved.snap
+++ b/v3/crates/metadata-resolve/tests/passing/missing_subgraph_when_ignoring_unknown_subgraphs/resolved.snap
@@ -818,9 +818,12 @@ input_file: crates/metadata-resolve/tests/passing/missing_subgraph_when_ignoring
                                     nullable: false,
                                 },
                                 value: SessionVariable(
-                                    SessionVariable(
-                                        "x-hasura-user-id",
-                                    ),
+                                    SessionVariableReference {
+                                        name: SessionVariableName(
+                                            "x-hasura-user-id",
+                                        ),
+                                        passed_as_json: false,
+                                    },
                                 ),
                                 deprecated: None,
                             },
diff --git a/v3/crates/open-dds/metadata.jsonschema b/v3/crates/open-dds/metadata.jsonschema
index 25d866a95f5..a3fd4b98a4c 100644
--- a/v3/crates/open-dds/metadata.jsonschema
+++ b/v3/crates/open-dds/metadata.jsonschema
@@ -3254,6 +3254,10 @@
         "allow_partial_supergraph": {
           "default": false,
           "type": "boolean"
+        },
+        "json_session_variables": {
+          "default": false,
+          "type": "boolean"
         }
       },
       "additionalProperties": false
@@ -3294,7 +3298,8 @@
                 "disallow_scalar_type_names_conflicting_with_inbuilt_types": false,
                 "propagate_boolean_expression_deprecation_status": false,
                 "require_unique_command_graphql_names": false,
-                "allow_partial_supergraph": false
+                "allow_partial_supergraph": false,
+                "json_session_variables": false
               },
               "allOf": [
                 {
@@ -3346,7 +3351,8 @@
                 "disallow_scalar_type_names_conflicting_with_inbuilt_types": false,
                 "propagate_boolean_expression_deprecation_status": false,
                 "require_unique_command_graphql_names": false,
-                "allow_partial_supergraph": false
+                "allow_partial_supergraph": false,
+                "json_session_variables": false
               },
               "allOf": [
                 {
@@ -3388,7 +3394,8 @@
                 "disallow_scalar_type_names_conflicting_with_inbuilt_types": false,
                 "propagate_boolean_expression_deprecation_status": false,
                 "require_unique_command_graphql_names": false,
-                "allow_partial_supergraph": false
+                "allow_partial_supergraph": false,
+                "json_session_variables": false
               },
               "allOf": [
                 {
diff --git a/v3/crates/open-dds/src/flags.rs b/v3/crates/open-dds/src/flags.rs
index 4630dfcca8d..5f49f8fd1f8 100644
--- a/v3/crates/open-dds/src/flags.rs
+++ b/v3/crates/open-dds/src/flags.rs
@@ -32,6 +32,9 @@ pub struct Flags {
 
     #[opendd(default, rename = "allow_partial_supergraph")]
     pub allow_partial_supergraph: bool,
+
+    #[opendd(default, rename = "json_session_variables")]
+    pub json_session_variables: bool,
 }
 
 impl Flags {
@@ -46,6 +49,7 @@ impl Flags {
             propagate_boolean_expression_deprecation_status: false,
             require_unique_command_graphql_names: false,
             allow_partial_supergraph: false,
+            json_session_variables: false,
         }
     }
 
diff --git a/v3/crates/open-dds/src/permissions.rs b/v3/crates/open-dds/src/permissions.rs
index 7f70b4edd61..41cd67db40a 100644
--- a/v3/crates/open-dds/src/permissions.rs
+++ b/v3/crates/open-dds/src/permissions.rs
@@ -8,7 +8,7 @@ use crate::{
     commands::CommandName,
     models::ModelName,
     relationships::RelationshipName,
-    session_variables::SessionVariable,
+    session_variables::SessionVariableName,
     traits,
     types::{CustomTypeName, FieldName, OperatorName},
 };
@@ -649,7 +649,7 @@ pub enum ValueExpression {
     #[schemars(title = "Literal")]
     Literal(JsonValue),
     #[schemars(title = "SessionVariable")]
-    SessionVariable(SessionVariable),
+    SessionVariable(SessionVariableName),
 }
 
 #[derive(
@@ -663,7 +663,7 @@ pub enum ValueExpressionOrPredicate {
     #[schemars(title = "Literal")]
     Literal(JsonValue),
     #[schemars(title = "SessionVariable")]
-    SessionVariable(SessionVariable),
+    SessionVariable(SessionVariableName),
     #[schemars(title = "BooleanExpression")]
     BooleanExpression(Box<ModelPredicate>),
 }
diff --git a/v3/crates/open-dds/src/session_variables.rs b/v3/crates/open-dds/src/session_variables.rs
index eebab4f1487..1c6fe825384 100644
--- a/v3/crates/open-dds/src/session_variables.rs
+++ b/v3/crates/open-dds/src/session_variables.rs
@@ -7,23 +7,33 @@ use std::str::FromStr;
 use schemars::JsonSchema;
 use serde::{Deserialize, Serialize};
 
+/// Used to represent a reference to a session variable,
+/// where one is required in the IR
+#[derive(Debug, Clone, Hash, PartialEq, Eq, JsonSchema, Serialize, Deserialize)]
+#[schemars(rename = "OpenDdSessionVariableReference")]
+pub struct SessionVariableReference {
+    pub name: SessionVariableName,
+    pub passed_as_json: bool,
+}
+
 /// Used to represent the name of a session variable, like
 /// "x-hasura-role".
 #[derive(Debug, Clone, Hash, PartialEq, Eq, JsonSchema, Serialize, Deserialize)]
 #[schemars(rename = "OpenDdSessionVariable")]
-pub struct SessionVariable(Cow<'static, str>);
+pub struct SessionVariableName(Cow<'static, str>);
 
-pub const SESSION_VARIABLE_ROLE: SessionVariable = SessionVariable(Cow::Borrowed("x-hasura-role"));
+pub const SESSION_VARIABLE_ROLE: SessionVariableName =
+    SessionVariableName(Cow::Borrowed("x-hasura-role"));
 
-impl FromStr for SessionVariable {
+impl FromStr for SessionVariableName {
     type Err = Infallible;
 
     fn from_str(s: &str) -> Result<Self, Self::Err> {
-        Ok(SessionVariable(s.trim().to_lowercase().into()))
+        Ok(SessionVariableName(s.trim().to_lowercase().into()))
     }
 }
 
-impl std::fmt::Display for SessionVariable {
+impl std::fmt::Display for SessionVariableName {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.0)
     }
@@ -54,12 +64,12 @@ mod tests {
     fn serialize_and_deserialize_session_variable() {
         let mut session_variables = HashMap::new();
         session_variables.insert(
-            SessionVariable("test-role".into()),
+            SessionVariableName("test-role".into()),
             SessionVariableValue("test-role".into()),
         );
         let json_str = serde_json::to_string(&session_variables).unwrap();
 
-        let parsed_from_string: HashMap<SessionVariable, SessionVariableValue> =
+        let parsed_from_string: HashMap<SessionVariableName, SessionVariableValue> =
             serde_json::from_str(json_str.trim()).unwrap();
 
         assert_eq!(parsed_from_string, session_variables);