diff --git a/src/iceberg/catalog/rest/json_serde.cc b/src/iceberg/catalog/rest/json_serde.cc
index 3ce753f18..098f2ba08 100644
--- a/src/iceberg/catalog/rest/json_serde.cc
+++ b/src/iceberg/catalog/rest/json_serde.cc
@@ -152,6 +152,25 @@ Result<StorageCredential> StorageCredentialFromJson(const nlohmann::json& json)
   return credential;
 }
 
+/// \brief Reads the optional `storage-credentials` array shared by the
+/// LoadTable and LoadCredentials responses.
+Result<std::vector<StorageCredential>> StorageCredentialsFromJson(
+    const nlohmann::json& json) {
+  std::vector<StorageCredential> credentials;
+  auto it = json.find(kStorageCredentials);
+  if (it == json.end() || it->is_null()) {
+    return credentials;
+  }
+  if (!it->is_array()) {
+    return JsonParseError("Cannot parse storage credentials from non-array");
+  }
+  for (const auto& entry : *it) {
+    ICEBERG_ASSIGN_OR_RAISE(auto credential, StorageCredentialFromJson(entry));
+    credentials.push_back(std::move(credential));
+  }
+  return credentials;
+}
+
 template <typename Value>
 Result<std::map<int32_t, Value>> KeyValueMapFromJson(const nlohmann::json& json,
                                                      std::string_view key) {
@@ -738,19 +757,24 @@ Result<LoadTableResult> LoadTableResultFromJson(const nlohmann::json& json) {
   ICEBERG_ASSIGN_OR_RAISE(result.metadata, TableMetadataFromJson(metadata_json));
   ICEBERG_ASSIGN_OR_RAISE(result.config,
                           GetJsonValueOrDefault<decltype(result.config)>(json, kConfig));
-  if (auto it = json.find(kStorageCredentials); it != json.end() && !it->is_null()) {
-    if (!it->is_array()) {
-      return JsonParseError("Cannot parse storage credentials from non-array");
-    }
-    for (const auto& entry : *it) {
-      ICEBERG_ASSIGN_OR_RAISE(auto cred, StorageCredentialFromJson(entry));
-      result.storage_credentials.push_back(std::move(cred));
-    }
-  }
+  ICEBERG_ASSIGN_OR_RAISE(result.storage_credentials, StorageCredentialsFromJson(json));
   ICEBERG_RETURN_UNEXPECTED(result.Validate());
   return result;
 }
 
+Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json) {
+  // Required here, unlike in LoadTable: reading a malformed response as "no
+  // credentials" would look like a refresh that succeeded and dropped them.
+  if (auto it = json.find(kStorageCredentials); it == json.end() || it->is_null()) {
+    return JsonParseError("Missing '{}'", kStorageCredentials);
+  }
+  LoadCredentialsResponse response;
+  ICEBERG_ASSIGN_OR_RAISE(response.storage_credentials, StorageCredentialsFromJson(json));
+  ICEBERG_RETURN_UNEXPECTED(response.Validate());
+  return response;
+}
+
 nlohmann::json ToJson(const ListNamespacesResponse& response) {
   nlohmann::json json;
   SetOptionalStringField(json, kNextPageToken, response.next_page_token);
diff --git a/src/iceberg/catalog/rest/json_serde_internal.h b/src/iceberg/catalog/rest/json_serde_internal.h
index 6e148e0d3..aedd2cd1a 100644
--- a/src/iceberg/catalog/rest/json_serde_internal.h
+++ b/src/iceberg/catalog/rest/json_serde_internal.h
@@ -71,6 +71,10 @@ template <>
 ICEBERG_REST_EXPORT Result<LoadTableResult> FromJson(const nlohmann::json& json);
 ICEBERG_REST_EXPORT Result<nlohmann::json> ToJson(const LoadTableResult& model);
 
+// Response-only model: a client never serializes it, so no ToJson.
+ICEBERG_REST_EXPORT Result<LoadCredentialsResponse> LoadCredentialsResponseFromJson(
+    const nlohmann::json& json);
+
 ICEBERG_REST_EXPORT Result<CreateTableRequest> CreateTableRequestFromJson(
     const nlohmann::json& json);
 template <>
diff --git a/src/iceberg/catalog/rest/rest_catalog.cc b/src/iceberg/catalog/rest/rest_catalog.cc
index 9635aebef..91c19d7b8 100644
--- a/src/iceberg/catalog/rest/rest_catalog.cc
+++ b/src/iceberg/catalog/rest/rest_catalog.cc
@@ -42,6 +42,7 @@
 #include "iceberg/catalog/rest/rest_util.h"
 #include "iceberg/catalog/rest/types.h"
 #include "iceberg/json_serde_internal.h"
+#include "iceberg/logging/log_macros.h"
 #include "iceberg/metrics/metrics_reporters.h"
 #include "iceberg/partition_spec.h"
 #include "iceberg/result.h"
@@ -508,12 +509,65 @@ Result<std::shared_ptr<auth::AuthSession>> RestCatalog::TableAuthSession(
                                      std::move(contextual_session));
 }
 
+StorageCredentialRefresher RestCatalog::MakeCredentialRefresher(
+    const TableIdentifier& identifier,
+    std::shared_ptr<auth::AuthSession> table_session) const {
+  if (!supported_endpoints_.contains(Endpoint::TableCredentials())) {
+    // Not an error, but it surfaces much later as credentials expiring.
+    ICEBERG_LOG_DEBUG(
+        "Catalog does not advertise {}; vended credentials for '{}' will not be "
+        "refreshed",
+        Endpoint::TableCredentials().ToString(), ToString(identifier));
+    return nullptr;
+  }
+  auto path = paths_->Credentials(identifier);
+  if (!path.has_value()) {
+    ICEBERG_LOG_WARN(
+        "Cannot build the credentials path for '{}' ({}); its vended credentials "
+        "will not be refreshed",
+        ToString(identifier), path.error().message);
+    return nullptr;
+  }
+  auto client = client_;
+  auto credentials_path = std::move(path.value());
+  auto session = std::move(table_session);
+  // The catalog's destructor closes the session, and a table's FileIO can
+  // outlive the table keeping the catalog alive. No cycle: the catalog's own
+  // FileIO never gets a refresher.
+  auto catalog = shared_from_this();
+  return [catalog, client, credentials_path,
+          session]() -> Result<std::vector<StorageCredential>> {
+    ICEBERG_ASSIGN_OR_RAISE(const auto response,
+                            client->Get(credentials_path, /*params=*/{}, /*headers=*/{},
+                                        *TableErrorHandler::Instance(), *session));
+    // Parse errors embed the offending input, and this body carries
+    // credentials; strip the message so it can never reach a log.
+    auto json = FromJsonString(response.body());
+    if (!json.has_value()) {
+      return JsonParseError("Malformed LoadCredentials response");
+    }
+    auto result = LoadCredentialsResponseFromJson(*json);
+    if (!result.has_value()) {
+      return std::unexpected<Error>(
+          {.kind = result.error().kind, .message = "Malformed LoadCredentials response"});
+    }
+    return std::move(result->storage_credentials);
+  };
+}
+
 Result<std::shared_ptr<FileIO>> RestCatalog::TableFileIO(
-    const SessionContext& /*context*/,
+    const SessionContext& /*context*/, const TableIdentifier& identifier,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) const {
+    const std::vector<StorageCredential>& storage_credentials,
+    std::shared_ptr<auth::AuthSession> table_session) const {
   if (!table_config.empty() || !storage_credentials.empty()) {
-    return MakeTableFileIO(config_.configs(), table_config, storage_credentials);
+    // Only vended credentials expire, so only they need a refresher.
+    StorageCredentialRefresher refresher;
+    if (!storage_credentials.empty()) {
+      refresher = MakeCredentialRefresher(identifier, std::move(table_session));
+    }
+    return MakeTableFileIO(config_.configs(), table_config, storage_credentials,
+                           std::move(refresher));
   }
 
   return file_io_;
@@ -772,11 +826,12 @@ Result<std::shared_ptr<Transaction>> RestCatalog::StageCreateTable(
                           /*stage_create=*/true, *contextual_session));
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, std::move(table_session),
@@ -890,11 +945,12 @@ Result<std::shared_ptr<Table>> RestCatalog::MakeTableFromLoadResult(
     std::shared_ptr<auth::AuthSession> contextual_session) {
   auto table_config = std::move(result.config);
   auto storage_credentials = std::move(result.storage_credentials);
-  ICEBERG_ASSIGN_OR_RAISE(auto table_io,
-                          TableFileIO(context, table_config, storage_credentials));
+  // Before the FileIO: refreshing its credentials reuses the table session.
   ICEBERG_ASSIGN_OR_RAISE(
       auto table_session,
       TableAuthSession(identifier, table_config, std::move(contextual_session)));
+  ICEBERG_ASSIGN_OR_RAISE(auto table_io, TableFileIO(context, identifier, table_config,
+                                                     storage_credentials, table_session));
   ICEBERG_ASSIGN_OR_RAISE(auto reporter, MakeTableReporter(identifier, table_session));
   auto table_catalog = std::make_shared<TableScopedCatalog>(
       shared_from_this(), context, identifier, table_config, table_session, table_io);
diff --git a/src/iceberg/catalog/rest/rest_catalog.h b/src/iceberg/catalog/rest/rest_catalog.h
index 65b0b5eab..bcb1be53a 100644
--- a/src/iceberg/catalog/rest/rest_catalog.h
+++ b/src/iceberg/catalog/rest/rest_catalog.h
@@ -85,9 +85,19 @@ class ICEBERG_REST_EXPORT RestCatalog final
       std::shared_ptr<auth::AuthSession> contextual_session);
 
   Result<std::shared_ptr<FileIO>> TableFileIO(
-      const SessionContext& context,
+      const SessionContext& context, const TableIdentifier& identifier,
       const std::unordered_map<std::string, std::string>& table_config,
-      const std::vector<StorageCredential>& storage_credentials) const;
+      const std::vector<StorageCredential>& storage_credentials,
+      std::shared_ptr<auth::AuthSession> table_session) const;
+
+  /// \brief Callback that reloads this table's vended credentials, or nullptr
+  /// when the catalog does not serve the LoadCredentials endpoint.
+  ///
+  /// Captures the catalog, so anything documented to outlive it -- the metrics
+  /// executor included -- must also outlive every FileIO holding a refresher.
+  StorageCredentialRefresher MakeCredentialRefresher(
+      const TableIdentifier& identifier,
+      std::shared_ptr<auth::AuthSession> table_session) const;
 
   Result<std::vector<Namespace>> ListNamespaces(const Namespace& ns,
                                                 auth::AuthSession& session) const;
diff --git a/src/iceberg/catalog/rest/rest_file_io.cc b/src/iceberg/catalog/rest/rest_file_io.cc
index cdc1204f1..2531008f3 100644
--- a/src/iceberg/catalog/rest/rest_file_io.cc
+++ b/src/iceberg/catalog/rest/rest_file_io.cc
@@ -21,6 +21,7 @@
 
 #include <string>
 #include <unordered_map>
+#include <utility>
 #include <vector>
 
 #include "iceberg/catalog/rest/types.h"
@@ -57,7 +58,8 @@ Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(const RestCatalogProperties& c
 Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials) {
+    const std::vector<StorageCredential>& storage_credentials,
+    StorageCredentialRefresher refresher) {
   const auto default_properties = MergeFileIOProperties(catalog_config, table_config);
   ICEBERG_ASSIGN_OR_RAISE(
       auto io, MakeCatalogFileIO(RestCatalogProperties::FromMap(default_properties)));
@@ -65,6 +67,10 @@ Result<std::unique_ptr<FileIO>> MakeTableFileIO(
   if (storage_credentials.empty()) {
     return io;
   } else if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+    // First, so the FileIO never briefly holds credentials it cannot replace.
+    if (refresher) {
+      credentialed->SetCredentialRefresher(std::move(refresher));
+    }
     ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(storage_credentials));
   } else {
     return NotSupported("Configured FileIO does not support vended storage credentials");
diff --git a/src/iceberg/catalog/rest/rest_file_io.h b/src/iceberg/catalog/rest/rest_file_io.h
index e2316c3e8..301bf62ff 100644
--- a/src/iceberg/catalog/rest/rest_file_io.h
+++ b/src/iceberg/catalog/rest/rest_file_io.h
@@ -41,9 +41,13 @@ ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeCatalogFileIO(
     const RestCatalogProperties& config);
 
 /// \brief Build the configured table FileIO and apply storage credentials if present.
+///
+/// \param refresher Optional callback used to replace the vended credentials
+/// before they expire; ignored when the FileIO cannot tell when they do.
 ICEBERG_REST_EXPORT Result<std::unique_ptr<FileIO>> MakeTableFileIO(
     const std::unordered_map<std::string, std::string>& catalog_config,
     const std::unordered_map<std::string, std::string>& table_config,
-    const std::vector<StorageCredential>& storage_credentials);
+    const std::vector<StorageCredential>& storage_credentials,
+    StorageCredentialRefresher refresher = nullptr);
 
 }  // namespace iceberg::rest
diff --git a/src/iceberg/catalog/rest/types.h b/src/iceberg/catalog/rest/types.h
index 20a59fa59..a47a8e851 100644
--- a/src/iceberg/catalog/rest/types.h
+++ b/src/iceberg/catalog/rest/types.h
@@ -209,6 +209,21 @@ using CreateTableResponse = LoadTableResult;
 /// \brief Alias of LoadTableResult used as the body of LoadTableResponse
 using LoadTableResponse = LoadTableResult;
 
+/// \brief Response body of the LoadCredentials API.
+struct ICEBERG_REST_EXPORT LoadCredentialsResponse {
+  std::vector<StorageCredential> storage_credentials;
+
+  /// \brief Validates the LoadCredentialsResponse.
+  Status Validate() const {
+    for (const auto& credential : storage_credentials) {
+      ICEBERG_RETURN_UNEXPECTED(credential.Validate());
+    }
+    return {};
+  }
+
+  bool operator==(const LoadCredentialsResponse& other) const = default;
+};
+
 /// \brief Response body for listing namespaces.
 struct ICEBERG_REST_EXPORT ListNamespacesResponse {
   PageToken next_page_token;
diff --git a/src/iceberg/file_io.h b/src/iceberg/file_io.h
index e22e5cf21..86ba316b9 100644
--- a/src/iceberg/file_io.h
+++ b/src/iceberg/file_io.h
@@ -198,6 +198,11 @@ class ICEBERG_EXPORT SupportsStorageCredentials {
   /// By value because a concurrent install may replace them. An implementation
   /// that delegates may report what was installed on it.
   virtual std::vector<StorageCredential> credentials() const = 0;
+
+  /// \brief Install a callback that re-fetches credentials before they expire.
+  ///
+  /// Ignored by implementations that cannot tell when theirs expire.
+  virtual void SetCredentialRefresher(StorageCredentialRefresher /*refresher*/) {}
 };
 
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.cc b/src/iceberg/resolving_file_io.cc
index 91be67eaf..4ba7bafe8 100644
--- a/src/iceberg/resolving_file_io.cc
+++ b/src/iceberg/resolving_file_io.cc
@@ -44,12 +44,17 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   // without static keys may wait on the EC2 metadata service), which would
   // stall every other operation. Forwards all credentials; each implementation
   // applies the prefixes it understands.
-  auto load = [&](const std::vector<StorageCredential>& credentials)
+  auto load = [&](const std::vector<StorageCredential>& credentials,
+                  const StorageCredentialRefresher& refresher)
       -> Result<std::shared_ptr<FileIO>> {
     ICEBERG_ASSIGN_OR_RAISE(std::shared_ptr<FileIO> io,
                             FileIORegistry::Load(name, properties_));
-    if (!credentials.empty()) {
-      if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+    if (auto* credentialed = io->AsSupportsStorageCredentials()) {
+      // Before the credentials, so the delegate can always replace them.
+      if (refresher) {
+        credentialed->SetCredentialRefresher(refresher);
+      }
+      if (!credentials.empty()) {
         ICEBERG_RETURN_UNEXPECTED(credentialed->SetStorageCredentials(credentials));
       }
     }
@@ -59,6 +64,7 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
   while (true) {
     uint64_t generation = 0;
     std::vector<StorageCredential> credentials;
+    StorageCredentialRefresher refresher;
     {
       std::shared_lock lock(mutex_);
       if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
@@ -66,13 +72,14 @@ Result<std::shared_ptr<FileIO>> ResolvingFileIO::FileIOForPath(
       }
       generation = credential_generation_;
       credentials = storage_credentials_;
+      refresher = refresher_;
     }
     // Declared before the lock, so a delegate that is not cached is torn down
     // only after the lock is released.
-    auto loaded = load(credentials);
+    auto loaded = load(credentials, refresher);
     std::unique_lock lock(mutex_);
     if (generation != credential_generation_) {
-      continue;  // Credentials were replaced mid-load; load again with them.
+      continue;  // Replaced mid-load; load again with what is installed now.
     }
     if (const auto cached = io_by_name_.find(name); cached != io_by_name_.end()) {
       return cached->second;
@@ -137,4 +144,21 @@ std::vector<StorageCredential> ResolvingFileIO::credentials() const {
   return storage_credentials_;
 }
 
+void ResolvingFileIO::SetCredentialRefresher(StorageCredentialRefresher refresher) {
+  // Drop the cached delegates so they are rebuilt with the refresher. Retired
+  // outside the lock: teardown can block, and the outgoing callback's captures
+  // must not destruct under `mutex_`.
+  decltype(io_by_name_) retired;
+  // Holds the incoming callback going in and the outgoing one coming out; a
+  // pure swap never destroys a target under the lock, which std::exchange's
+  // move is permitted to do.
+  StorageCredentialRefresher handoff = std::move(refresher);
+  {
+    std::unique_lock lock(mutex_);
+    refresher_.swap(handoff);
+    ++credential_generation_;
+    retired.swap(io_by_name_);
+  }
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/resolving_file_io.h b/src/iceberg/resolving_file_io.h
index 9a5cf047e..c192f8ada 100644
--- a/src/iceberg/resolving_file_io.h
+++ b/src/iceberg/resolving_file_io.h
@@ -40,8 +40,8 @@ namespace iceberg {
 
 /// \brief FileIO that resolves and caches implementations by registry name.
 ///
-/// Vended credentials are forwarded to every resolved implementation that
-/// supports them; each applies what it understands.
+/// Vended credentials and any refresher are forwarded to every resolved
+/// implementation that supports them; each applies what it understands.
 class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
                                              public SupportsStorageCredentials {
  public:
@@ -62,8 +62,18 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   Status SetStorageCredentials(
       const std::vector<StorageCredential>& storage_credentials) override;
 
+  /// \brief Return the credentials installed on this resolver.
+  ///
+  /// Not necessarily the ones in use: a resolved implementation refreshes its
+  /// own without reporting back.
   std::vector<StorageCredential> credentials() const override;
 
+  /// \brief Install a refresher and rebuild cached delegates with it.
+  ///
+  /// A delegate dropped here is rebuilt from the credentials installed on this
+  /// resolver, which may be older than what it had refreshed itself to.
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override;
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
  private:
@@ -74,8 +84,9 @@ class ICEBERG_EXPORT ResolvingFileIO final : public FileIO,
   // Guards lazy resolution and credential state.
   mutable std::shared_mutex mutex_;
   std::vector<StorageCredential> storage_credentials_;
-  // Bumped by every credential install, so a delegate loaded from an older set
-  // never reaches the cache.
+  StorageCredentialRefresher refresher_;
+  // Bumped by every credential or refresher install, so a delegate loaded from
+  // an older state never reaches the cache.
   uint64_t credential_generation_ = 0;
   std::unordered_map<std::string, std::shared_ptr<FileIO>, StringHash, StringEqual>
       io_by_name_;
diff --git a/src/iceberg/storage_credential.h b/src/iceberg/storage_credential.h
index 746bad086..e5d8022d7 100644
--- a/src/iceberg/storage_credential.h
+++ b/src/iceberg/storage_credential.h
@@ -22,8 +22,10 @@
 /// \file iceberg/storage_credential.h
 /// \brief Define storage credential metadata.
 
+#include <functional>
 #include <string>
 #include <unordered_map>
+#include <vector>
 
 #include "iceberg/iceberg_export.h"
 #include "iceberg/result.h"
@@ -48,4 +50,9 @@ struct ICEBERG_EXPORT StorageCredential {
   bool operator==(const StorageCredential& other) const = default;
 };
 
+/// \brief Re-fetches the currently valid vended credentials, without the
+/// FileIO knowing how they are delivered. Returns the whole vended list.
+using StorageCredentialRefresher =
+    std::function<Result<std::vector<StorageCredential>>()>;
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/resolving_file_io_test.cc b/src/iceberg/test/resolving_file_io_test.cc
index 2189b64a6..db1c0d6b0 100644
--- a/src/iceberg/test/resolving_file_io_test.cc
+++ b/src/iceberg/test/resolving_file_io_test.cc
@@ -67,10 +67,25 @@ class RecordingCredentialedFileIO : public RecordingFileIO,
 
   std::vector<StorageCredential> credentials() const override { return credentials_; }
 
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
+    refresher_ = std::move(refresher);
+  }
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 
+  Status Refresh() {
+    if (!refresher_) {
+      return NotFound("no refresher installed");
+    }
+    ICEBERG_ASSIGN_OR_RAISE(auto refreshed, refresher_());
+    return SetStorageCredentials(refreshed);
+  }
+
+  bool has_refresher() const { return static_cast<bool>(refresher_); }
+
  private:
   std::vector<StorageCredential> credentials_;
+  StorageCredentialRefresher refresher_;
 };
 
 // File-scope recording state: registry factories are process-global, so they
@@ -318,4 +333,41 @@ TEST(ResolvingFileIOTest, LoadsWithoutTheLockAndDropsStaleDelegates) {
   EXPECT_EQ(last->credentials(), fresh);
 }
 
+TEST(ResolvingFileIOTest, ForwardsCredentialRefresherToResolvedImplementations) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k2", "v2"}}}};
+  io.SetCredentialRefresher(
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+              IsOk());
+
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  ASSERT_TRUE(last_s3_io->has_refresher());
+  EXPECT_THAT(last_s3_io->Refresh(), IsOk());
+  EXPECT_EQ(last_s3_io->credentials(), refreshed);
+}
+
+TEST(ResolvingFileIOTest, RebuildsResolvedImplementationsForALaterRefresher) {
+  RegisterRecordingFileIOs();
+  ResolvingFileIO io({});
+
+  EXPECT_THAT(io.SetStorageCredentials({{.prefix = "s3", .config = {{"k1", "v1"}}}}),
+              IsOk());
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/file.parquet");
+  ASSERT_NE(last_s3_io, nullptr);
+  EXPECT_FALSE(last_s3_io->has_refresher());
+  EXPECT_EQ(s3_factory_calls, 1);
+
+  io.SetCredentialRefresher([]() -> Result<std::vector<StorageCredential>> {
+    return std::vector<StorageCredential>{};
+  });
+  std::ignore = io.NewInputFile("s3://bucket/db/table/data/other.parquet");
+  EXPECT_EQ(s3_factory_calls, 2);
+  ASSERT_NE(last_s3_io, nullptr);
+  EXPECT_TRUE(last_s3_io->has_refresher());
+}
+
 }  // namespace iceberg
diff --git a/src/iceberg/test/rest_file_io_test.cc b/src/iceberg/test/rest_file_io_test.cc
index 2fce9dacf..ed03f45d2 100644
--- a/src/iceberg/test/rest_file_io_test.cc
+++ b/src/iceberg/test/rest_file_io_test.cc
@@ -52,6 +52,7 @@ class MockFileIO : public FileIO {
 
 std::vector<StorageCredential> captured_storage_credentials;
 std::unordered_map<std::string, std::string> captured_file_io_properties;
+StorageCredentialRefresher captured_refresher;
 
 class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredentials {
  public:
@@ -65,6 +66,10 @@ class MockCredentialedFileIO : public MockFileIO, public SupportsStorageCredenti
     return captured_storage_credentials;
   }
 
+  void SetCredentialRefresher(StorageCredentialRefresher refresher) override {
+    captured_refresher = std::move(refresher);
+  }
+
   SupportsStorageCredentials* AsSupportsStorageCredentials() override { return this; }
 };
 
@@ -155,6 +160,35 @@ TEST(RestFileIOTest, TableFileIOMergesConfigAndCredentials) {
   EXPECT_EQ(credentialed->credentials(), captured_storage_credentials);
 }
 
+TEST(RestFileIOTest, InstallsCredentialRefresherOnlyForVendedCredentials) {
+  const std::string custom_impl = "rest-file-io-test-refresher";
+  FileIORegistry::Register(
+      custom_impl,
+      {.create = [](const std::unordered_map<std::string, std::string>& /*properties*/)
+           -> Result<std::unique_ptr<FileIO>> {
+        return std::make_unique<MockCredentialedFileIO>();
+      }});
+  const std::unordered_map<std::string, std::string> table_config{
+      {"io-impl", custom_impl}};
+
+  captured_refresher = nullptr;
+  std::vector<StorageCredential> refreshed = {{.prefix = "s3", .config = {{"k", "v2"}}}};
+  auto result = MakeTableFileIO(
+      {}, table_config, {{.prefix = "s3", .config = {{"k", "v1"}}}},
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  ASSERT_THAT(result, IsOk());
+  ASSERT_TRUE(captured_refresher);
+  EXPECT_THAT(captured_refresher(), IsOk());
+  EXPECT_EQ(captured_refresher().value(), refreshed);
+
+  captured_refresher = nullptr;
+  result = MakeTableFileIO(
+      {}, table_config, /*storage_credentials=*/{},
+      [&]() -> Result<std::vector<StorageCredential>> { return refreshed; });
+  ASSERT_THAT(result, IsOk());
+  EXPECT_FALSE(captured_refresher);
+}
+
 TEST(RestFileIOTest, TableImplOverridesWarehouseScheme) {
   captured_file_io_properties.clear();
   FileIORegistry::Register(
diff --git a/src/iceberg/test/rest_json_serde_test.cc b/src/iceberg/test/rest_json_serde_test.cc
index ec41e4a66..70b760b48 100644
--- a/src/iceberg/test/rest_json_serde_test.cc
+++ b/src/iceberg/test/rest_json_serde_test.cc
@@ -1229,6 +1229,32 @@ INSTANTIATE_TEST_SUITE_P(
       return info.param.test_name;
     });
 
+TEST(LoadCredentialsResponseTest, ParsesVendedCredentials) {
+  auto json = nlohmann::json::parse(
+      R"({"storage-credentials":[{"prefix":"s3","config":{"s3.access-key-id":"AKIAtest"}}]})");
+  auto response = LoadCredentialsResponseFromJson(json);
+  ASSERT_THAT(response, IsOk());
+  EXPECT_EQ(response->storage_credentials,
+            (std::vector<StorageCredential>{
+                {.prefix = "s3", .config = {{"s3.access-key-id", "AKIAtest"}}}}));
+}
+
+TEST(LoadCredentialsResponseTest, RejectsResponseWithoutCredentials) {
+  for (std::string_view body : {"{}", R"({"storage-credentials": null})"}) {
+    auto response = LoadCredentialsResponseFromJson(nlohmann::json::parse(body));
+    EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+    EXPECT_THAT(response, HasErrorMessage("Missing 'storage-credentials'"));
+  }
+}
+
+TEST(LoadCredentialsResponseTest, RejectsNonArrayCredentials) {
+  auto response = LoadCredentialsResponseFromJson(
+      nlohmann::json::parse(R"({"storage-credentials":"oops"})"));
+  EXPECT_THAT(response, IsError(ErrorKind::kJsonParseError));
+  EXPECT_THAT(response,
+              HasErrorMessage("Cannot parse storage credentials from non-array"));
+}
+
 DECLARE_ROUNDTRIP_TEST(CommitTableRequest)
 
 INSTANTIATE_TEST_SUITE_P(