Merge remote-tracking branch 'origin/pr/416' into codex/pr-416-420-integration

# Conflicts:
#	crates/aether-data/src/lifecycle/bootstrap/postgres.rs
#	crates/aether-data/src/lifecycle/migrate/tests.rs

apps/aether-gateway/src/control/management_token_permissions.rs

@@ -524,4 +524,83 @@ mod tests {
             }
         }
     }
+
+    #[test]
+    fn catalog_covers_admin_route_signatures_from_route_sources() {
+        let route_sources = [
+            ("route/admin.rs", include_str!("route/admin.rs")),
+            ("route/oauth.rs", include_str!("route/oauth.rs")),
+            (
+                "route/public_support.rs",
+                include_str!("route/public_support.rs"),
+            ),
+            (
+                "route/admin/basic_families.rs",
+                include_str!("route/admin/basic_families.rs"),
+            ),
+            (
+                "route/admin/endpoints_families.rs",
+                include_str!("route/admin/endpoints_families.rs"),
+            ),
+            (
+                "route/admin/model_provider_families.rs",
+                include_str!("route/admin/model_provider_families.rs"),
+            ),
+            (
+                "route/admin/observability_families.rs",
+                include_str!("route/admin/observability_families.rs"),
+            ),
+            (
+                "route/admin/operations_families.rs",
+                include_str!("route/admin/operations_families.rs"),
+            ),
+            (
+                "route/admin/provider_ops_routes.rs",
+                include_str!("route/admin/provider_ops_routes.rs"),
+            ),
+            (
+                "route/admin/system_families.rs",
+                include_str!("route/admin/system_families.rs"),
+            ),
+        ];
+        let mut route_scopes = BTreeSet::new();
+
+        for (file, source) in route_sources {
+            for scope in extract_admin_route_scopes(source) {
+                assert!(
+                    is_known_management_token_permission_scope(scope),
+                    "missing management token permission scope {scope} referenced by {file}"
+                );
+                route_scopes.insert(scope);
+            }
+        }
+
+        assert!(
+            !route_scopes.is_empty(),
+            "admin route scope scanner did not find any route signatures"
+        );
+    }
+
+    fn extract_admin_route_scopes(source: &'static str) -> BTreeSet<&'static str> {
+        let mut scopes = BTreeSet::new();
+        let mut remaining = source;
+
+        while let Some(start) = remaining.find("\"admin:") {
+            let signature_start = start + 1;
+            let after_start = &remaining[signature_start..];
+            let Some(end) = after_start.find('"') else {
+                break;
+            };
+            let signature = &after_start[..end];
+            let mut parts = signature.split(':');
+            if parts.next() == Some("admin") {
+                if let (Some(scope), None) = (parts.next(), parts.next()) {
+                    scopes.insert(scope);
+                }
+            }
+            remaining = &after_start[end + 1..];
+        }
+
+        scopes
+    }
 }

apps/aether-gateway/src/control/route/oauth.rs

@@ -230,7 +230,7 @@ pub(super) fn classify_oauth_route(
             "admin_proxy",
             "provider_oauth_manage",
             "batch_import_oauth",
-            "admin:provider_oauth",
+            "admin:pool",
             false,
         ))
     } else if method == http::Method::POST
@@ -241,7 +241,7 @@ pub(super) fn classify_oauth_route(
             "admin_proxy",
             "provider_oauth_manage",
             "start_batch_import_oauth_task",
-            "admin:provider_oauth",
+            "admin:pool",
             false,
         ))
     } else if method == http::Method::GET
@@ -252,7 +252,7 @@ pub(super) fn classify_oauth_route(
             "admin_proxy",
             "provider_oauth_manage",
             "get_batch_import_task_status",
-            "admin:provider_oauth",
+            "admin:pool",
             false,
         ))
     } else if method == http::Method::POST

apps/aether-gateway/src/control/tests/admin_oauth.rs

@@ -1,5 +1,7 @@
 use http::Uri;
 
+use crate::control::management_token_required_permission;
+
 use super::{classify_control_route, headers};
 
 #[test]
@@ -66,7 +68,11 @@ fn classifies_admin_provider_oauth_batch_import_task_status_as_admin_proxy_route
     );
     assert_eq!(
         decision.auth_endpoint_signature.as_deref(),
-        Some("admin:provider_oauth")
+        Some("admin:pool")
+    );
+    assert_eq!(
+        management_token_required_permission(&http::Method::GET, &decision).as_deref(),
+        Some("admin:pool:read")
     );
     assert!(!decision.is_execution_runtime_candidate());
 }
@@ -74,51 +80,69 @@ fn classifies_admin_provider_oauth_batch_import_task_status_as_admin_proxy_route
 #[test]
 fn classifies_admin_provider_oauth_maintenance_routes_as_admin_proxy_route() {
     let headers = headers(&[]);
-    for (method, path, route_kind) in [
+    for (method, path, route_kind, expected_signature, expected_required_permission) in [
         (
             http::Method::POST,
             "/api/admin/provider-oauth/keys/key-123/complete",
             "complete_key_oauth",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/keys/key-123/refresh",
             "refresh_key_oauth",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/complete",
             "complete_provider_oauth",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/import-refresh-token",
             "import_refresh_token",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/batch-import",
             "batch_import_oauth",
+            "admin:pool",
+            "admin:pool:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/batch-import/tasks",
             "start_batch_import_oauth_task",
+            "admin:pool",
+            "admin:pool:write",
         ),
         (
             http::Method::GET,
             "/api/admin/provider-oauth/providers/provider-123/batch-import/tasks/task-123",
             "get_batch_import_task_status",
+            "admin:pool",
+            "admin:pool:read",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/device-authorize",
             "device_authorize",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
         (
             http::Method::POST,
             "/api/admin/provider-oauth/providers/provider-123/device-poll",
             "device_poll",
+            "admin:provider_oauth",
+            "admin:provider_oauth:write",
         ),
     ] {
         let uri: Uri = path.parse().expect("uri should parse");
@@ -133,7 +157,11 @@ fn classifies_admin_provider_oauth_maintenance_routes_as_admin_proxy_route() {
         assert_eq!(decision.route_kind.as_deref(), Some(route_kind));
         assert_eq!(
             decision.auth_endpoint_signature.as_deref(),
-            Some("admin:provider_oauth")
+            Some(expected_signature)
+        );
+        assert_eq!(
+            management_token_required_permission(&method, &decision).as_deref(),
+            Some(expected_required_permission)
         );
         assert!(!decision.is_execution_runtime_candidate());
     }

apps/aether-gateway/src/tests/control/admin/oauth.rs

@@ -25,9 +25,9 @@ use http::{HeaderMap, HeaderValue, StatusCode};
 use serde_json::json;
 
 use super::super::{
-    build_router_with_state, build_state_with_execution_runtime_override, sample_endpoint,
+    build_router_with_state, build_state_with_execution_runtime_override, hash_management_token,
-    sample_key, sample_management_token, sample_oauth_provider_config, sample_provider,
+    sample_endpoint, sample_key, sample_management_token, sample_oauth_provider_config,
-    sample_proxy_node, start_server, AppState,
+    sample_provider, sample_proxy_node, start_server, AppState,
 };
 use crate::admin_api::{
     maybe_build_local_admin_provider_oauth_response, AdminAppState, AdminRequestContext,
@@ -7201,6 +7201,125 @@ async fn gateway_creates_updates_and_regenerates_admin_management_token_locally_
     drop(upstream_url);
 }
 
+#[tokio::test]
+async fn gateway_allows_management_token_with_pool_write_for_provider_oauth_batch_import() {
+    let raw_token = "ae-provider-oauth-batch-pool-write";
+    let state = AppState::new().expect("gateway should build");
+    let admin_user = state
+        .create_local_auth_user_with_settings(
+            Some("provider-oauth-pool@example.com".to_string()),
+            true,
+            "admin".to_string(),
+            "hash".to_string(),
+            "admin".to_string(),
+            None,
+            None,
+            None,
+            None,
+        )
+        .await
+        .expect("admin user should be created")
+        .expect("admin user should exist");
+    let mut management_token = sample_management_token(
+        "token-provider-oauth-batch-pool",
+        &admin_user.id,
+        "provider-oauth-pool",
+        true,
+    );
+    management_token.token.allowed_ips = None;
+    management_token.token.permissions = Some(json!(["admin:pool:read", "admin:pool:write"]));
+    let management_token_repository =
+        Arc::new(InMemoryManagementTokenRepository::seed_with_hashes(
+            vec![management_token],
+            vec![(
+                hash_management_token(raw_token),
+                "token-provider-oauth-batch-pool".to_string(),
+            )],
+        ));
+
+    let gateway = build_router_with_state(state.with_data_state_for_tests(
+        GatewayDataState::with_management_token_repository_for_tests(management_token_repository),
+    ));
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+
+    let response = reqwest::Client::new()
+        .post(format!(
+            "{gateway_url}/api/admin/provider-oauth/providers/provider-123/batch-import"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .bearer_auth(raw_token)
+        .send()
+        .await
+        .expect("request should succeed");
+
+    let status = response.status();
+    let payload: serde_json::Value = response.json().await.expect("json body should parse");
+    assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{payload}");
+    assert_eq!(payload["detail"], "Admin provider OAuth data unavailable");
+
+    gateway_handle.abort();
+}
+
+#[tokio::test]
+async fn gateway_rejects_management_token_without_pool_write_for_provider_oauth_batch_import() {
+    let raw_token = "ae-provider-oauth-batch-pool-denied";
+    let state = AppState::new().expect("gateway should build");
+    let admin_user = state
+        .create_local_auth_user_with_settings(
+            Some("provider-oauth-pool-denied@example.com".to_string()),
+            true,
+            "admin".to_string(),
+            "hash".to_string(),
+            "admin".to_string(),
+            None,
+            None,
+            None,
+            None,
+        )
+        .await
+        .expect("admin user should be created")
+        .expect("admin user should exist");
+    let mut management_token = sample_management_token(
+        "token-provider-oauth-batch-denied",
+        &admin_user.id,
+        "provider-oauth-denied",
+        true,
+    );
+    management_token.token.allowed_ips = None;
+    management_token.token.permissions = Some(json!(["admin:usage:read"]));
+    let management_token_repository =
+        Arc::new(InMemoryManagementTokenRepository::seed_with_hashes(
+            vec![management_token],
+            vec![(
+                hash_management_token(raw_token),
+                "token-provider-oauth-batch-denied".to_string(),
+            )],
+        ));
+
+    let gateway = build_router_with_state(state.with_data_state_for_tests(
+        GatewayDataState::with_management_token_repository_for_tests(management_token_repository),
+    ));
+    let (gateway_url, gateway_handle) = start_server(gateway).await;
+
+    let response = reqwest::Client::new()
+        .post(format!(
+            "{gateway_url}/api/admin/provider-oauth/providers/provider-123/batch-import"
+        ))
+        .header(crate::constants::GATEWAY_HEADER, "rust-phase3b")
+        .bearer_auth(raw_token)
+        .send()
+        .await
+        .expect("request should succeed");
+
+    assert_eq!(response.status(), StatusCode::FORBIDDEN);
+    let payload: serde_json::Value = response.json().await.expect("json body should parse");
+    assert_eq!(payload["detail"], "management token permission denied");
+    assert_eq!(payload["required_permission"], "admin:pool:write");
+    assert_eq!(payload["route_family"], "provider_oauth_manage");
+
+    gateway_handle.abort();
+}
+
 #[tokio::test]
 async fn gateway_deletes_admin_management_token_locally_with_trusted_admin_principal() {
     let upstream_hits = Arc::new(Mutex::new(0usize));

crates/aether-data/migrations/postgres/20260403000000_baseline.sql

@@ -351,7 +351,8 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     token_prefix character varying(12),
     name character varying(100) NOT NULL,
     description text,
-    allowed_ips json,
+    allowed_ips jsonb,
+    permissions jsonb,
     expires_at timestamp with time zone,
     last_used_at timestamp with time zone,
     last_used_ip character varying(45),
@@ -359,7 +360,7 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     is_active boolean DEFAULT true NOT NULL,
     created_at timestamp with time zone DEFAULT now() NOT NULL,
     updated_at timestamp with time zone DEFAULT now() NOT NULL,
-    CONSTRAINT check_allowed_ips_not_empty CHECK (((allowed_ips IS NULL) OR ((allowed_ips)::text = 'null'::text) OR (json_array_length(allowed_ips) > 0)))
+    CONSTRAINT check_allowed_ips_not_empty CHECK (CASE WHEN ((allowed_ips IS NULL) OR (allowed_ips = 'null'::jsonb)) THEN true WHEN (jsonb_typeof(allowed_ips) = 'array'::text) THEN (jsonb_array_length(allowed_ips) > 0) ELSE false END)
 );
 
 

crates/aether-data/migrations/postgres/20260510000000_normalize_management_token_jsonb.sql

@@ -0,0 +1,18 @@
+ALTER TABLE public.management_tokens
+    DROP CONSTRAINT IF EXISTS check_allowed_ips_not_empty;
+
+ALTER TABLE public.management_tokens
+    ADD COLUMN IF NOT EXISTS permissions jsonb;
+
+ALTER TABLE public.management_tokens
+    ALTER COLUMN allowed_ips TYPE jsonb USING allowed_ips::jsonb,
+    ALTER COLUMN permissions TYPE jsonb USING permissions::jsonb;
+
+ALTER TABLE public.management_tokens
+    ADD CONSTRAINT check_allowed_ips_not_empty CHECK (
+        CASE
+            WHEN allowed_ips IS NULL OR allowed_ips = 'null'::jsonb THEN TRUE
+            WHEN jsonb_typeof(allowed_ips) = 'array' THEN jsonb_array_length(allowed_ips) > 0
+            ELSE FALSE
+        END
+    );

crates/aether-data/schema/bootstrap/postgres/001_types_and_tables.sql

@@ -352,7 +352,8 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     token_prefix character varying(12),
     name character varying(100) NOT NULL,
     description text,
-    allowed_ips json,
+    allowed_ips jsonb,
+    permissions jsonb,
     expires_at timestamp with time zone,
     last_used_at timestamp with time zone,
     last_used_ip character varying(45),
@@ -360,7 +361,7 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     is_active boolean DEFAULT true NOT NULL,
     created_at timestamp with time zone DEFAULT now() NOT NULL,
     updated_at timestamp with time zone DEFAULT now() NOT NULL,
-    CONSTRAINT check_allowed_ips_not_empty CHECK (((allowed_ips IS NULL) OR ((allowed_ips)::text = 'null'::text) OR (json_array_length(allowed_ips) > 0)))
+    CONSTRAINT check_allowed_ips_not_empty CHECK (CASE WHEN ((allowed_ips IS NULL) OR (allowed_ips = 'null'::jsonb)) THEN true WHEN (jsonb_typeof(allowed_ips) = 'array'::text) THEN (jsonb_array_length(allowed_ips) > 0) ELSE false END)
 );
 
 

crates/aether-data/schema/drivers/postgres/baseline/001_types_and_tables.sql

@@ -351,7 +351,8 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     token_prefix character varying(12),
     name character varying(100) NOT NULL,
     description text,
-    allowed_ips json,
+    allowed_ips jsonb,
+    permissions jsonb,
     expires_at timestamp with time zone,
     last_used_at timestamp with time zone,
     last_used_ip character varying(45),
@@ -359,7 +360,7 @@ CREATE TABLE IF NOT EXISTS public.management_tokens (
     is_active boolean DEFAULT true NOT NULL,
     created_at timestamp with time zone DEFAULT now() NOT NULL,
     updated_at timestamp with time zone DEFAULT now() NOT NULL,
-    CONSTRAINT check_allowed_ips_not_empty CHECK (((allowed_ips IS NULL) OR ((allowed_ips)::text = 'null'::text) OR (json_array_length(allowed_ips) > 0)))
+    CONSTRAINT check_allowed_ips_not_empty CHECK (CASE WHEN ((allowed_ips IS NULL) OR (allowed_ips = 'null'::jsonb)) THEN true WHEN (jsonb_typeof(allowed_ips) = 'array'::text) THEN (jsonb_array_length(allowed_ips) > 0) ELSE false END)
 );
 
 

crates/aether-data/src/lifecycle/bootstrap/postgres.rs

@@ -7,7 +7,7 @@ use tracing::info;
 // Generated by build.rs from schema/bootstrap/postgres.
 pub(crate) static EMPTY_DATABASE_SNAPSHOT_SQL: &str =
     include_str!(concat!(env!("OUT_DIR"), "/empty_database_snapshot.sql"));
-pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260509120000;
+pub(crate) const EMPTY_DATABASE_SNAPSHOT_CUTOFF_VERSION: i64 = 20260510000000;
 
 const PUBLIC_BASE_TABLE_COUNT_SQL: &str = r#"
 SELECT COUNT(*)::BIGINT

crates/aether-data/src/lifecycle/migrate/tests.rs

@@ -296,6 +296,7 @@ fn empty_database_snapshot_covers_current_cutoff_versions() {
             20260508000000,
             20260509000000,
             20260509120000,
+            20260510000000,
         ]
     );
 }
@@ -396,6 +397,44 @@ fn provider_api_keys_api_formats_remains_nullable_in_baselines() {
         .contains("pak.allow_auth_channel_mismatch_formats IS NULL"));
 }
 
+#[test]
+fn management_tokens_json_columns_are_normalized_to_jsonb_in_postgres_schema_paths() {
+    let normalization_migration = POSTGRES_MIGRATOR
+        .iter()
+        .find(|migration| migration.version == 20260510000000)
+        .expect("management token jsonb normalization migration should be embedded");
+    assert!(normalization_migration
+        .sql
+        .contains("ALTER COLUMN allowed_ips TYPE jsonb USING allowed_ips::jsonb"));
+    assert!(normalization_migration
+        .sql
+        .contains("ALTER COLUMN permissions TYPE jsonb USING permissions::jsonb"));
+    assert!(normalization_migration
+        .sql
+        .contains("jsonb_array_length(allowed_ips) > 0"));
+
+    assert!(EMPTY_DATABASE_SNAPSHOT_SQL.contains("allowed_ips jsonb,"));
+    assert!(EMPTY_DATABASE_SNAPSHOT_SQL.contains("permissions jsonb,"));
+    assert!(EMPTY_DATABASE_SNAPSHOT_SQL.contains("jsonb_array_length(allowed_ips)"));
+
+    let bootstrap_schema =
+        include_str!("../../../schema/bootstrap/postgres/001_types_and_tables.sql");
+    assert!(bootstrap_schema.contains("allowed_ips jsonb,"));
+    assert!(bootstrap_schema.contains("permissions jsonb,"));
+    assert!(bootstrap_schema.contains("jsonb_array_length(allowed_ips)"));
+
+    let driver_schema =
+        include_str!("../../../schema/drivers/postgres/baseline/001_types_and_tables.sql");
+    assert!(driver_schema.contains("allowed_ips jsonb,"));
+    assert!(driver_schema.contains("permissions jsonb,"));
+    assert!(driver_schema.contains("jsonb_array_length(allowed_ips)"));
+
+    let generated_identity =
+        include_str!("../../../schema/generated/postgres/baseline/001_identity.sql");
+    assert!(generated_identity.contains("allowed_ips jsonb,"));
+    assert!(generated_identity.contains("permissions jsonb,"));
+}
+
 #[test]
 fn provider_api_keys_api_key_is_nullable() {
     let baseline_migration = POSTGRES_MIGRATOR
@@ -1038,6 +1077,7 @@ fn pending_migrations_from_applied_skips_versions_already_applied() {
             20260508000000,
             20260509000000,
             20260509120000,
+            20260510000000,
         ]
     );
 }

crates/aether-data/src/repository/management_tokens/postgres.rs

@@ -103,7 +103,15 @@ DELETE FROM management_tokens
 WHERE id = $1
 "#;
 
-const CREATE_MANAGEMENT_TOKEN_SQL: &str = r#"
+const MANAGEMENT_TOKEN_JSON_COLUMN_TYPES_SQL: &str = r#"
+SELECT column_name, udt_name
+FROM information_schema.columns
+WHERE table_schema = 'public'
+  AND table_name = 'management_tokens'
+  AND column_name IN ('allowed_ips', 'permissions')
+"#;
+
+const CREATE_MANAGEMENT_TOKEN_SQL_PREFIX: &str = r#"
 INSERT INTO management_tokens (
   id,
   user_id,
@@ -123,8 +131,9 @@ VALUES (
   $4,
   $5,
   $6,
-  $7,
+"#;
-  $8,
+
+const CREATE_MANAGEMENT_TOKEN_SQL_SUFFIX: &str = r#",
   CASE
     WHEN $9::bigint IS NULL THEN NULL
     ELSE to_timestamp($9::double precision)
@@ -148,26 +157,23 @@ RETURNING
   EXTRACT(EPOCH FROM updated_at)::bigint AS updated_at_unix_secs
 "#;
 
-const UPDATE_MANAGEMENT_TOKEN_SQL: &str = r#"
+const UPDATE_MANAGEMENT_TOKEN_SQL_PREFIX: &str = r#"
 UPDATE management_tokens
-SET name = COALESCE($2, name),
+SET name = $2,
-    description = CASE
+    description = $3,
-      WHEN $3 THEN NULL
+    allowed_ips =
-      WHEN $4::text IS NULL THEN description
+"#;
-      ELSE $4
+
-    END,
+const UPDATE_MANAGEMENT_TOKEN_SQL_MIDDLE: &str = r#",
-    allowed_ips = CASE
+    permissions =
-      WHEN $5 THEN NULL
+"#;
-      WHEN $6::json IS NULL THEN allowed_ips
+
-      ELSE $6
+const UPDATE_MANAGEMENT_TOKEN_SQL_SUFFIX: &str = r#",
-    END,
-    permissions = COALESCE($7::json, permissions),
     expires_at = CASE
-      WHEN $8 THEN NULL
+      WHEN $6::bigint IS NULL THEN NULL
-      WHEN $9::bigint IS NULL THEN expires_at
+      ELSE to_timestamp($6::double precision)
-      ELSE to_timestamp($9::double precision)
     END,
-    is_active = COALESCE($10, is_active),
+    is_active = $7,
     updated_at = NOW()
 WHERE id = $1
 RETURNING
@@ -261,10 +267,93 @@ pub struct SqlxManagementTokenRepository {
     pool: PgPool,
 }
 
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum JsonColumnType {
+    Json,
+    Jsonb,
+}
+
+impl JsonColumnType {
+    fn from_udt_name(value: &str) -> Option<Self> {
+        match value {
+            "json" => Some(Self::Json),
+            "jsonb" => Some(Self::Jsonb),
+            _ => None,
+        }
+    }
+
+    fn sql_type(self) -> &'static str {
+        match self {
+            Self::Json => "json",
+            Self::Jsonb => "jsonb",
+        }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+struct ManagementTokenJsonColumnTypes {
+    allowed_ips: JsonColumnType,
+    permissions: JsonColumnType,
+}
+
 impl SqlxManagementTokenRepository {
     pub fn new(pool: PgPool) -> Self {
         Self { pool }
     }
+
+    async fn json_column_types(&self) -> Result<ManagementTokenJsonColumnTypes, DataLayerError> {
+        let rows = sqlx::query(MANAGEMENT_TOKEN_JSON_COLUMN_TYPES_SQL)
+            .fetch_all(&self.pool)
+            .await
+            .map_postgres_err()?;
+        let mut allowed_ips = None;
+        let mut permissions = None;
+        for row in rows {
+            let column_name: String = row.try_get("column_name").map_postgres_err()?;
+            let udt_name: String = row.try_get("udt_name").map_postgres_err()?;
+            let Some(column_type) = JsonColumnType::from_udt_name(udt_name.as_str()) else {
+                return Err(DataLayerError::UnexpectedValue(format!(
+                    "unsupported management_tokens.{column_name} column type: {udt_name}"
+                )));
+            };
+            match column_name.as_str() {
+                "allowed_ips" => allowed_ips = Some(column_type),
+                "permissions" => permissions = Some(column_type),
+                _ => {}
+            }
+        }
+
+        match (allowed_ips, permissions) {
+            (Some(allowed_ips), Some(permissions)) => Ok(ManagementTokenJsonColumnTypes {
+                allowed_ips,
+                permissions,
+            }),
+            _ => Err(DataLayerError::UnexpectedValue(
+                "management_tokens JSON column metadata missing".to_string(),
+            )),
+        }
+    }
+}
+
+fn create_management_token_sql(types: ManagementTokenJsonColumnTypes) -> String {
+    format!(
+        "{}  $7::text::{},\n  $8::text::{}{}",
+        CREATE_MANAGEMENT_TOKEN_SQL_PREFIX,
+        types.allowed_ips.sql_type(),
+        types.permissions.sql_type(),
+        CREATE_MANAGEMENT_TOKEN_SQL_SUFFIX
+    )
+}
+
+fn update_management_token_sql(types: ManagementTokenJsonColumnTypes) -> String {
+    format!(
+        "{} $4::text::{}{} $5::text::{}{}",
+        UPDATE_MANAGEMENT_TOKEN_SQL_PREFIX,
+        types.allowed_ips.sql_type(),
+        UPDATE_MANAGEMENT_TOKEN_SQL_MIDDLE,
+        types.permissions.sql_type(),
+        UPDATE_MANAGEMENT_TOKEN_SQL_SUFFIX
+    )
 }
 
 #[async_trait]
@@ -330,15 +419,19 @@ impl ManagementTokenWriteRepository for SqlxManagementTokenRepository {
         record: &CreateManagementTokenRecord,
     ) -> Result<StoredManagementToken, DataLayerError> {
         record.validate()?;
-        let row = sqlx::query(CREATE_MANAGEMENT_TOKEN_SQL)
+        let json_column_types = self.json_column_types().await?;
+        let sql = create_management_token_sql(json_column_types);
+        let allowed_ips = json_to_string(record.allowed_ips.as_ref())?;
+        let permissions = json_to_string(record.permissions.as_ref())?;
+        let row = sqlx::query(sql.as_str())
             .bind(&record.id)
             .bind(&record.user_id)
             .bind(&record.token_hash)
             .bind(record.token_prefix.as_deref())
             .bind(&record.name)
             .bind(record.description.as_deref())
-            .bind(record.allowed_ips.as_ref())
+            .bind(allowed_ips)
-            .bind(record.permissions.as_ref())
+            .bind(permissions)
             .bind(
                 record
                     .expires_at_unix_secs
@@ -356,21 +449,56 @@ impl ManagementTokenWriteRepository for SqlxManagementTokenRepository {
         record: &UpdateManagementTokenRecord,
     ) -> Result<Option<StoredManagementToken>, DataLayerError> {
         record.validate()?;
-        let row = sqlx::query(UPDATE_MANAGEMENT_TOKEN_SQL)
+        let Some(current) = self
+            .get_management_token_with_user(&record.token_id)
+            .await?
+        else {
+            return Ok(None);
+        };
+        let json_column_types = self.json_column_types().await?;
+        let sql = update_management_token_sql(json_column_types);
+        let name = record
+            .name
+            .as_deref()
+            .unwrap_or(current.token.name.as_str());
+        let description = if record.clear_description {
+            None
+        } else {
+            record
+                .description
+                .as_deref()
+                .or(current.token.description.as_deref())
+        };
+        let allowed_ips = if record.clear_allowed_ips {
+            None
+        } else {
+            record
+                .allowed_ips
+                .as_ref()
+                .or(current.token.allowed_ips.as_ref())
+        };
+        let permissions = record
+            .permissions
+            .as_ref()
+            .or(current.token.permissions.as_ref());
+        let expires_at_unix_secs = if record.clear_expires_at {
+            None
+        } else {
+            record
+                .expires_at_unix_secs
+                .or(current.token.expires_at_unix_secs)
+        };
+        let is_active = record.is_active.unwrap_or(current.token.is_active);
+        let allowed_ips = json_to_string(allowed_ips)?;
+        let permissions = json_to_string(permissions)?;
+        let row = sqlx::query(sql.as_str())
             .bind(&record.token_id)
-            .bind(record.name.as_deref())
+            .bind(name)
-            .bind(record.clear_description)
+            .bind(description)
-            .bind(record.description.as_deref())
+            .bind(allowed_ips)
-            .bind(record.clear_allowed_ips)
+            .bind(permissions)
-            .bind(record.allowed_ips.as_ref())
+            .bind(expires_at_unix_secs.and_then(|value| i64::try_from(value).ok()))
-            .bind(record.permissions.as_ref())
+            .bind(is_active)
-            .bind(record.clear_expires_at)
-            .bind(
-                record
-                    .expires_at_unix_secs
-                    .and_then(|value| i64::try_from(value).ok()),
-            )
-            .bind(record.is_active)
             .fetch_optional(&self.pool)
             .await
             .map_err(|err| map_management_token_write_error(err, record.name.as_deref()))?;
@@ -434,6 +562,18 @@ fn optional_unix_secs(value: Option<i64>) -> Option<u64> {
     value.and_then(|value| u64::try_from(value).ok())
 }
 
+fn json_to_string(value: Option<&serde_json::Value>) -> Result<Option<String>, DataLayerError> {
+    value
+        .map(|value| {
+            serde_json::to_string(value).map_err(|err| {
+                DataLayerError::UnexpectedValue(format!(
+                    "invalid management token JSON field: {err}"
+                ))
+            })
+        })
+        .transpose()
+}
+
 fn map_management_token_write_error(
     err: sqlx::Error,
     requested_name: Option<&str>,
@@ -503,7 +643,10 @@ fn map_token_with_user_row(row: &PgRow) -> Result<StoredManagementTokenWithUser,
 
 #[cfg(test)]
 mod tests {
-    use super::SqlxManagementTokenRepository;
+    use super::{
+        create_management_token_sql, update_management_token_sql, JsonColumnType,
+        ManagementTokenJsonColumnTypes, SqlxManagementTokenRepository,
+    };
     use crate::driver::postgres::{PostgresPoolConfig, PostgresPoolFactory};
 
     #[tokio::test]
@@ -523,4 +666,42 @@ mod tests {
         let pool = factory.connect_lazy().expect("pool should build");
         let _repository = SqlxManagementTokenRepository::new(pool);
     }
+
+    #[test]
+    fn repository_sql_casts_json_fields_to_detected_column_types_without_json_case() {
+        let jsonb_types = ManagementTokenJsonColumnTypes {
+            allowed_ips: JsonColumnType::Jsonb,
+            permissions: JsonColumnType::Jsonb,
+        };
+        let json_types = ManagementTokenJsonColumnTypes {
+            allowed_ips: JsonColumnType::Json,
+            permissions: JsonColumnType::Json,
+        };
+        let jsonb_create_sql = create_management_token_sql(jsonb_types);
+        let jsonb_update_sql = update_management_token_sql(jsonb_types);
+        let json_create_sql = create_management_token_sql(json_types);
+        let json_update_sql = update_management_token_sql(json_types);
+
+        assert!(jsonb_create_sql.contains("$7::text::jsonb"));
+        assert!(jsonb_create_sql.contains("$8::text::jsonb"));
+        assert!(jsonb_update_sql.contains("allowed_ips =\n $4::text::jsonb"));
+        assert!(jsonb_update_sql.contains("permissions =\n $5::text::jsonb"));
+
+        assert!(json_create_sql.contains("$7::text::json"));
+        assert!(json_create_sql.contains("$8::text::json"));
+        assert!(json_update_sql.contains("allowed_ips =\n $4::text::json"));
+        assert!(json_update_sql.contains("permissions =\n $5::text::json"));
+
+        for sql in [
+            jsonb_create_sql.as_str(),
+            jsonb_update_sql.as_str(),
+            json_create_sql.as_str(),
+            json_update_sql.as_str(),
+        ] {
+            assert!(!sql.contains("allowed_ips = CASE"));
+            assert!(!sql.contains("permissions = CASE"));
+            assert!(!sql.contains("$6::json IS NULL"));
+            assert!(!sql.contains("COALESCE($7::json"));
+        }
+    }
 }