diff --git a/src/iceberg/catalog/rest/auth/auth_manager.cc b/src/iceberg/catalog/rest/auth/auth_manager.cc
index 10290489a..5facfa3eb 100644
--- a/src/iceberg/catalog/rest/auth/auth_manager.cc
+++ b/src/iceberg/catalog/rest/auth/auth_manager.cc
@@ -19,22 +19,67 @@
 
 #include "iceberg/catalog/rest/auth/auth_manager.h"
 
+#include <array>
+#include <chrono>
 #include <optional>
+#include <string_view>
+#include <utility>
 
 #include "iceberg/catalog/rest/auth/auth_manager_internal.h"
 #include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/auth_session_internal.h"
 #include "iceberg/catalog/rest/auth/oauth2_util.h"
+#include "iceberg/catalog/session_context.h"
 #include "iceberg/util/base64.h"
 #include "iceberg/util/macros.h"
 
 namespace iceberg::rest::auth {
 
+namespace {
+
+constexpr std::string_view kAuthorizationHeader = "Authorization";
+
+const std::array<std::string_view, 5> kTokenPreferenceOrder = {
+    AuthProperties::kIdTokenType,    AuthProperties::kAccessTokenType,
+    AuthProperties::kJwtTokenType,   AuthProperties::kSaml2TokenType,
+    AuthProperties::kSaml1TokenType,
+};
+
+std::optional<std::pair<std::string, std::string>> FindPreferredTypedToken(
+    const std::unordered_map<std::string, std::string>& credentials) {
+  for (std::string_view token_type : kTokenPreferenceOrder) {
+    auto token_it = credentials.find(std::string(token_type));
+    if (token_it != credentials.end()) {
+      return std::pair{token_it->first, token_it->second};
+    }
+  }
+  return std::nullopt;
+}
+
+std::unordered_map<std::string, std::string> FilterTableSessionProperties(
+    const std::unordered_map<std::string, std::string>& properties) {
+  std::unordered_map<std::string, std::string> filtered;
+  if (auto token_it = properties.find(AuthProperties::kToken.key());
+      token_it != properties.end()) {
+    filtered.emplace(token_it->first, token_it->second);
+  }
+  for (std::string_view token_type : kTokenPreferenceOrder) {
+    auto token_it = properties.find(std::string(token_type));
+    if (token_it != properties.end()) {
+      filtered.emplace(token_it->first, token_it->second);
+    }
+  }
+  return filtered;
+}
+
+}  // namespace
+
 Result<std::shared_ptr<AuthSession>> AuthManager::InitSession(
-    HttpClient& init_client,
+    std::shared_ptr<HttpClient> init_client,
     const std::unordered_map<std::string, std::string>& properties) {
   // By default, use the catalog session for initialization
-  return CatalogSession(init_client, properties);
+  return CatalogSession(std::move(init_client), properties);
 }
 
 Result<std::shared_ptr<AuthSession>> AuthManager::ContextualSession(
@@ -55,7 +100,7 @@ Result<std::shared_ptr<AuthSession>> AuthManager::TableSession(
 class NoopAuthManager : public AuthManager {
  public:
   Result<std::shared_ptr<AuthSession>> CatalogSession(
-      [[maybe_unused]] HttpClient& client,
+      [[maybe_unused]] std::shared_ptr<HttpClient> client,
       [[maybe_unused]] const std::unordered_map<std::string, std::string>& properties)
       override {
     return AuthSession::MakeDefault({});
@@ -72,7 +117,7 @@ Result<std::unique_ptr<AuthManager>> MakeNoopAuthManager(
 class BasicAuthManager : public AuthManager {
  public:
   Result<std::shared_ptr<AuthSession>> CatalogSession(
-      [[maybe_unused]] HttpClient& client,
+      [[maybe_unused]] std::shared_ptr<HttpClient> client,
       const std::unordered_map<std::string, std::string>& properties) override {
     auto username_it = properties.find(AuthProperties::kBasicUsername);
     ICEBERG_PRECHECK(username_it != properties.end() && !username_it->second.empty(),
@@ -96,67 +141,184 @@ Result<std::unique_ptr<AuthManager>> MakeBasicAuthManager(
 class OAuth2Manager : public AuthManager {
  public:
   Result<std::shared_ptr<AuthSession>> InitSession(
-      HttpClient& init_client,
+      std::shared_ptr<HttpClient> init_client,
       const std::unordered_map<std::string, std::string>& properties) override {
+    ICEBERG_PRECHECK(init_client != nullptr,
+                     "OAuth2 initialization HTTP client must not be null");
     ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties));
     // No token refresh during init (short-lived session).
     config.Set(AuthProperties::kKeepRefreshed, false);
 
     // Credential takes priority: fetch a fresh token for the config request.
     if (!config.credential().empty()) {
-      auto init_session = AuthSession::MakeDefault(AuthHeaders(config.token()));
-      ICEBERG_ASSIGN_OR_RAISE(init_token_response_,
-                              FetchToken(init_client, *init_session, config));
-      return AuthSession::MakeDefault(AuthHeaders(init_token_response_->access_token));
+      auto init_session =
+          AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token()));
+      start_time_ = std::chrono::steady_clock::now();
+      ICEBERG_ASSIGN_OR_RAISE(
+          auth_response_, OAuth2Util::FetchToken(*init_client, *init_session, config));
+      // TODO(lishuxu): Match Java OAuth2Util.AuthSession.fromTokenResponse here.
+      return AuthSession::MakeDefault(
+          OAuth2Util::AuthHeaders(auth_response_->access_token));
     }
 
     if (!config.token().empty()) {
-      return AuthSession::MakeDefault(AuthHeaders(config.token()));
+      // TODO(lishuxu): Match Java OAuth2Util.AuthSession.fromAccessToken here.
+      return AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token()));
     }
 
     return AuthSession::MakeDefault({});
   }
 
   Result<std::shared_ptr<AuthSession>> CatalogSession(
-      HttpClient& client,
+      std::shared_ptr<HttpClient> shared_client,
       const std::unordered_map<std::string, std::string>& properties) override {
     ICEBERG_ASSIGN_OR_RAISE(auto config, AuthProperties::FromProperties(properties));
-
-    // Reuse token from init phase.
-    if (init_token_response_.has_value()) {
-      auto token_response = std::move(*init_token_response_);
-      init_token_response_.reset();
-      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
-                                     config.client_id(), config.client_secret(),
-                                     config.scope(), config.keep_refreshed(),
-                                     config.optional_oauth_params(), client);
+    ICEBERG_PRECHECK(shared_client != nullptr,
+                     "OAuth2 catalog session HTTP client must not be null");
+    refresh_client_ = std::move(shared_client);
+    // Reuse the token response and start time from the init phase.
+    if (auth_response_.has_value()) {
+      return internal::MakeOAuth2Session(
+          *auth_response_, config.oauth2_server_uri(), config.client_id(),
+          config.client_secret(), config.scope(), config.keep_refreshed(),
+          config.optional_oauth_params(), refresh_client_, start_time_);
     }
 
-    // If token is provided, use it directly.
+    // TODO(lishuxu): Honor token-refresh-enabled for catalog bearer tokens, matching
+    // Java. If token is provided, use it directly.
     if (!config.token().empty()) {
-      return AuthSession::MakeDefault(AuthHeaders(config.token()));
+      OAuthTokenResponse token_response{
+          .access_token = config.token(),
+          .token_type = "bearer",
+          .issued_token_type = AuthProperties::kAccessTokenType,
+      };
+      return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
+                                     config.client_id(), config.client_secret(),
+                                     config.scope(), /*keep_refreshed=*/false,
+                                     config.optional_oauth_params(), refresh_client_);
     }
 
     // Fetch a new token using client_credentials grant.
     if (!config.credential().empty()) {
-      auto base_session = AuthSession::MakeDefault(AuthHeaders(config.token()));
+      auto base_session =
+          AuthSession::MakeDefault(OAuth2Util::AuthHeaders(config.token()));
       OAuthTokenResponse token_response;
-      ICEBERG_ASSIGN_OR_RAISE(token_response, FetchToken(client, *base_session, config));
+      ICEBERG_ASSIGN_OR_RAISE(
+          token_response,
+          OAuth2Util::FetchToken(*refresh_client_, *base_session, config));
       return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
                                      config.client_id(), config.client_secret(),
                                      config.scope(), config.keep_refreshed(),
-                                     config.optional_oauth_params(), client);
+                                     config.optional_oauth_params(), refresh_client_);
     }
 
-    return AuthSession::MakeDefault({});
+    return MakeSession(AccessTokenResponse(""), config, /*keep_refreshed=*/false);
   }
 
-  // TODO(lishuxu): Override TableSession() for token exchange (RFC 8693).
-  // TODO(lishuxu): Override ContextualSession() for per-context exchange.
+  Result<std::shared_ptr<AuthSession>> ContextualSession(
+      const SessionContext& context, std::shared_ptr<AuthSession> parent) override {
+    // TODO(lishuxu): Add child-session caching and refresh, matching Java
+    // AuthSessionCache.
+    return MaybeCreateChildSession(context.credentials, /*allow_credential=*/true,
+                                   std::move(parent));
+  }
+
+  Result<std::shared_ptr<AuthSession>> TableSession(
+      [[maybe_unused]] const TableIdentifier& table,
+      const std::unordered_map<std::string, std::string>& properties,
+      std::shared_ptr<AuthSession> parent) override {
+    return MaybeCreateChildSession(FilterTableSessionProperties(properties),
+                                   /*allow_credential=*/false, std::move(parent));
+  }
+
+  Status Close() override {
+    refresh_client_.reset();
+    return {};
+  }
 
  private:
-  /// Cached token from InitSession
-  std::optional<OAuthTokenResponse> init_token_response_;
+  static OAuthTokenResponse AccessTokenResponse(std::string token) {
+    return {
+        .access_token = std::move(token),
+        .token_type = "bearer",
+        .issued_token_type = AuthProperties::kAccessTokenType,
+    };
+  }
+
+  static Result<AuthProperties> ChildConfig(const OAuth2SessionInfo& parent_info,
+                                            const std::string& credential) {
+    auto properties = parent_info.optional_oauth_params;
+    properties[AuthProperties::kCredential.key()] = credential;
+    properties[AuthProperties::kScope.key()] = parent_info.scope;
+    properties[AuthProperties::kOAuth2ServerUri.key()] = parent_info.oauth2_server_uri;
+    return AuthProperties::FromProperties(properties);
+  }
+
+  Result<std::shared_ptr<AuthSession>> MakeSession(
+      const OAuthTokenResponse& token_response, const AuthProperties& config,
+      bool keep_refreshed) const {
+    ICEBERG_PRECHECK(refresh_client_ != nullptr,
+                     "OAuth2 catalog session must be initialized before child sessions");
+    return AuthSession::MakeOAuth2(token_response, config.oauth2_server_uri(),
+                                   config.client_id(), config.client_secret(),
+                                   config.scope(), keep_refreshed,
+                                   config.optional_oauth_params(), refresh_client_);
+  }
+
+  Result<std::shared_ptr<AuthSession>> MaybeCreateChildSession(
+      const std::unordered_map<std::string, std::string>& credentials,
+      bool allow_credential, std::shared_ptr<AuthSession> parent) {
+    auto token_it = credentials.find(AuthProperties::kToken.key());
+    auto credential_it = credentials.find(AuthProperties::kCredential.key());
+    auto typed_token = FindPreferredTypedToken(credentials);
+    if (token_it == credentials.end() &&
+        (!allow_credential || credential_it == credentials.end()) &&
+        !typed_token.has_value()) {
+      return parent;
+    }
+
+    ICEBERG_PRECHECK(refresh_client_ != nullptr,
+                     "OAuth2 catalog session must be initialized before child sessions");
+    auto parent_info = parent->OAuth2Info();
+    ICEBERG_PRECHECK(parent_info.has_value(),
+                     "OAuth2 child session requires OAuth2 parent metadata");
+
+    if (token_it != credentials.end()) {
+      ICEBERG_ASSIGN_OR_RAISE(auto config,
+                              ChildConfig(*parent_info, parent_info->credential));
+      return MakeSession(AccessTokenResponse(token_it->second), config,
+                         /*keep_refreshed=*/false);
+    }
+
+    if (allow_credential && credential_it != credentials.end()) {
+      ICEBERG_ASSIGN_OR_RAISE(auto config,
+                              ChildConfig(*parent_info, credential_it->second));
+      ICEBERG_ASSIGN_OR_RAISE(auto response,
+                              OAuth2Util::FetchToken(*refresh_client_, *parent, config));
+      return MakeSession(response, config, /*keep_refreshed=*/false);
+    }
+
+    std::optional<std::string> actor_token;
+    std::optional<std::string> actor_token_type;
+    if (!parent_info->token.empty()) {
+      actor_token = parent_info->token;
+      actor_token_type = parent_info->issued_token_type;
+    }
+    ICEBERG_ASSIGN_OR_RAISE(
+        auto response,
+        OAuth2Util::ExchangeToken(*refresh_client_, *parent, {}, typed_token->second,
+                                  typed_token->first, actor_token, actor_token_type,
+                                  parent_info->scope, parent_info->oauth2_server_uri,
+                                  parent_info->optional_oauth_params));
+    ICEBERG_ASSIGN_OR_RAISE(auto config,
+                            ChildConfig(*parent_info, parent_info->credential));
+    return MakeSession(response, config, /*keep_refreshed=*/false);
+  }
+
+  /// Token response and start time captured by InitSession.
+  std::optional<OAuthTokenResponse> auth_response_;
+  std::optional<std::chrono::steady_clock::time_point> start_time_;
+  std::shared_ptr<HttpClient> refresh_client_;
 };
 
 Result<std::unique_ptr<AuthManager>> MakeOAuth2Manager(
diff --git a/src/iceberg/catalog/rest/auth/auth_manager.h b/src/iceberg/catalog/rest/auth/auth_manager.h
index 0a97c9b2a..6ba25c284 100644
--- a/src/iceberg/catalog/rest/auth/auth_manager.h
+++ b/src/iceberg/catalog/rest/auth/auth_manager.h
@@ -48,7 +48,7 @@ class ICEBERG_REST_EXPORT AuthManager {
   /// \param properties Client configuration supplied by the catalog.
   /// \return Session for initialization or an error if credentials cannot be acquired.
   virtual Result<std::shared_ptr<AuthSession>> InitSession(
-      HttpClient& init_client,
+      std::shared_ptr<HttpClient> init_client,
       const std::unordered_map<std::string, std::string>& properties);
 
   /// \brief Create the long-lived catalog session that acts as the parent session.
@@ -62,7 +62,7 @@ class ICEBERG_REST_EXPORT AuthManager {
   /// \return Session for catalog operations or an error if authentication cannot be set
   /// up.
   virtual Result<std::shared_ptr<AuthSession>> CatalogSession(
-      HttpClient& shared_client,
+      std::shared_ptr<HttpClient> shared_client,
       const std::unordered_map<std::string, std::string>& properties) = 0;
 
   /// \brief Create or reuse a session for a specific context.
diff --git a/src/iceberg/catalog/rest/auth/auth_properties.cc b/src/iceberg/catalog/rest/auth/auth_properties.cc
index dcf16782c..f373df617 100644
--- a/src/iceberg/catalog/rest/auth/auth_properties.cc
+++ b/src/iceberg/catalog/rest/auth/auth_properties.cc
@@ -22,6 +22,7 @@
 #include <utility>
 
 #include "iceberg/catalog/rest/catalog_properties.h"
+#include "iceberg/catalog/rest/rest_util.h"
 
 namespace iceberg::rest::auth {
 
@@ -35,6 +36,31 @@ std::pair<std::string, std::string> ParseCredential(const std::string& credentia
   return {credential.substr(0, colon_pos), credential.substr(colon_pos + 1)};
 }
 
+Result<std::string> ResolveOAuth2ServerUri(
+    const std::unordered_map<std::string, std::string>& properties) {
+  auto endpoint_it = properties.find(AuthProperties::kOAuth2ServerUri.key());
+  std::string endpoint = endpoint_it == properties.end()
+                             ? AuthProperties::kOAuth2ServerUri.value()
+                             : endpoint_it->second;
+
+  if (endpoint.starts_with("http://") || endpoint.starts_with("https://")) {
+    return endpoint;
+  }
+  if (endpoint.empty()) {
+    return endpoint;
+  }
+  auto uri_it = properties.find(RestCatalogProperties::kUri.key());
+  if (uri_it == properties.end() || uri_it->second.empty()) {
+    return endpoint;
+  }
+
+  auto base_uri = std::string(TrimTrailingSlash(uri_it->second));
+  if (endpoint.starts_with('/')) {
+    return base_uri + endpoint;
+  }
+  return base_uri + "/" + endpoint;
+}
+
 }  // namespace
 
 std::unordered_map<std::string, std::string> AuthProperties::optional_oauth_params()
@@ -61,19 +87,8 @@ Result<AuthProperties> AuthProperties::FromProperties(
     config.client_secret_ = std::move(secret);
   }
 
-  // Resolve token endpoint: if not explicitly set, derive from catalog URI
-  if (properties.find(kOAuth2ServerUri.key()) == properties.end() ||
-      properties.at(kOAuth2ServerUri.key()).empty()) {
-    auto uri_it = properties.find(RestCatalogProperties::kUri.key());
-    if (uri_it != properties.end() && !uri_it->second.empty()) {
-      std::string_view base = uri_it->second;
-      while (!base.empty() && base.back() == '/') {
-        base.remove_suffix(1);
-      }
-      config.Set(kOAuth2ServerUri,
-                 std::string(base) + "/" + std::string(kOAuth2ServerUri.value()));
-    }
-  }
+  ICEBERG_ASSIGN_OR_RAISE(auto oauth2_server_uri, ResolveOAuth2ServerUri(properties));
+  config.Set(kOAuth2ServerUri, std::move(oauth2_server_uri));
 
   // TODO(lishuxu): Parse JWT exp claim from token to set expires_at_millis_.
 
diff --git a/src/iceberg/catalog/rest/auth/auth_properties.h b/src/iceberg/catalog/rest/auth/auth_properties.h
index a699569c1..8784194cc 100644
--- a/src/iceberg/catalog/rest/auth/auth_properties.h
+++ b/src/iceberg/catalog/rest/auth/auth_properties.h
@@ -82,6 +82,20 @@ class ICEBERG_REST_EXPORT AuthProperties : public ConfigBase<AuthProperties> {
   inline static Entry<std::string> kAudience{"audience", ""};
   inline static Entry<std::string> kResource{"resource", ""};
 
+  // ---- OAuth2 token type constants ----
+
+  inline static const std::string kAccessTokenType =
+      "urn:ietf:params:oauth:token-type:access_token";
+  inline static const std::string kRefreshTokenType =
+      "urn:ietf:params:oauth:token-type:refresh_token";
+  inline static const std::string kIdTokenType =
+      "urn:ietf:params:oauth:token-type:id_token";
+  inline static const std::string kSaml1TokenType =
+      "urn:ietf:params:oauth:token-type:saml1";
+  inline static const std::string kSaml2TokenType =
+      "urn:ietf:params:oauth:token-type:saml2";
+  inline static const std::string kJwtTokenType = "urn:ietf:params:oauth:token-type:jwt";
+
   /// \brief Build an AuthProperties from a properties map.
   static Result<AuthProperties> FromProperties(
       const std::unordered_map<std::string, std::string>& properties);
diff --git a/src/iceberg/catalog/rest/auth/auth_session.cc b/src/iceberg/catalog/rest/auth/auth_session.cc
index 545ee00b1..60aa0a3b4 100644
--- a/src/iceberg/catalog/rest/auth/auth_session.cc
+++ b/src/iceberg/catalog/rest/auth/auth_session.cc
@@ -28,6 +28,7 @@
 #include <utility>
 
 #include "iceberg/catalog/rest/auth/auth_properties.h"
+#include "iceberg/catalog/rest/auth/auth_session_internal.h"
 #include "iceberg/catalog/rest/auth/oauth2_util.h"
 #include "iceberg/catalog/rest/auth/token_refresh_scheduler.h"
 #include "iceberg/catalog/rest/http_client.h"
@@ -54,229 +55,6 @@ class DefaultAuthSession : public AuthSession {
   std::unordered_map<std::string, std::string> headers_;
 };
 
-/// \brief OAuth2 session with automatic token refresh.
-class OAuth2AuthSession : public AuthSession,
-                          public std::enable_shared_from_this<OAuth2AuthSession> {
- public:
-  struct Config {
-    std::string token_endpoint;
-    std::string client_id;
-    std::string client_secret;
-    std::string scope;
-    std::unordered_map<std::string, std::string> optional_oauth_params;
-    bool keep_refreshed;
-  };
-
-  /// \brief Create an OAuth2 session and optionally schedule refresh.
-  static Result<std::shared_ptr<OAuth2AuthSession>> Make(
-      const OAuthTokenResponse& initial_token, Config config, HttpClient& client) {
-    ICEBERG_ASSIGN_OR_RAISE(auto refresh_properties, MakeRefreshProperties(config));
-    auto session = std::shared_ptr<OAuth2AuthSession>(
-        new OAuth2AuthSession(std::move(config), std::move(refresh_properties), client));
-    session->SetInitialToken(initial_token);
-    return session;
-  }
-
-  Result<HttpRequest> Authenticate(HttpRequest request) override {
-    std::shared_lock lock(mutex_);
-    for (const auto& [key, value] : headers_) {
-      request.headers.try_emplace(key, value);
-    }
-    return request;
-  }
-
-  Status Close() override { return CloseImpl(); }
-
-  ~OAuth2AuthSession() override { std::ignore = CloseImpl(); }
-
- private:
-  OAuth2AuthSession(Config config, AuthProperties refresh_properties, HttpClient& client)
-      : config_(std::move(config)),
-        refresh_properties_(std::move(refresh_properties)),
-        client_(client) {}
-
-  Status CloseImpl() {
-    bool expected = false;
-    if (!closed_.compare_exchange_strong(expected, true)) {
-      return {};  // Already closed
-    }
-    TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
-    std::unique_lock lock(refresh_mutex_);
-    refresh_cv_.wait(lock, [this] { return active_refresh_count_ == 0; });
-    TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
-    return {};
-  }
-
-  static Result<AuthProperties> MakeRefreshProperties(const Config& config) {
-    std::unordered_map<std::string, std::string> properties =
-        config.optional_oauth_params;
-    properties[AuthProperties::kCredential.key()] =
-        config.client_id.empty() ? config.client_secret
-                                 : config.client_id + ":" + config.client_secret;
-    properties[AuthProperties::kScope.key()] = config.scope;
-    properties[AuthProperties::kOAuth2ServerUri.key()] = config.token_endpoint;
-
-    return AuthProperties::FromProperties(properties);
-  }
-
-  class RefreshAttemptGuard {
-   public:
-    explicit RefreshAttemptGuard(OAuth2AuthSession& session) : session_(session) {
-      std::lock_guard lock(session_.refresh_mutex_);
-      ++session_.active_refresh_count_;
-    }
-
-    ~RefreshAttemptGuard() {
-      bool notify = false;
-      {
-        std::lock_guard lock(session_.refresh_mutex_);
-        notify = --session_.active_refresh_count_ == 0;
-      }
-      if (notify) {
-        session_.refresh_cv_.notify_all();
-      }
-    }
-
-   private:
-    OAuth2AuthSession& session_;
-  };
-
-  void SetInitialToken(const OAuthTokenResponse& token_response) {
-    token_ = token_response.access_token;
-    headers_ = {{std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token_}};
-
-    // Determine expiration time
-    if (token_response.expires_in_secs.has_value()) {
-      expires_at_ = std::chrono::steady_clock::now() +
-                    std::chrono::seconds(*token_response.expires_in_secs);
-    } else if (auto exp_ms = ExpiresAtMillis(token_); exp_ms.has_value()) {
-      // Convert absolute epoch millis to steady_clock time_point
-      auto now_sys = std::chrono::system_clock::now();
-      auto now_steady = std::chrono::steady_clock::now();
-      auto exp_sys =
-          std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms));
-      expires_at_ = now_steady + (exp_sys - now_sys);
-    }
-
-    if (config_.keep_refreshed &&
-        expires_at_ != std::chrono::steady_clock::time_point{}) {
-      ScheduleRefresh();
-    }
-  }
-
-  void DoRefresh() { DoRefreshAttempt(0, std::chrono::milliseconds(200)); }
-
-  /// \brief Single refresh attempt. On failure, schedules a retry via the
-  /// scheduler (non-blocking) instead of sleeping on the worker thread.
-  void DoRefreshAttempt(int attempt, std::chrono::milliseconds backoff) {
-    static constexpr int kMaxRetries = 5;
-    static constexpr auto kMaxBackoff = std::chrono::milliseconds(10'000);
-
-    RefreshAttemptGuard guard(*this);
-    if (closed_.load()) return;
-
-    // Use an empty session for the refresh request (no auth headers —
-    // avoids circular dependency of using an expired token to refresh itself)
-    auto empty_session = AuthSession::MakeDefault({});
-
-    auto result = FetchToken(client_, *empty_session, refresh_properties_);
-    if (result.has_value()) {
-      auto& response = result.value();
-      {
-        std::unique_lock lock(mutex_);
-        token_ = response.access_token;
-        headers_ = {
-            {std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token_}};
-
-        // Reset before deriving new expiry
-        expires_at_ = std::chrono::steady_clock::time_point{};
-
-        if (response.expires_in_secs.has_value()) {
-          expires_at_ = std::chrono::steady_clock::now() +
-                        std::chrono::seconds(*response.expires_in_secs);
-        } else if (auto exp_ms = ExpiresAtMillis(token_); exp_ms.has_value()) {
-          auto now_sys = std::chrono::system_clock::now();
-          auto now_steady = std::chrono::steady_clock::now();
-          auto exp_sys =
-              std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms));
-          expires_at_ = now_steady + (exp_sys - now_sys);
-        }
-      }
-      // Note: ScheduleRefresh must be called outside the lock.
-      ScheduleRefresh();
-      return;  // Success
-    }
-
-    // Schedule retry with exponential backoff (non-blocking)
-    if (attempt + 1 < kMaxRetries && !closed_.load()) {
-      auto next_backoff =
-          std::min(std::chrono::duration_cast<std::chrono::milliseconds>(backoff * 2),
-                   kMaxBackoff);
-      std::weak_ptr<OAuth2AuthSession> weak_self = shared_from_this();
-      auto retry_id = TokenRefreshScheduler::Instance().Schedule(
-          backoff,
-          [weak_self = std::move(weak_self), next_attempt = attempt + 1, next_backoff] {
-            if (auto self = weak_self.lock()) {
-              self->DoRefreshAttempt(next_attempt, next_backoff);
-            }
-          });
-      scheduled_task_id_.store(retry_id);
-    }
-    // All retries exhausted — stop refreshing silently.
-    // Next request will use the expired token; server returns 401.
-  }
-
-  /// \brief Schedule the next token refresh based on expiration time.
-  ///
-  /// Must be called outside any lock on mutex_ (CalculateRefreshDelay
-  /// acquires shared_lock internally).
-  void ScheduleRefresh() {
-    if (!config_.keep_refreshed || closed_.load()) return;
-
-    auto delay = CalculateRefreshDelay();
-    if (delay < std::chrono::milliseconds::zero()) return;
-
-    std::weak_ptr<OAuth2AuthSession> weak_self = shared_from_this();
-    auto new_id = TokenRefreshScheduler::Instance().Schedule(
-        delay, [weak_self = std::move(weak_self)] {
-          if (auto self = weak_self.lock()) {
-            self->DoRefresh();
-          }
-        });
-    scheduled_task_id_.store(new_id);
-  }
-
-  std::chrono::milliseconds CalculateRefreshDelay() const {
-    std::shared_lock lock(mutex_);
-    auto now = std::chrono::steady_clock::now();
-    if (expires_at_ == std::chrono::steady_clock::time_point{}) {
-      return std::chrono::milliseconds(-1);
-    }
-    if (expires_at_ <= now) return std::chrono::milliseconds::zero();
-
-    auto expires_in =
-        std::chrono::duration_cast<std::chrono::milliseconds>(expires_at_ - now);
-    // Refresh window: 10% of remaining time, capped at 5 minutes
-    auto refresh_window = std::min(expires_in / 10, std::chrono::milliseconds(300'000));
-    auto wait_time = expires_in - refresh_window;
-    return std::max(wait_time, std::chrono::milliseconds(10));
-  }
-
-  mutable std::shared_mutex mutex_;  // protects token_, headers_, expires_at_
-  std::string token_;
-  std::unordered_map<std::string, std::string> headers_;
-  std::chrono::steady_clock::time_point expires_at_{};
-
-  Config config_;
-  AuthProperties refresh_properties_;
-  HttpClient& client_;  // It should outlive the session
-  std::atomic<uint64_t> scheduled_task_id_{0};
-  std::atomic<bool> closed_{false};
-  std::mutex refresh_mutex_;
-  std::condition_variable refresh_cv_;
-  int active_refresh_count_ = 0;
-};
-
 }  // namespace
 
 std::shared_ptr<AuthSession> AuthSession::MakeDefault(
@@ -289,8 +67,20 @@ Result<std::shared_ptr<AuthSession>> AuthSession::MakeOAuth2(
     const std::string& client_id, const std::string& client_secret,
     const std::string& scope, bool keep_refreshed,
     const std::unordered_map<std::string, std::string>& optional_oauth_params,
-    HttpClient& client) {
-  OAuth2AuthSession::Config config{
+    std::shared_ptr<HttpClient> client) {
+  return internal::MakeOAuth2Session(
+      initial_token, token_endpoint, client_id, client_secret, scope, keep_refreshed,
+      optional_oauth_params, std::move(client), std::nullopt);
+}
+
+Result<std::shared_ptr<AuthSession>> internal::MakeOAuth2Session(
+    const OAuthTokenResponse& initial_token, const std::string& token_endpoint,
+    const std::string& client_id, const std::string& client_secret,
+    const std::string& scope, bool keep_refreshed,
+    const std::unordered_map<std::string, std::string>& optional_oauth_params,
+    std::shared_ptr<HttpClient> client,
+    std::optional<std::chrono::steady_clock::time_point> token_request_started_at) {
+  internal::OAuth2Session::Config config{
       .token_endpoint = token_endpoint,
       .client_id = client_id,
       .client_secret = client_secret,
@@ -298,8 +88,9 @@ Result<std::shared_ptr<AuthSession>> AuthSession::MakeOAuth2(
       .optional_oauth_params = optional_oauth_params,
       .keep_refreshed = keep_refreshed,
   };
-  ICEBERG_ASSIGN_OR_RAISE(
-      auto session, OAuth2AuthSession::Make(initial_token, std::move(config), client));
+  ICEBERG_ASSIGN_OR_RAISE(auto session, internal::OAuth2Session::Make(
+                                            initial_token, std::move(config),
+                                            std::move(client), token_request_started_at));
   return std::static_pointer_cast<AuthSession>(std::move(session));
 }
 
diff --git a/src/iceberg/catalog/rest/auth/auth_session.h b/src/iceberg/catalog/rest/auth/auth_session.h
index 3d0063a04..bdc77ebc4 100644
--- a/src/iceberg/catalog/rest/auth/auth_session.h
+++ b/src/iceberg/catalog/rest/auth/auth_session.h
@@ -20,6 +20,7 @@
 #pragma once
 
 #include <memory>
+#include <optional>
 #include <string>
 #include <unordered_map>
 
@@ -33,6 +34,16 @@
 
 namespace iceberg::rest::auth {
 
+/// \brief OAuth2 metadata used to derive child authentication sessions.
+struct ICEBERG_REST_EXPORT OAuth2SessionInfo {
+  std::string token;
+  std::string issued_token_type;
+  std::string credential;
+  std::string scope;
+  std::string oauth2_server_uri;
+  std::unordered_map<std::string, std::string> optional_oauth_params;
+};
+
 /// \brief An authentication session that can authenticate outgoing HTTP requests.
 class ICEBERG_REST_EXPORT AuthSession {
  public:
@@ -54,6 +65,9 @@ class ICEBERG_REST_EXPORT AuthSession {
   ///         - RestError: HTTP errors from authentication service
   virtual Result<HttpRequest> Authenticate(HttpRequest request) = 0;
 
+  /// \brief Return OAuth2 metadata when this is an OAuth2 session.
+  virtual std::optional<OAuth2SessionInfo> OAuth2Info() const { return std::nullopt; }
+
   /// \brief Close the session and release any resources.
   ///
   /// This method is called when the session is no longer needed. For stateful
@@ -63,7 +77,7 @@ class ICEBERG_REST_EXPORT AuthSession {
   /// \return Status indicating success or failure of closing the session.
   virtual Status Close() { return {}; }
 
-  /// \brief Create a default session with static headers.
+  /// \brief Create a session with static headers.
   ///
   /// This factory method creates a session that adds a fixed set of headers to each
   /// request. It is suitable for authentication methods that use static credentials,
@@ -88,15 +102,15 @@ class ICEBERG_REST_EXPORT AuthSession {
   /// \param scope OAuth2 scope for refresh requests.
   /// \param keep_refreshed Whether to schedule automatic token refresh.
   /// \param optional_oauth_params Optional OAuth params (audience, resource) for refresh.
-  /// \param client HTTP client for making refresh requests. The caller owns the
-  ///        client and must keep it alive until the session is closed.
+  /// \param client HTTP client for making refresh requests. The session retains
+  ///        ownership of the client.
   /// \return A new session that manages token lifecycle automatically.
   static Result<std::shared_ptr<AuthSession>> MakeOAuth2(
       const OAuthTokenResponse& initial_token, const std::string& token_endpoint,
       const std::string& client_id, const std::string& client_secret,
       const std::string& scope, bool keep_refreshed,
       const std::unordered_map<std::string, std::string>& optional_oauth_params,
-      HttpClient& client);
+      std::shared_ptr<HttpClient> client);
 };
 
 }  // namespace iceberg::rest::auth
diff --git a/src/iceberg/catalog/rest/auth/auth_session_internal.h b/src/iceberg/catalog/rest/auth/auth_session_internal.h
new file mode 100644
index 000000000..b8db3836e
--- /dev/null
+++ b/src/iceberg/catalog/rest/auth/auth_session_internal.h
@@ -0,0 +1,282 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+#pragma once
+
+#include <algorithm>
+#include <atomic>
+#include <chrono>
+#include <condition_variable>
+#include <memory>
+#include <mutex>
+#include <optional>
+#include <shared_mutex>
+#include <string>
+#include <unordered_map>
+#include <utility>
+
+#include "iceberg/catalog/rest/auth/auth_properties.h"
+#include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/oauth2_util.h"
+#include "iceberg/catalog/rest/auth/token_refresh_scheduler.h"
+#include "iceberg/catalog/rest/http_client.h"
+#include "iceberg/util/macros.h"
+
+namespace iceberg::rest::auth::internal {
+
+inline std::optional<std::chrono::steady_clock::time_point> TokenExpirationTime(
+    const OAuthTokenResponse& response,
+    std::chrono::steady_clock::time_point request_started_at,
+    std::chrono::system_clock::time_point now_system = std::chrono::system_clock::now(),
+    std::chrono::steady_clock::time_point now_steady = std::chrono::steady_clock::now()) {
+  if (auto exp_ms = OAuth2Util::ExpiresAtMillis(response.access_token);
+      exp_ms.has_value()) {
+    auto expiration_system =
+        std::chrono::system_clock::time_point(std::chrono::milliseconds(*exp_ms));
+    return now_steady + (expiration_system - now_system);
+  }
+  if (response.expires_in_secs.has_value()) {
+    return request_started_at + std::chrono::seconds(*response.expires_in_secs);
+  }
+  return std::nullopt;
+}
+
+/// \brief Internal OAuth2 authentication session.
+class OAuth2Session final : public AuthSession,
+                            public std::enable_shared_from_this<OAuth2Session> {
+ public:
+  struct Config {
+    std::string token_endpoint;
+    std::string client_id;
+    std::string client_secret;
+    std::string scope;
+    std::unordered_map<std::string, std::string> optional_oauth_params;
+    bool keep_refreshed;
+  };
+
+  static Result<std::shared_ptr<OAuth2Session>> Make(
+      const OAuthTokenResponse& initial_token, Config config,
+      std::shared_ptr<HttpClient> client,
+      std::optional<std::chrono::steady_clock::time_point> token_request_started_at) {
+    ICEBERG_PRECHECK(client != nullptr, "OAuth2 session HTTP client must not be null");
+    ICEBERG_ASSIGN_OR_RAISE(auto refresh_properties, MakeRefreshProperties(config));
+    auto session = std::shared_ptr<OAuth2Session>(new OAuth2Session(
+        std::move(config), std::move(refresh_properties), std::move(client)));
+    session->SetInitialToken(initial_token, token_request_started_at);
+    return session;
+  }
+
+  Result<HttpRequest> Authenticate(HttpRequest request) override {
+    std::shared_lock lock(mutex_);
+    for (const auto& [key, value] : headers_) {
+      request.headers.try_emplace(key, value);
+    }
+    return request;
+  }
+
+  std::optional<OAuth2SessionInfo> OAuth2Info() const override {
+    std::shared_lock lock(mutex_);
+    return OAuth2SessionInfo{
+        .token = token_,
+        .issued_token_type = issued_token_type_,
+        .credential = Credential(config_),
+        .scope = config_.scope,
+        .oauth2_server_uri = config_.token_endpoint,
+        .optional_oauth_params = config_.optional_oauth_params,
+    };
+  }
+
+  Status Close() override { return CloseImpl(); }
+
+  ~OAuth2Session() override { std::ignore = CloseImpl(); }
+
+ private:
+  OAuth2Session(Config config, AuthProperties refresh_properties,
+                std::shared_ptr<HttpClient> client)
+      : config_(std::move(config)),
+        refresh_properties_(std::move(refresh_properties)),
+        client_(std::move(client)) {}
+
+  Status CloseImpl() {
+    bool expected = false;
+    if (!closed_.compare_exchange_strong(expected, true)) {
+      return {};
+    }
+    TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
+    std::unique_lock lock(refresh_mutex_);
+    refresh_cv_.wait(lock, [this] { return active_refresh_count_ == 0; });
+    TokenRefreshScheduler::Instance().Cancel(scheduled_task_id_.exchange(0));
+    return {};
+  }
+
+  static std::string Credential(const Config& config) {
+    return config.client_id.empty() ? config.client_secret
+                                    : config.client_id + ":" + config.client_secret;
+  }
+
+  static Result<AuthProperties> MakeRefreshProperties(const Config& config) {
+    std::unordered_map<std::string, std::string> properties =
+        config.optional_oauth_params;
+    properties[AuthProperties::kCredential.key()] = Credential(config);
+    properties[AuthProperties::kScope.key()] = config.scope;
+    properties[AuthProperties::kOAuth2ServerUri.key()] = config.token_endpoint;
+    return AuthProperties::FromProperties(properties);
+  }
+
+  class RefreshAttemptGuard {
+   public:
+    explicit RefreshAttemptGuard(OAuth2Session& session) : session_(session) {
+      std::lock_guard lock(session_.refresh_mutex_);
+      ++session_.active_refresh_count_;
+    }
+
+    ~RefreshAttemptGuard() {
+      bool notify = false;
+      {
+        std::lock_guard lock(session_.refresh_mutex_);
+        notify = --session_.active_refresh_count_ == 0;
+      }
+      if (notify) {
+        session_.refresh_cv_.notify_all();
+      }
+    }
+
+   private:
+    OAuth2Session& session_;
+  };
+
+  void UpdateTokenState(const OAuthTokenResponse& token_response,
+                        std::optional<std::chrono::steady_clock::time_point>
+                            token_request_started_at = std::nullopt) {
+    token_ = token_response.access_token;
+    issued_token_type_ = token_response.issued_token_type.empty()
+                             ? AuthProperties::kAccessTokenType
+                             : token_response.issued_token_type;
+    headers_ = OAuth2Util::AuthHeaders(token_);
+    expires_at_ = std::chrono::steady_clock::time_point{};
+    auto request_started_at =
+        token_request_started_at.value_or(std::chrono::steady_clock::now());
+    if (auto expiration = TokenExpirationTime(token_response, request_started_at);
+        expiration.has_value()) {
+      expires_at_ = *expiration;
+    }
+  }
+
+  void SetInitialToken(
+      const OAuthTokenResponse& token_response,
+      std::optional<std::chrono::steady_clock::time_point> token_request_started_at) {
+    UpdateTokenState(token_response, token_request_started_at);
+    if (config_.keep_refreshed &&
+        expires_at_ != std::chrono::steady_clock::time_point{}) {
+      ScheduleRefresh();
+    }
+  }
+
+  void DoRefresh() {
+    DoRefreshAttempt(0, std::chrono::milliseconds(200), std::chrono::steady_clock::now());
+  }
+
+  void DoRefreshAttempt(int attempt, std::chrono::milliseconds backoff,
+                        std::chrono::steady_clock::time_point refresh_started_at) {
+    static constexpr int kMaxRetries = 5;
+    static constexpr auto kMaxBackoff = std::chrono::milliseconds(10'000);
+    RefreshAttemptGuard guard(*this);
+    if (closed_.load()) return;
+
+    auto empty_session = AuthSession::MakeDefault({});
+    // TODO(lishuxu): Honor token-exchange-enabled and refresh via token exchange,
+    // matching Java.
+    auto result = OAuth2Util::FetchToken(*client_, *empty_session, refresh_properties_);
+    if (result.has_value()) {
+      auto& response = result.value();
+      {
+        std::unique_lock lock(mutex_);
+        UpdateTokenState(response, refresh_started_at);
+      }
+      ScheduleRefresh();
+      return;
+    }
+
+    if (attempt + 1 < kMaxRetries && !closed_.load()) {
+      auto next_backoff =
+          std::min(std::chrono::duration_cast<std::chrono::milliseconds>(backoff * 2),
+                   kMaxBackoff);
+      std::weak_ptr<OAuth2Session> weak_self = shared_from_this();
+      auto retry_id = TokenRefreshScheduler::Instance().Schedule(
+          backoff, [weak_self = std::move(weak_self), next_attempt = attempt + 1,
+                    next_backoff, refresh_started_at] {
+            if (auto self = weak_self.lock()) {
+              self->DoRefreshAttempt(next_attempt, next_backoff, refresh_started_at);
+            }
+          });
+      scheduled_task_id_.store(retry_id);
+    }
+  }
+
+  void ScheduleRefresh() {
+    if (!config_.keep_refreshed || closed_.load()) return;
+    auto delay = CalculateRefreshDelay();
+    if (delay < std::chrono::milliseconds::zero()) return;
+
+    std::weak_ptr<OAuth2Session> weak_self = shared_from_this();
+    auto new_id = TokenRefreshScheduler::Instance().Schedule(
+        delay, [weak_self = std::move(weak_self)] {
+          if (auto self = weak_self.lock()) self->DoRefresh();
+        });
+    scheduled_task_id_.store(new_id);
+  }
+
+  std::chrono::milliseconds CalculateRefreshDelay() const {
+    std::shared_lock lock(mutex_);
+    auto now = std::chrono::steady_clock::now();
+    if (expires_at_ == std::chrono::steady_clock::time_point{}) {
+      return std::chrono::milliseconds(-1);
+    }
+    if (expires_at_ <= now) return std::chrono::milliseconds::zero();
+    auto expires_in =
+        std::chrono::duration_cast<std::chrono::milliseconds>(expires_at_ - now);
+    auto refresh_window = std::min(expires_in / 10, std::chrono::milliseconds(300'000));
+    auto wait_time = expires_in - refresh_window;
+    return std::max(wait_time, std::chrono::milliseconds(10));
+  }
+
+  mutable std::shared_mutex mutex_;
+  std::string token_;
+  std::string issued_token_type_;
+  std::unordered_map<std::string, std::string> headers_;
+  std::chrono::steady_clock::time_point expires_at_{};
+  Config config_;
+  AuthProperties refresh_properties_;
+  std::shared_ptr<HttpClient> client_;
+  std::atomic<uint64_t> scheduled_task_id_{0};
+  std::atomic<bool> closed_{false};
+  std::mutex refresh_mutex_;
+  std::condition_variable refresh_cv_;
+  int active_refresh_count_ = 0;
+};
+
+Result<std::shared_ptr<AuthSession>> MakeOAuth2Session(
+    const OAuthTokenResponse& initial_token, const std::string& token_endpoint,
+    const std::string& client_id, const std::string& client_secret,
+    const std::string& scope, bool keep_refreshed,
+    const std::unordered_map<std::string, std::string>& optional_oauth_params,
+    std::shared_ptr<HttpClient> client,
+    std::optional<std::chrono::steady_clock::time_point> token_request_started_at);
+
+}  // namespace iceberg::rest::auth::internal
diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.cc b/src/iceberg/catalog/rest/auth/oauth2_util.cc
index d5e94821c..d62ac7e1a 100644
--- a/src/iceberg/catalog/rest/auth/oauth2_util.cc
+++ b/src/iceberg/catalog/rest/auth/oauth2_util.cc
@@ -36,21 +36,101 @@ namespace {
 
 constexpr std::string_view kGrantType = "grant_type";
 constexpr std::string_view kClientCredentials = "client_credentials";
+constexpr std::string_view kTokenExchange =
+    "urn:ietf:params:oauth:grant-type:token-exchange";
 constexpr std::string_view kClientId = "client_id";
 constexpr std::string_view kClientSecret = "client_secret";
 constexpr std::string_view kScope = "scope";
+constexpr std::string_view kSubjectToken = "subject_token";
+constexpr std::string_view kSubjectTokenType = "subject_token_type";
+constexpr std::string_view kActorToken = "actor_token";
+constexpr std::string_view kActorTokenType = "actor_token_type";
+constexpr std::string_view kAuthorizationHeader = "Authorization";
+constexpr std::string_view kBearerPrefix = "Bearer ";
+
+Result<OAuthTokenResponse> ParseTokenResponse(const std::string& response_body) {
+  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response_body));
+  ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson<OAuthTokenResponse>(json));
+  ICEBERG_RETURN_UNEXPECTED(token_response.Validate());
+  return token_response;
+}
+
+bool IsValidTokenType(std::string_view token_type) {
+  return token_type == AuthProperties::kAccessTokenType ||
+         token_type == AuthProperties::kRefreshTokenType ||
+         token_type == AuthProperties::kIdTokenType ||
+         token_type == AuthProperties::kSaml1TokenType ||
+         token_type == AuthProperties::kSaml2TokenType ||
+         token_type == AuthProperties::kJwtTokenType;
+}
 
 }  // namespace
 
-std::unordered_map<std::string, std::string> AuthHeaders(const std::string& token) {
+std::unordered_map<std::string, std::string> OAuth2Util::AuthHeaders(
+    const std::string& token) {
   if (!token.empty()) {
     return {{std::string(kAuthorizationHeader), std::string(kBearerPrefix) + token}};
   }
   return {};
 }
 
-Result<OAuthTokenResponse> FetchToken(HttpClient& client, AuthSession& session,
-                                      const AuthProperties& properties) {
+Result<std::unordered_map<std::string, std::string>> OAuth2Util::TokenExchangeRequest(
+    const std::string& subject_token, const std::string& subject_token_type,
+    const std::optional<std::string>& actor_token,
+    const std::optional<std::string>& actor_token_type, const std::string& scope,
+    const std::unordered_map<std::string, std::string>& optional_params) {
+  if (subject_token.empty()) {
+    return InvalidArgument("OAuth2 subject token must not be empty");
+  }
+  if (!IsValidTokenType(subject_token_type)) {
+    return InvalidArgument("Invalid OAuth2 subject token type: '{}'", subject_token_type);
+  }
+  if (actor_token.has_value()) {
+    if (actor_token->empty()) {
+      return InvalidArgument("OAuth2 actor token must not be empty");
+    }
+    if (!actor_token_type.has_value() || !IsValidTokenType(*actor_token_type)) {
+      return InvalidArgument("Invalid OAuth2 actor token type: '{}'",
+                             actor_token_type.value_or(""));
+    }
+  }
+
+  std::unordered_map<std::string, std::string> form_data{
+      {std::string(kGrantType), std::string(kTokenExchange)},
+      {std::string(kScope), scope},
+      {std::string(kSubjectToken), subject_token},
+      {std::string(kSubjectTokenType), subject_token_type},
+  };
+  if (actor_token.has_value()) {
+    form_data.emplace(kActorToken, *actor_token);
+    form_data.emplace(kActorTokenType, *actor_token_type);
+  }
+  for (const auto& [key, value] : optional_params) {
+    form_data.insert_or_assign(key, value);
+  }
+  return form_data;
+}
+
+Result<OAuthTokenResponse> OAuth2Util::ExchangeToken(
+    HttpClient& client, AuthSession& session,
+    const std::unordered_map<std::string, std::string>& extra_headers,
+    const std::string& subject_token, const std::string& subject_token_type,
+    const std::optional<std::string>& actor_token,
+    const std::optional<std::string>& actor_token_type, const std::string& scope,
+    const std::string& oauth2_server_uri,
+    const std::unordered_map<std::string, std::string>& optional_params) {
+  ICEBERG_ASSIGN_OR_RAISE(
+      auto form_data, TokenExchangeRequest(subject_token, subject_token_type, actor_token,
+                                           actor_token_type, scope, optional_params));
+  ICEBERG_ASSIGN_OR_RAISE(auto response,
+                          client.PostForm(oauth2_server_uri, form_data, extra_headers,
+                                          *OAuthErrorHandler::Instance(), session));
+  return ParseTokenResponse(response.body());
+}
+
+Result<OAuthTokenResponse> OAuth2Util::FetchToken(HttpClient& client,
+                                                  AuthSession& session,
+                                                  const AuthProperties& properties) {
   std::unordered_map<std::string, std::string> form_data{
       {std::string(kGrantType), std::string(kClientCredentials)},
       {std::string(kClientSecret), properties.client_secret()},
@@ -67,14 +147,10 @@ Result<OAuthTokenResponse> FetchToken(HttpClient& client, AuthSession& session,
       auto response,
       client.PostForm(properties.oauth2_server_uri(), form_data,
                       /*headers=*/{}, *OAuthErrorHandler::Instance(), session));
-
-  ICEBERG_ASSIGN_OR_RAISE(auto json, FromJsonString(response.body()));
-  ICEBERG_ASSIGN_OR_RAISE(auto token_response, FromJson<OAuthTokenResponse>(json));
-  ICEBERG_RETURN_UNEXPECTED(token_response.Validate());
-  return token_response;
+  return ParseTokenResponse(response.body());
 }
 
-std::optional<int64_t> ExpiresAtMillis(std::string_view token) {
+std::optional<int64_t> OAuth2Util::ExpiresAtMillis(std::string_view token) {
   if (token.empty()) {
     return std::nullopt;
   }
diff --git a/src/iceberg/catalog/rest/auth/oauth2_util.h b/src/iceberg/catalog/rest/auth/oauth2_util.h
index 428ebc385..78a29050b 100644
--- a/src/iceberg/catalog/rest/auth/oauth2_util.h
+++ b/src/iceberg/catalog/rest/auth/oauth2_util.h
@@ -35,34 +35,65 @@
 
 namespace iceberg::rest::auth {
 
-inline constexpr std::string_view kAuthorizationHeader = "Authorization";
-inline constexpr std::string_view kBearerPrefix = "Bearer ";
+/// \brief OAuth2 token and authentication utilities.
+class ICEBERG_REST_EXPORT OAuth2Util {
+ public:
+  OAuth2Util() = delete;
 
-/// \brief Fetch an OAuth2 token using the client_credentials grant type.
-///
-/// \param client HTTP client to use for the request.
-/// \param session Auth session for the request headers.
-/// \param properties Auth configuration containing credential, scope,
-///        token endpoint, and optional OAuth params.
-/// \return The token response or an error.
-ICEBERG_REST_EXPORT Result<OAuthTokenResponse> FetchToken(
-    HttpClient& client, AuthSession& session, const AuthProperties& properties);
+  /// \brief Fetch an OAuth2 token using the client_credentials grant type.
+  ///
+  /// \param client HTTP client to use for the request.
+  /// \param session Auth session for the request headers.
+  /// \param properties Auth configuration containing credential, scope,
+  ///        token endpoint, and optional OAuth params.
+  /// \return The token response or an error.
+  static Result<OAuthTokenResponse> FetchToken(HttpClient& client, AuthSession& session,
+                                               const AuthProperties& properties);
 
-/// \brief Build auth headers from a token string.
-///
-/// \param token Bearer token string (may be empty).
-/// \return Headers map with Authorization header if token is non-empty.
-ICEBERG_REST_EXPORT std::unordered_map<std::string, std::string> AuthHeaders(
-    const std::string& token);
+  /// \brief Build auth headers from a token string.
+  ///
+  /// \param token Bearer token string (may be empty).
+  /// \return Headers map with Authorization header if token is non-empty.
+  static std::unordered_map<std::string, std::string> AuthHeaders(
+      const std::string& token);
 
-/// \brief Extract expiration time from a JWT token.
-///
-/// Decodes the JWT payload (base64url) and reads the "exp" claim.
-/// Returns std::nullopt if the token is not a valid JWT or has no "exp" claim.
-///
-/// \param token A token string. If it is a JWT (three dot-separated base64url
-///        segments), the "exp" claim is extracted from the payload.
-/// \return Expiration time as milliseconds since epoch, or std::nullopt.
-ICEBERG_REST_EXPORT std::optional<int64_t> ExpiresAtMillis(std::string_view token);
+  /// \brief Exchange an OAuth2 token using the RFC 8693 grant type.
+  ///
+  /// \param client HTTP client to use for the request.
+  /// \param session Auth session for the request headers.
+  /// \param extra_headers Request headers applied before session authentication.
+  /// \param subject_token Subject token to exchange.
+  /// \param subject_token_type Subject token type.
+  /// \param actor_token Optional actor token.
+  /// \param actor_token_type Optional actor token type.
+  /// \param scope OAuth2 scope.
+  /// \param oauth2_server_uri Token exchange endpoint.
+  /// \param optional_params Optional OAuth parameters.
+  /// \return The token response or an error.
+  static Result<OAuthTokenResponse> ExchangeToken(
+      HttpClient& client, AuthSession& session,
+      const std::unordered_map<std::string, std::string>& extra_headers,
+      const std::string& subject_token, const std::string& subject_token_type,
+      const std::optional<std::string>& actor_token,
+      const std::optional<std::string>& actor_token_type, const std::string& scope,
+      const std::string& oauth2_server_uri,
+      const std::unordered_map<std::string, std::string>& optional_params);
+
+  /// \brief Extract expiration time from a JWT token.
+  ///
+  /// Decodes the JWT payload (base64url) and reads the "exp" claim.
+  /// Returns std::nullopt if the token is not a valid JWT or has no "exp" claim.
+  ///
+  /// \param token A token string containing three dot-separated JWT segments.
+  /// \return Expiration time as milliseconds since epoch, or std::nullopt.
+  static std::optional<int64_t> ExpiresAtMillis(std::string_view token);
+
+ private:
+  static Result<std::unordered_map<std::string, std::string>> TokenExchangeRequest(
+      const std::string& subject_token, const std::string& subject_token_type,
+      const std::optional<std::string>& actor_token,
+      const std::optional<std::string>& actor_token_type, const std::string& scope,
+      const std::unordered_map<std::string, std::string>& optional_params);
+};
 
 }  // namespace iceberg::rest::auth
diff --git a/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h b/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h
index 53ec0853a..7bb7a3aa3 100644
--- a/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h
+++ b/src/iceberg/catalog/rest/auth/sigv4_auth_manager_internal.h
@@ -118,11 +118,11 @@ class ICEBERG_REST_EXPORT SigV4AuthManager : public AuthManager {
   ~SigV4AuthManager() override;
 
   Result<std::shared_ptr<AuthSession>> InitSession(
-      HttpClient& init_client,
+      std::shared_ptr<HttpClient> init_client,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::shared_ptr<AuthSession>> CatalogSession(
-      HttpClient& shared_client,
+      std::shared_ptr<HttpClient> shared_client,
       const std::unordered_map<std::string, std::string>& properties) override;
 
   Result<std::shared_ptr<AuthSession>> ContextualSession(
diff --git a/src/iceberg/catalog/rest/auth/sigv4_manager.cc b/src/iceberg/catalog/rest/auth/sigv4_manager.cc
index 6678f1b3f..6e0dce86b 100644
--- a/src/iceberg/catalog/rest/auth/sigv4_manager.cc
+++ b/src/iceberg/catalog/rest/auth/sigv4_manager.cc
@@ -51,6 +51,7 @@ namespace iceberg::rest::auth {
 
 namespace {
 
+constexpr std::string_view kAuthorizationHeader = "Authorization";
 constexpr std::string_view kAmzContentSha256Header = "x-amz-content-sha256";
 
 class AwsSdkLifecycle {
@@ -379,7 +380,7 @@ SigV4AuthManager::SigV4AuthManager(std::unique_ptr<AuthManager> delegate)
 SigV4AuthManager::~SigV4AuthManager() = default;
 
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::InitSession(
-    HttpClient& init_client,
+    std::shared_ptr<HttpClient> init_client,
     const std::unordered_map<std::string, std::string>& properties) {
   ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized());
   ICEBERG_ASSIGN_OR_RAISE(auto delegate_session,
@@ -389,7 +390,7 @@ Result<std::shared_ptr<AuthSession>> SigV4AuthManager::InitSession(
 }
 
 Result<std::shared_ptr<AuthSession>> SigV4AuthManager::CatalogSession(
-    HttpClient& shared_client,
+    std::shared_ptr<HttpClient> shared_client,
     const std::unordered_map<std::string, std::string>& properties) {
   ICEBERG_RETURN_UNEXPECTED(AwsSdkLifecycle::Instance().EnsureInitialized());
   catalog_properties_ = properties;
diff --git a/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h b/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h
index 02dc0e14f..5ef20ed94 100644
--- a/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h
+++ b/src/iceberg/catalog/rest/auth/token_refresh_scheduler.h
@@ -37,7 +37,7 @@ namespace iceberg::rest::auth {
 /// \brief A process-global scheduler for delayed token refresh tasks.
 ///
 /// Uses a single background thread that sleeps until the next task is due.
-/// All OAuth2AuthSession instances share this scheduler. Tasks are lightweight
+/// All OAuth2Session instances share this scheduler. Tasks are lightweight
 /// (a single HTTP POST to refresh a token), so one thread is sufficient.
 ///
 /// Thread safety: All public methods are thread-safe.
diff --git a/src/iceberg/catalog/rest/resource_paths.cc b/src/iceberg/catalog/rest/resource_paths.cc
index d18dd4636..3a70eb113 100644
--- a/src/iceberg/catalog/rest/resource_paths.cc
+++ b/src/iceberg/catalog/rest/resource_paths.cc
@@ -51,7 +51,7 @@ Result<std::string> ResourcePaths::Config() const {
 }
 
 Result<std::string> ResourcePaths::OAuth2Tokens() const {
-  return std::format("{}/v1/{}oauth/tokens", base_uri_, prefix_);
+  return std::format("{}/v1/oauth/tokens", base_uri_);
 }
 
 Result<std::string> ResourcePaths::Namespaces() const {
diff --git a/src/iceberg/catalog/rest/resource_paths.h b/src/iceberg/catalog/rest/resource_paths.h
index 27135bb22..99e748231 100644
--- a/src/iceberg/catalog/rest/resource_paths.h
+++ b/src/iceberg/catalog/rest/resource_paths.h
@@ -49,7 +49,7 @@ class ICEBERG_REST_EXPORT ResourcePaths {
   /// \brief Get the /v1/config endpoint path.
   Result<std::string> Config() const;
 
-  /// \brief Get the /v1/{prefix}/oauth/tokens endpoint path.
+  /// \brief Get the /v1/oauth/tokens endpoint path.
   Result<std::string> OAuth2Tokens() const;
 
   /// \brief Get the /v1/{prefix}/namespaces endpoint path.
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 349071f42..4a4f990ea 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -399,7 +399,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
                           config.Get(RestCatalogProperties::kNamespaceSeparator)));
 
   // Create init session for fetching server configuration
-  HttpClient init_client(config.ExtractHeaders());
+  auto init_client = std::make_shared<HttpClient>(config.ExtractHeaders());
   ICEBERG_ASSIGN_OR_RAISE(auto init_session,
                           auth_manager->InitSession(init_client, config.configs()));
   ICEBERG_ASSIGN_OR_RAISE(auto server_config,
@@ -432,7 +432,7 @@ Result<std::shared_ptr<RestCatalog>> RestCatalog::Make(
 
   auto client = std::make_shared<HttpClient>(final_config.ExtractHeaders());
   ICEBERG_ASSIGN_OR_RAISE(auto catalog_session,
-                          auth_manager->CatalogSession(*client, final_config.configs()));
+                          auth_manager->CatalogSession(client, final_config.configs()));
 
   // Create FileIO with the final configuration
   ICEBERG_ASSIGN_OR_RAISE(auto file_io, MakeCatalogFileIO(final_config));
diff --git a/src/iceberg/test/auth_manager_test.cc b/src/iceberg/test/auth_manager_test.cc
index 19526b7e3..61962d1c3 100644
--- a/src/iceberg/test/auth_manager_test.cc
+++ b/src/iceberg/test/auth_manager_test.cc
@@ -22,6 +22,7 @@
 #include <atomic>
 #include <chrono>
 #include <condition_variable>
+#include <memory>
 #include <mutex>
 #include <string>
 #include <thread>
@@ -35,13 +36,16 @@
 #include "iceberg/catalog/rest/auth/auth_managers.h"
 #include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
+#include "iceberg/catalog/rest/auth/auth_session_internal.h"
 #include "iceberg/catalog/rest/auth/oauth2_util.h"
 #include "iceberg/catalog/rest/auth/token_refresh_scheduler.h"
+#include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/error_handlers.h"
 #include "iceberg/catalog/rest/http_client.h"
 #include "iceberg/catalog/rest/json_serde_internal.h"
 #include "iceberg/catalog/session_context.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/table_identifier.h"
 #include "iceberg/test/matchers.h"
 #include "iceberg/util/base64.h"
 
@@ -66,9 +70,83 @@ std::string MakeJwt(const std::string& payload_json) {
 
 class AuthManagerTest : public ::testing::Test {
  protected:
-  HttpClient client_{{}};
+  std::shared_ptr<HttpClient> client_ = std::make_shared<HttpClient>();
 };
 
+TEST(AuthPropertiesTest, ResolvesDefaultOAuth2ServerUri) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto config,
+      AuthProperties::FromProperties({
+          {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"},
+          {RestCatalogProperties::kPrefix.key(), "warehouse"},
+      }));
+
+  EXPECT_EQ(config.oauth2_server_uri(),
+            "https://catalog.example.com/api/v1/oauth/tokens");
+}
+
+TEST(AuthPropertiesTest, PreservesEmptyOAuth2ServerUri) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto config, AuthProperties::FromProperties({
+                       {RestCatalogProperties::kUri.key(), "https://catalog.example.com"},
+                       {AuthProperties::kOAuth2ServerUri.key(), ""},
+                   }));
+
+  EXPECT_TRUE(config.oauth2_server_uri().empty());
+}
+
+TEST(AuthPropertiesTest, ResolvesExplicitRelativeOAuth2ServerUri) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto config,
+      AuthProperties::FromProperties({
+          {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"},
+          {AuthProperties::kOAuth2ServerUri.key(), "oauth/token/"},
+      }));
+
+  EXPECT_EQ(config.oauth2_server_uri(), "https://catalog.example.com/api/oauth/token/");
+}
+
+TEST(AuthPropertiesTest, PreservesExplicitAbsoluteOAuth2ServerUri) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto config,
+      AuthProperties::FromProperties({
+          {RestCatalogProperties::kUri.key(), "https://catalog.example.com"},
+          {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token/"},
+      }));
+
+  EXPECT_EQ(config.oauth2_server_uri(), "https://auth.example.com/token/");
+}
+
+TEST(AuthPropertiesTest, PreservesRelativeOAuth2ServerUriWithoutCatalogUri) {
+  ICEBERG_UNWRAP_OR_FAIL(auto config,
+                         AuthProperties::FromProperties({
+                             {AuthProperties::kOAuth2ServerUri.key(), "oauth/token"},
+                         }));
+
+  EXPECT_EQ(config.oauth2_server_uri(), "oauth/token");
+}
+
+TEST(AuthPropertiesTest, PreservesAbsolutePathWithoutCatalogUri) {
+  ICEBERG_UNWRAP_OR_FAIL(auto config,
+                         AuthProperties::FromProperties({
+                             {AuthProperties::kOAuth2ServerUri.key(), "/oauth/token"},
+                         }));
+
+  EXPECT_EQ(config.oauth2_server_uri(), "/oauth/token");
+}
+
+TEST(AuthPropertiesTest, ResolvesOAuth2ServerUriWithLeadingSlash) {
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto config,
+      AuthProperties::FromProperties({
+          {RestCatalogProperties::kUri.key(), "https://catalog.example.com/api/"},
+          {AuthProperties::kOAuth2ServerUri.key(), "/v1/oauth/tokens"},
+      }));
+
+  EXPECT_EQ(config.oauth2_server_uri(),
+            "https://catalog.example.com/api/v1/oauth/tokens");
+}
+
 // Verifies loading NoopAuthManager with explicit "none" auth type
 TEST_F(AuthManagerTest, LoadNoopAuthManagerExplicit) {
   std::unordered_map<std::string, std::string> properties = {
@@ -115,11 +193,67 @@ TEST_F(AuthManagerTest, HttpHeadersAreCaseInsensitiveSingleValueMap) {
   EXPECT_EQ(headers.at("AUTHORIZATION"), "Bearer first");
 }
 
+TEST_F(AuthManagerTest, DefaultSessionPreservesRequestAuthorizationHeader) {
+  auto session = AuthSession::MakeDefault(OAuth2Util::AuthHeaders("parent-token"));
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto authenticated,
+      session->Authenticate({.headers = {{"Authorization", "Basic credentials"}}}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"), "Basic credentials");
+  EXPECT_FALSE(session->OAuth2Info().has_value());
+}
+
+TEST_F(AuthManagerTest, OAuth2SessionPreservesRequestAuthorizationHeader) {
+  OAuthTokenResponse token_response{
+      .access_token = "parent-token",
+      .token_type = "bearer",
+  };
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto session,
+      AuthSession::MakeOAuth2(token_response, "https://auth.example.com/token", "", "",
+                              "catalog", /*keep_refreshed=*/false, {}, client_));
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto authenticated,
+      session->Authenticate({.headers = {{"Authorization", "Basic credentials"}}}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"), "Basic credentials");
+
+  auto info = session->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType);
+}
+
+TEST_F(AuthManagerTest, OAuth2SessionExposesMetadata) {
+  OAuthTokenResponse token_response{
+      .access_token = "parent-token",
+      .token_type = "bearer",
+      .issued_token_type = AuthProperties::kJwtTokenType,
+  };
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto session,
+      AuthSession::MakeOAuth2(
+          token_response, "https://auth.example.com/token", "client-id", "client-secret",
+          "catalog", /*keep_refreshed=*/false,
+          {{AuthProperties::kAudience.key(), "catalog-audience"}}, client_));
+
+  auto info = session->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->token, "parent-token");
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kJwtTokenType);
+  EXPECT_EQ(info->credential, "client-id:client-secret");
+  EXPECT_EQ(info->scope, "catalog");
+  EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token");
+  EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()),
+            "catalog-audience");
+}
+
 TEST_F(AuthManagerTest, HttpClientRejectsParamsWhenUrlAlreadyHasQuery) {
   auto session = AuthSession::MakeDefault({});
   auto result =
-      client_.Get("http://127.0.0.1/v1/config?existing=true", {{"warehouse", "prod"}},
-                  /*headers=*/{}, *rest::DefaultErrorHandler::Instance(), *session);
+      client_->Get("http://127.0.0.1/v1/config?existing=true", {{"warehouse", "prod"}},
+                   /*headers=*/{}, *rest::DefaultErrorHandler::Instance(), *session);
 
   EXPECT_THAT(result, IsError(ErrorKind::kInvalidArgument));
   EXPECT_THAT(result, HasErrorMessage("must not contain a query string"));
@@ -240,7 +374,8 @@ TEST_F(AuthManagerTest, RegisterCustomAuthManager) {
         class CustomAuthManager : public AuthManager {
          public:
           Result<std::shared_ptr<AuthSession>> CatalogSession(
-              HttpClient&, const std::unordered_map<std::string, std::string>&) override {
+              std::shared_ptr<HttpClient>,
+              const std::unordered_map<std::string, std::string>&) override {
             return AuthSession::MakeDefault({{"X-Custom-Auth", "custom-value"}});
           }
         };
@@ -266,6 +401,11 @@ TEST_F(AuthManagerTest, OAuth2StaticToken) {
   std::unordered_map<std::string, std::string> properties = {
       {AuthProperties::kAuthType, "oauth2"},
       {AuthProperties::kToken.key(), "my-static-token"},
+      {AuthProperties::kCredential.key(), "client-id:client-secret"},
+      {AuthProperties::kScope.key(), "catalog"},
+      {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"},
+      {AuthProperties::kAudience.key(), "catalog-audience"},
+      {AuthProperties::kResource.key(), "catalog-resource"},
   };
 
   auto manager_result = AuthManagers::Load("test-catalog", properties);
@@ -277,6 +417,18 @@ TEST_F(AuthManagerTest, OAuth2StaticToken) {
   auto auth_result = session_result.value()->Authenticate({});
   ASSERT_THAT(auth_result, IsOk());
   EXPECT_EQ(auth_result.value().headers["Authorization"], "Bearer my-static-token");
+
+  auto info = session_result.value()->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->token, "my-static-token");
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType);
+  EXPECT_EQ(info->credential, "client-id:client-secret");
+  EXPECT_EQ(info->scope, "catalog");
+  EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token");
+  EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()),
+            "catalog-audience");
+  EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kResource.key()),
+            "catalog-resource");
 }
 
 // Verifies OAuth2 type is inferred from token property
@@ -314,6 +466,106 @@ TEST_F(AuthManagerTest, OAuth2MissingCredentials) {
   ASSERT_TRUE(auth_result.has_value());
   EXPECT_EQ(auth_result.value().headers.find("Authorization"),
             auth_result.value().headers.end());
+
+  auto info = session_result.value()->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_TRUE(info->token.empty());
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType);
+}
+
+TEST_F(AuthManagerTest, OAuth2ContextTokenCreatesChildAndHasPriority) {
+  std::unordered_map<std::string, std::string> properties = {
+      {AuthProperties::kAuthType, "oauth2"},
+      {AuthProperties::kScope.key(), "catalog"},
+      {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"},
+      {AuthProperties::kAudience.key(), "catalog-audience"},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties));
+
+  SessionContext context{
+      .session_id = "tenant-a",
+      .credentials =
+          {
+              {AuthProperties::kToken.key(), "context-token"},
+              {AuthProperties::kCredential.key(), "unused-credential"},
+              {AuthProperties::kIdTokenType, "unused-id-token"},
+          },
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent));
+
+  EXPECT_NE(child, parent);
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+  EXPECT_EQ(authenticated.headers.at("Authorization"), "Bearer context-token");
+
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->token, "context-token");
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType);
+  EXPECT_TRUE(info->credential.empty());
+  EXPECT_EQ(info->scope, "catalog");
+  EXPECT_EQ(info->oauth2_server_uri, "https://auth.example.com/token");
+  EXPECT_EQ(info->optional_oauth_params.at(AuthProperties::kAudience.key()),
+            "catalog-audience");
+}
+
+TEST_F(AuthManagerTest, OAuth2ContextTypedTokenOnlyUsesCredentials) {
+  std::unordered_map<std::string, std::string> properties = {
+      {AuthProperties::kAuthType, "oauth2"},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties));
+
+  SessionContext context{
+      .session_id = "tenant-a",
+      .credentials = {{"unrelated", "value"}},
+      .properties = {{AuthProperties::kIdTokenType, "property-id-token"}},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent));
+
+  EXPECT_EQ(child, parent);
+}
+
+TEST_F(AuthManagerTest, OAuth2TableTokenCreatesChild) {
+  std::unordered_map<std::string, std::string> properties = {
+      {AuthProperties::kAuthType, "oauth2"},
+      {AuthProperties::kScope.key(), "catalog"},
+      {AuthProperties::kOAuth2ServerUri.key(), "https://auth.example.com/token"},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties));
+  TableIdentifier table{.ns = Namespace{{"db"}}, .name = "table"};
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto child,
+      manager->TableSession(table,
+                            {{AuthProperties::kToken.key(), "table-token"},
+                             {AuthProperties::kCredential.key(), "ignored-credential"}},
+                            parent));
+
+  EXPECT_NE(child, parent);
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+  EXPECT_EQ(authenticated.headers.at("Authorization"), "Bearer table-token");
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, AuthProperties::kAccessTokenType);
+  EXPECT_TRUE(info->credential.empty());
+}
+
+TEST_F(AuthManagerTest, OAuth2TableIgnoresCredential) {
+  std::unordered_map<std::string, std::string> properties = {
+      {AuthProperties::kAuthType, "oauth2"},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager, AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client_, properties));
+  TableIdentifier table{.ns = Namespace{{"db"}}, .name = "table"};
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto child,
+      manager->TableSession(
+          table, {{AuthProperties::kCredential.key(), "ignored-credential"}}, parent));
+
+  EXPECT_EQ(child, parent);
 }
 
 // Verifies that when both token and credential are provided, token takes priority
@@ -411,12 +663,12 @@ TEST_F(AuthManagerTest, OAuthTokenResponseNATokenType) {
   EXPECT_EQ(result->token_type, "N_A");
 }
 
-// ---- ExpiresAtMillis tests ----
+// ---- OAuth2Util expiry tests ----
 
 TEST_F(AuthManagerTest, ExpiresAtMillisValidJwt) {
   std::string token = MakeJwt(R"({"sub":"user","exp":1700000000})");
 
-  auto result = ExpiresAtMillis(token);
+  auto result = OAuth2Util::ExpiresAtMillis(token);
 
   ASSERT_TRUE(result.has_value());
   EXPECT_EQ(result.value(), 1700000000LL * 1000);
@@ -436,7 +688,7 @@ TEST_F(AuthManagerTest, ExpiresAtMillisInvalidTokensReturnNullopt) {
   };
 
   for (const auto& token : tokens) {
-    EXPECT_FALSE(ExpiresAtMillis(token).has_value()) << token;
+    EXPECT_FALSE(OAuth2Util::ExpiresAtMillis(token).has_value()) << token;
   }
 }
 
@@ -515,10 +767,10 @@ TEST(TokenRefreshSchedulerTest, CancelInvalidHandleIsNoop) {
   scheduler.Shutdown();
 }
 
-// ---- OAuth2AuthSession tests ----
+// ---- OAuth2Session tests ----
 
-TEST(OAuth2AuthSessionTest, InitialTokenIsUsed) {
-  HttpClient client({});
+TEST(OAuth2SessionTest, InitialTokenIsUsed) {
+  auto client = std::make_shared<HttpClient>();
   OAuthTokenResponse token_response;
   token_response.access_token = "initial-token-123";
   token_response.token_type = "bearer";
@@ -539,4 +791,35 @@ TEST(OAuth2AuthSessionTest, InitialTokenIsUsed) {
   session->Close();
 }
 
+TEST(OAuth2SessionTest, InitTokenExpirationUsesRequestStartTime) {
+  OAuthTokenResponse token_response{
+      .access_token = "opaque-token",
+      .token_type = "bearer",
+      .expires_in_secs = 60,
+  };
+  auto request_started_at = std::chrono::steady_clock::time_point{};
+
+  auto expiration = internal::TokenExpirationTime(token_response, request_started_at);
+
+  ASSERT_TRUE(expiration.has_value());
+  EXPECT_EQ(*expiration, request_started_at + std::chrono::seconds(60));
+}
+
+TEST(OAuth2SessionTest, JwtExpirationTakesPriorityOverExpiresIn) {
+  OAuthTokenResponse token_response{
+      .access_token = MakeJwt(R"({"exp":120})"),
+      .token_type = "bearer",
+      .expires_in_secs = 60,
+  };
+  auto request_started_at = std::chrono::steady_clock::time_point{};
+  auto now_system = std::chrono::system_clock::time_point(std::chrono::seconds(100));
+  auto now_steady = std::chrono::steady_clock::time_point(std::chrono::seconds(50));
+
+  auto expiration = internal::TokenExpirationTime(token_response, request_started_at,
+                                                  now_system, now_steady);
+
+  ASSERT_TRUE(expiration.has_value());
+  EXPECT_EQ(*expiration, std::chrono::steady_clock::time_point(std::chrono::seconds(70)));
+}
+
 }  // namespace iceberg::rest::auth
diff --git a/src/iceberg/test/rest_catalog_integration_test.cc b/src/iceberg/test/rest_catalog_integration_test.cc
index 96f392533..25e2b1955 100644
--- a/src/iceberg/test/rest_catalog_integration_test.cc
+++ b/src/iceberg/test/rest_catalog_integration_test.cc
@@ -34,6 +34,8 @@
 #include <nlohmann/json.hpp>
 #include <sys/socket.h>
 
+#include "iceberg/catalog/rest/auth/auth_managers.h"
+#include "iceberg/catalog/rest/auth/auth_properties.h"
 #include "iceberg/catalog/rest/auth/auth_session.h"
 #include "iceberg/catalog/rest/catalog_properties.h"
 #include "iceberg/catalog/rest/error_handlers.h"
@@ -101,6 +103,8 @@ bool CheckServiceReady(uint16_t port) {
 
 std::string CatalogUri() { return std::format("{}:{}", kLocalhostUri, kRestCatalogPort); }
 
+std::string OAuthTokenUri() { return CatalogUri() + "/v1/oauth/tokens"; }
+
 }  // namespace
 
 /// \brief Integration test fixture for REST catalog with Docker Compose.
@@ -211,6 +215,130 @@ TEST_F(RestCatalogIntegrationTest, MakeCatalogSuccess) {
   EXPECT_THAT(root->WithContext(SessionContext{}), IsError(ErrorKind::kInvalidArgument));
 }
 
+TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialEndToEnd) {
+  auto client = std::make_shared<HttpClient>();
+  std::unordered_map<std::string, std::string> properties = {
+      {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2},
+      {auth::AuthProperties::kToken.key(), "catalog-token"},
+      {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager,
+                         auth::AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties));
+  SessionContext context{
+      .session_id = "tenant-context-credential",
+      .credentials = {{auth::AuthProperties::kCredential.key(), "context-client:secret"}},
+  };
+
+  ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent));
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"),
+            "Bearer client-credentials-token:sub=context-client");
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType);
+}
+
+TEST_F(RestCatalogIntegrationTest, OAuthContextCredentialThroughRestCatalog) {
+  auto config = RestCatalogProperties::default_properties();
+  config.Set(RestCatalogProperties::kUri, CatalogUri())
+      .Set(RestCatalogProperties::kName, std::string(kCatalogName))
+      .Set(RestCatalogProperties::kWarehouse, std::string(kWarehouseName));
+  config.mutable_configs()[std::string(RestCatalogProperties::kIOImpl.key())] =
+      std::string(kStdFileIOImpl);
+  config.mutable_configs()[auth::AuthProperties::kAuthType] =
+      auth::AuthProperties::kAuthTypeOAuth2;
+  config.mutable_configs()[auth::AuthProperties::kToken.key()] = "catalog-token";
+  config.mutable_configs()[auth::AuthProperties::kOAuth2ServerUri.key()] =
+      OAuthTokenUri();
+
+  ICEBERG_UNWRAP_OR_FAIL(auto root, RestCatalog::Make(config));
+  SessionContext context{
+      .session_id = "tenant-context-credential",
+      .credentials = {{auth::AuthProperties::kCredential.key(), "context-client:secret"}},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto catalog, root->WithContext(context));
+  ICEBERG_UNWRAP_OR_FAIL(auto namespaces,
+                         catalog->ListNamespaces(Namespace{.levels = {}}));
+
+  EXPECT_TRUE(namespaces.empty());
+}
+
+TEST_F(RestCatalogIntegrationTest, OAuthContextTypedTokenEndToEnd) {
+  auto client = std::make_shared<HttpClient>();
+  std::unordered_map<std::string, std::string> properties = {
+      {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2},
+      {auth::AuthProperties::kToken.key(), "catalog-token"},
+      {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager,
+                         auth::AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties));
+  SessionContext context{
+      .session_id = "tenant-context-token",
+      .credentials = {{auth::AuthProperties::kIdTokenType, "context-id-token"}},
+  };
+
+  ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent));
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"),
+            "Bearer token-exchange-token:sub=context-id-token,act=catalog-token");
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType);
+}
+
+TEST_F(RestCatalogIntegrationTest, OAuthTableTypedTokenEndToEnd) {
+  auto client = std::make_shared<HttpClient>();
+  std::unordered_map<std::string, std::string> properties = {
+      {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2},
+      {auth::AuthProperties::kToken.key(), "catalog-token"},
+      {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager,
+                         auth::AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties));
+  TableIdentifier table{.ns = Namespace{{"db"}}, .name = "events"};
+
+  ICEBERG_UNWRAP_OR_FAIL(
+      auto child,
+      manager->TableSession(
+          table, {{auth::AuthProperties::kJwtTokenType, "table-jwt-token"}}, parent));
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"),
+            "Bearer token-exchange-token:sub=table-jwt-token,act=catalog-token");
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType);
+}
+
+TEST_F(RestCatalogIntegrationTest, OAuthTokenExchangeWithoutActorEndToEnd) {
+  auto client = std::make_shared<HttpClient>();
+  std::unordered_map<std::string, std::string> properties = {
+      {auth::AuthProperties::kAuthType, auth::AuthProperties::kAuthTypeOAuth2},
+      {auth::AuthProperties::kOAuth2ServerUri.key(), OAuthTokenUri()},
+  };
+  ICEBERG_UNWRAP_OR_FAIL(auto manager,
+                         auth::AuthManagers::Load("test-catalog", properties));
+  ICEBERG_UNWRAP_OR_FAIL(auto parent, manager->CatalogSession(client, properties));
+  SessionContext context{
+      .session_id = "tenant-no-actor",
+      .credentials = {{auth::AuthProperties::kIdTokenType, "context-id-token"}},
+  };
+
+  ICEBERG_UNWRAP_OR_FAIL(auto child, manager->ContextualSession(context, parent));
+  ICEBERG_UNWRAP_OR_FAIL(auto authenticated, child->Authenticate({}));
+
+  EXPECT_EQ(authenticated.headers.at("Authorization"),
+            "Bearer token-exchange-token:sub=context-id-token");
+  auto info = child->OAuth2Info();
+  ASSERT_TRUE(info.has_value());
+  EXPECT_EQ(info->issued_token_type, auth::AuthProperties::kAccessTokenType);
+}
+
 TEST_F(RestCatalogIntegrationTest, LoadsConfiguredMetricsReporter) {
   auto loaded = std::make_shared<std::atomic<bool>>(false);
   ASSERT_THAT(MetricsReporters::Register(
diff --git a/src/iceberg/test/sigv4_auth_test.cc b/src/iceberg/test/sigv4_auth_test.cc
index 12d18792f..6dad2d2ff 100644
--- a/src/iceberg/test/sigv4_auth_test.cc
+++ b/src/iceberg/test/sigv4_auth_test.cc
@@ -21,6 +21,7 @@
 
 #  include <algorithm>
 #  include <cctype>
+#  include <memory>
 #  include <string>
 #  include <string_view>
 #  include <unordered_map>
@@ -201,7 +202,7 @@ class SigV4AuthTest : public ::testing::Test {
     return session->Authenticate(std::move(request));
   }
 
-  HttpClient client_{{}};
+  std::shared_ptr<HttpClient> client_ = std::make_shared<HttpClient>();
 };
 
 TEST_F(SigV4AuthTest, LifecycleInitializeIsIdempotent) {