diff --git a/lib/ldclient-rb/impl/file_data/reloader.rb b/lib/ldclient-rb/impl/file_data/reloader.rb
index 224f5b9f..132d0d33 100644
--- a/lib/ldclient-rb/impl/file_data/reloader.rb
+++ b/lib/ldclient-rb/impl/file_data/reloader.rb
@@ -3,7 +3,6 @@
 require "ldclient-rb/impl/file_data/document"
 require "ldclient-rb/impl/file_data/merge"
 require "ldclient-rb/impl/file_data/poller"
-require "ldclient-rb/impl/util"
 
 require "digest"
 
@@ -97,6 +96,7 @@ class Reloader
         #   is parsed as it is.
         # @param skip_unchanged [Boolean] if true, `apply` is not invoked when the files' raw
         #   contents are byte-identical to the last successfully applied contents.
+        # @param log_prefix [String] the prefix of every log line this reloader writes
         # @param clock [#call, nil] returns the seconds of a monotonic clock, for the time a file has
         #   been held back; nil for the system's. Scheduling always uses the system's clock.
         #
@@ -106,9 +106,11 @@ def initialize(paths:, logger:, apply:, on_error: nil,
                        debounce_delay: DEFAULT_DEBOUNCE_DELAY,
                        retry_delay: DEFAULT_RETRY_DELAY,
                        skip_unchanged: false,
+                       log_prefix: "[LDClient]",
                        clock: nil)
           @paths = paths
           @logger = logger
+          @log_prefix = log_prefix
           @apply = apply
           @on_error = on_error
           @duplicate_keys_handling = duplicate_keys_handling
@@ -221,7 +223,7 @@ def stop
 
             # An unexpected error must not end the worker: no reload would happen after it.
             begin
-              @logger.info { "[LDClient] Reloading flag data after detecting a change" } if action == :reload
+              @logger.info { "#{@log_prefix} Reloading flag data after detecting a change" } if action == :reload
               retry_in = reload(retrying: action == :retry)
               @mutex.synchronize do
                 # A pending retry is superseded by this reload. The reload either succeeded, or it
@@ -240,7 +242,8 @@ def stop
         end
 
         private def log_unexpected_error(error)
-          Util.log_exception(@logger, "Unexpected error in file data reloader", error)
+          @logger.error { "#{@log_prefix} Unexpected error in file data reloader: #{error.inspect}" }
+          @logger.debug { "#{@log_prefix} Exception trace: #{error.backtrace}" }
         end
 
         #
@@ -296,9 +299,9 @@ def stop
 
             if retrying
               if @read_again_pending
-                @logger.debug { "[LDClient] Reading flag data again after a file was found empty" }
+                @logger.debug { "#{@log_prefix} Reading flag data again after a file was found empty" }
               else
-                @logger.debug { "[LDClient] Retrying flag data load after earlier failure" }
+                @logger.debug { "#{@log_prefix} Retrying flag data load after earlier failure" }
               end
             end
             @read_again_pending = false
@@ -349,7 +352,7 @@ def stop
                 reached << path
               end
               if e.missing && @skip_missing_paths
-                @logger.debug { "[LDClient] File #{path} does not exist; it contributes no data" }
+                @logger.debug { "#{@log_prefix} File #{path} does not exist; it contributes no data" }
                 had_content[path] = false
                 append_to_digest(digest, path, nil)
                 files << FileSummary.new(path, false, 0, 0)
@@ -520,7 +523,7 @@ def stop
           retry_at = nil
           paths.each do |path|
             @logger.debug do
-              "[LDClient] File #{path} is empty, possibly while it is being rewritten; reading it again before applying it"
+              "#{@log_prefix} File #{path} is empty, possibly while it is being rewritten; reading it again before applying it"
             end
             empty = @empty_files[path]
             over = [empty.since + @retry_delay, empty.first_held + MAX_HOLD_RETRY_DELAYS * @retry_delay].min
@@ -539,12 +542,12 @@ def stop
           # level and do not re-invoke on_error.
           message = error.message
           if message == @last_error_message
-            @logger.debug { "[LDClient] Unable to load flags: #{message}" }
+            @logger.debug { "#{@log_prefix} Unable to load flags: #{message}" }
             return @retry_delay
           end
 
           @last_error_message = message
-          @logger.error { "[LDClient] Unable to load flags: #{message}" }
+          @logger.error { "#{@log_prefix} Unable to load flags: #{message}" }
           @on_error&.call(error)
           @retry_delay
         end
diff --git a/lib/ldclient-rb/impl/integrations/file_override_source.rb b/lib/ldclient-rb/impl/integrations/file_override_source.rb
new file mode 100644
index 00000000..6320d619
--- /dev/null
+++ b/lib/ldclient-rb/impl/integrations/file_override_source.rb
@@ -0,0 +1,151 @@
+# frozen_string_literal: true
+
+require "ldclient-rb/impl/file_data"
+require "ldclient-rb/interfaces/overrides"
+
+module LaunchDarkly
+  module Impl
+    module Integrations
+      #
+      # The file-based override source. It reads flag and segment overrides from one or more local
+      # files and reloads them as the files change. See
+      # {LaunchDarkly::Integrations::FileData.override_source} for the public API and its options.
+      #
+      # Flag overrides are currently experimental and subject to change.
+      #
+      # @private
+      #
+      class FileOverrideSource
+        include LaunchDarkly::Interfaces::Overrides::OverrideSource
+
+        LOG_PREFIX = "[LDClient] FileOverrideSource:"
+
+        #
+        # @param paths [Array<String>] absolute file paths, in precedence order
+        # @param duplicate_keys_handling [Symbol] `:fail` or `:ignore`
+        # @param change_detection [Symbol] `:polling` or `:watching`
+        # @param poll_interval [Numeric] seconds between examinations in polling mode
+        # @param logger [Logger]
+        #
+        def initialize(paths:, duplicate_keys_handling:, change_detection:, poll_interval:, logger:)
+          @paths = paths
+          @duplicate_keys_handling = duplicate_keys_handling
+          @change_detection = change_detection
+          @poll_interval = poll_interval
+          @logger = logger
+          @lock = Mutex.new
+          @reloader = nil
+          @change_detector = nil
+          @stopped = false
+        end
+
+        # @return [Array<String>]
+        attr_reader :paths
+
+        # @return [Symbol]
+        attr_reader :duplicate_keys_handling
+
+        # @return [Symbol]
+        attr_reader :change_detection
+
+        # @return [Numeric]
+        attr_reader :poll_interval
+
+        #
+        # Performs the initial load synchronously, so overrides present in the files are in effect
+        # when this method returns, then starts change detection. A file that does not exist yet
+        # contributes no overrides. A file that cannot be read or parsed is not fatal: the client
+        # runs with the last good overrides, the failure is logged, and the retry plus the change
+        # signal recover once the file is readable.
+        #
+        # (see LaunchDarkly::Interfaces::Overrides::OverrideSource#start)
+        #
+        def start(sink)
+          @lock.synchronize do
+            return if @stopped
+
+            @reloader = FileData::Reloader.new(
+              paths: @paths,
+              logger: @logger,
+              log_prefix: LOG_PREFIX,
+              duplicate_keys_handling: @duplicate_keys_handling,
+              skip_missing_paths: true,
+              skip_unchanged: true,
+              apply: lambda do |merged|
+                sink.set_overrides(merged.flags.values, merged.segments.values)
+                log_overrides_in_effect(merged)
+              end
+            )
+          end
+
+          @reloader.reload_now
+
+          @lock.synchronize do
+            return if @stopped
+
+            trigger = @reloader.method(:trigger)
+            @change_detector =
+              if @change_detection == :watching
+                FileData::Watcher.new(@paths, trigger, @logger)
+              else
+                FileData::Poller.new(@paths, @poll_interval, trigger, @logger)
+              end
+          end
+        end
+
+        # (see LaunchDarkly::Interfaces::Overrides::OverrideSource#stop)
+        def stop
+          reloader = nil
+          change_detector = nil
+          @lock.synchronize do
+            @stopped = true
+            reloader = @reloader
+            change_detector = @change_detector
+            @reloader = nil
+            @change_detector = nil
+          end
+          change_detector&.stop
+          reloader&.stop
+        end
+
+        #
+        # Reports the overrides now in effect and the file each came from. The reloader applies a
+        # snapshot only when the content changed, so this logs each change once.
+        #
+        # @param merged [LaunchDarkly::Impl::FileData::MergeResult]
+        #
+        private def log_overrides_in_effect(merged)
+          details = merged.files.map do |file|
+            if !file.present
+              "#{file.path}: absent"
+            elsif file.flags.zero? && file.segments.zero?
+              "#{file.path}: no entries"
+            else
+              "#{file.path}: #{counts_text(file.flags, file.segments)}"
+            end
+          end.join("; ")
+
+          if merged.empty?
+            @logger.info { "#{LOG_PREFIX} Flag overrides: none in effect (#{details})" }
+          else
+            @logger.info do
+              "#{LOG_PREFIX} Flag overrides in effect: #{counts_text(merged.flags.length, merged.segments.length)} (#{details})"
+            end
+          end
+        end
+
+        # Formats flag and segment counts, for example "2 flags, 1 segment".
+        private def counts_text(flags, segments)
+          parts = []
+          parts << pluralize(flags, "flag") if flags > 0
+          parts << pluralize(segments, "segment") if segments > 0
+          parts.join(", ")
+        end
+
+        private def pluralize(count, noun)
+          count == 1 ? "1 #{noun}" : "#{count} #{noun}s"
+        end
+      end
+    end
+  end
+end
diff --git a/lib/ldclient-rb/integrations/file_data.rb b/lib/ldclient-rb/integrations/file_data.rb
index 1f4662e1..1064c8b8 100644
--- a/lib/ldclient-rb/integrations/file_data.rb
+++ b/lib/ldclient-rb/integrations/file_data.rb
@@ -1,5 +1,7 @@
+require 'ldclient-rb/impl/file_data'
 require 'ldclient-rb/impl/integrations/file_data_source'
 require 'ldclient-rb/impl/integrations/file_data_source_v2'
+require 'ldclient-rb/impl/integrations/file_override_source'
 
 module LaunchDarkly
   module Integrations
@@ -81,6 +83,11 @@ module Integrations
     # If the data source encounters any error in any file-- malformed content, a missing file, or a
     # duplicate key-- it will not load flags from any of the files.
     #
+    # The same file format serves the file-based override source, {FileData.override_source}, which
+    # supplies flag and segment overrides that take precedence over LaunchDarkly data instead of
+    # replacing the connection to LaunchDarkly. Flag overrides are currently experimental and subject
+    # to change.
+    #
     module FileData
       #
       # Returns a factory for the file data source component.
@@ -149,6 +156,170 @@ def self.data_source_v2(options = {})
 
         FileDataSourceV2Builder.new(paths, poll_interval)
       end
+
+      #
+      # Returns a builder for a file-based override source. Flag overrides are currently experimental
+      # and subject to change.
+      #
+      # Overrides are flag and segment definitions that take precedence over data received from
+      # LaunchDarkly at evaluation time, on a per-key basis. They exist for resilience during an
+      # incident. An operator can force one or more flags to a known state on a running application,
+      # whether or not the application can reach LaunchDarkly, by writing them to a file. The override
+      # stays in effect until it is removed from the file. Flags not present in the files are
+      # completely unaffected.
+      #
+      # The files use the same document format as the file data source: a JSON or YAML document with
+      # optional `flags`, `flagValues`, and `segments` members. A `flagValues` entry expands into a
+      # flag that is on and serves the given value as its fallthrough for every context. Such a flag is
+      # not available to client-side SDKs, so `all_flags_state` with `client_side_only: true` leaves it
+      # out, even when the flag it overrides is available to them. To override such a flag, define it
+      # in `flags` with `"clientSide": true`. When several files are configured, their
+      # entries are combined in the configured order, and the duplicate keys handling decides what
+      # happens when the same key appears in more than one file.
+      #
+      # A file whose name ends in `.yaml` or `.yml`, in any case, is read as YAML, and any other file is
+      # read as JSON. Files must be UTF-8. Whether a JSON file may contain comments or repeated keys
+      # depends on the application's version of the `json` gem.
+      #
+      # A reload replaces the entire set of overrides, so removing an entry from a file removes the
+      # override. A configured file that does not exist contributes no overrides. It can be created
+      # later, and deleting a file removes its overrides. A file that exists but cannot be read or
+      # parsed fails that whole reload. The previously loaded overrides stay in effect, the failure is
+      # logged, and the source retries after a short delay and recovers on its own once the file is
+      # readable. Whenever the set of overrides in effect changes, including at startup, the source
+      # logs the overrides in effect and what each file supplied, at Info level.
+      #
+      # Files can be rewritten in place: if the source finds a file empty while it is being rewritten,
+      # it keeps the overrides in effect until the new content is written. To remove a file's
+      # overrides, delete the file or write a document with no entries, such as `{}`. An empty JSON
+      # file is not a valid document. For a YAML file, replacing its contents in one operation, such
+      # as by renaming a fully written temporary file over it, avoids briefly applying a partly
+      # written file.
+      #
+      # An evaluation that an override affects, directly or through a prerequisite or segment, is
+      # marked: its reason reports {EvaluationReason#override_affected}, it produces no individual
+      # analytics event, and it is counted under a separate summary counter.
+      #
+      # Pass the returned builder to {LaunchDarkly::DataSystem::ConfigBuilder#overrides}. The
+      # configuration is validated when the client is created, and an invalid configuration raises
+      # `ArgumentError` from `LDClient.new`.
+      #
+      # @example
+      #   overrides = LaunchDarkly::Integrations::FileData.override_source(paths: ["/etc/launchdarkly/overrides.json"])
+      #   config = LaunchDarkly::Config.new(data_system: LaunchDarkly::DataSystem.default.overrides(overrides))
+      #   client = LaunchDarkly::LDClient.new(sdk_key, config)
+      #
+      # @param options [Hash] the configuration options
+      # @option options [Array<String>, String] :paths  One or more files, in precedence order. Required.
+      #   Each path must be a non-empty String, absolute or relative to the current working directory.
+      # @option options [Symbol] :duplicate_keys_handling  What to do when the same key appears in more
+      #   than one file. `:fail` (the default) treats the reload as failed and keeps the previous
+      #   overrides. `:ignore` keeps the entry from the first configured file that defines the key.
+      # @option options [Symbol] :change_detection  How the source detects file changes. `:polling` (the
+      #   default) examines the files on an interval and works on every file system, including network
+      #   mounts and directories whose contents are swapped through symbolic links. `:watching` uses file
+      #   system change notifications through the `listen` gem, which the host application must provide.
+      # @option options [Numeric] :poll_interval  Seconds between examinations of the files in polling
+      #   mode. The default is 1. An interval below 1 is raised to 1 with a warning.
+      # @return [FileOverrideSourceBuilder] a builder for {LaunchDarkly::DataSystem::ConfigBuilder#overrides}
+      #
+      def self.override_source(options = {})
+        FileOverrideSourceBuilder.new(options)
+      end
+    end
+
+    #
+    # Builder for the file-based override source. Create it with {FileData.override_source}.
+    #
+    # Flag overrides are currently experimental and subject to change.
+    #
+    class FileOverrideSourceBuilder
+      # The default interval, in seconds, between examinations of the files in polling mode.
+      DEFAULT_POLL_INTERVAL = 1
+
+      # The shortest allowed polling interval, in seconds. A configured interval below this is raised
+      # to it. The minimum exists only to prevent a tight loop over the file system.
+      MINIMUM_POLL_INTERVAL = 1
+
+      DUPLICATE_KEYS_HANDLING_VALUES = [:fail, :ignore].freeze
+      CHANGE_DETECTION_VALUES = [:polling, :watching].freeze
+      OPTION_KEYS = [:paths, :duplicate_keys_handling, :change_detection, :poll_interval].freeze
+      private_constant :DUPLICATE_KEYS_HANDLING_VALUES, :CHANGE_DETECTION_VALUES, :OPTION_KEYS
+
+      #
+      # @param options [Hash] see {FileData.override_source}
+      #
+      def initialize(options)
+        raise ArgumentError, "options for the file-based override source must be a Hash" unless options.is_a?(Hash)
+
+        @options = options
+      end
+
+      #
+      # Builds the override source. Called by the SDK when the client is created.
+      #
+      # @param sdk_key [String] unused
+      # @param config [LaunchDarkly::Config] the client configuration, for its logger
+      # @return [LaunchDarkly::Interfaces::Overrides::OverrideSource]
+      # @raise [ArgumentError] if the options are invalid
+      #
+      def build(sdk_key, config)
+        unknown = @options.keys - OPTION_KEYS
+        raise ArgumentError, "unknown options for the file-based override source: #{unknown.join(', ')}" unless unknown.empty?
+
+        paths = Array(@options[:paths])
+        raise ArgumentError, "no file paths were specified for the file-based override source" if paths.empty?
+
+        invalid = paths.reject { |path| path.is_a?(String) && !path.empty? }
+        unless invalid.empty?
+          raise ArgumentError,
+            "file paths for the file-based override source must be non-empty strings; got #{invalid.map(&:inspect).join(', ')}"
+        end
+        paths = Impl::FileData.absolute_paths(paths)
+        # A path listed more than once would make every entry of that file a duplicate of itself, so
+        # the repeats are dropped and the first occurrence keeps its place in the order. File names
+        # on Windows ignore case, so there paths that differ only in case name the same file.
+        unique_paths = Gem.win_platform? ? paths.uniq(&:downcase) : paths.uniq
+        if unique_paths.size != paths.size
+          config.logger.warn do
+            "#{Impl::Integrations::FileOverrideSource::LOG_PREFIX} File paths list the same file more than once; the repeats are ignored"
+          end
+          paths = unique_paths
+        end
+
+        duplicate_keys_handling = @options.fetch(:duplicate_keys_handling, :fail)
+        unless DUPLICATE_KEYS_HANDLING_VALUES.include?(duplicate_keys_handling)
+          raise ArgumentError,
+            "unrecognized duplicate keys handling #{duplicate_keys_handling.inspect} for the file-based override source"
+        end
+
+        change_detection = @options.fetch(:change_detection, :polling)
+        unless CHANGE_DETECTION_VALUES.include?(change_detection)
+          raise ArgumentError, "unrecognized change detection mode #{change_detection.inspect} for the file-based override source"
+        end
+        if change_detection == :watching && !Impl::FileData::Watcher.available?
+          raise ArgumentError, "change detection mode :watching for the file-based override source requires the listen gem"
+        end
+
+        poll_interval = @options.fetch(:poll_interval, DEFAULT_POLL_INTERVAL)
+        unless poll_interval.is_a?(Numeric)
+          raise ArgumentError, "poll interval #{poll_interval.inspect} for the file-based override source must be a number"
+        end
+        if change_detection == :polling && poll_interval < MINIMUM_POLL_INTERVAL
+          config.logger.warn do
+            "#{Impl::Integrations::FileOverrideSource::LOG_PREFIX} Poll interval #{poll_interval}s is below the minimum; using #{MINIMUM_POLL_INTERVAL}s"
+          end
+          poll_interval = MINIMUM_POLL_INTERVAL
+        end
+
+        Impl::Integrations::FileOverrideSource.new(
+          paths: paths,
+          duplicate_keys_handling: duplicate_keys_handling,
+          change_detection: change_detection,
+          poll_interval: poll_interval,
+          logger: config.logger
+        )
+      end
     end
 
     #
diff --git a/spec/impl/file_data/reloader_spec.rb b/spec/impl/file_data/reloader_spec.rb
index 194fd1d5..df3807c6 100644
--- a/spec/impl/file_data/reloader_spec.rb
+++ b/spec/impl/file_data/reloader_spec.rb
@@ -448,6 +448,19 @@ def with_reloader(paths, logger: $null_log, reject: 0, **options)
           end
         end
 
+        it "writes the configured prefix once on the log line for an unexpected error" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          logger = CapturingLogger.new
+          allow(FileData).to receive(:read_file).and_raise(RuntimeError, "unexpected")
+
+          with_reloader([a], logger: logger, debounce_delay: 0, log_prefix: "[LDClient] Prefix:") do |reloader, _recorder|
+            reloader.trigger
+
+            expect(wait_for { logger.output.include?("Unexpected error") }).to be true
+            expect(logger.output).to match(/ERROR -- : \[LDClient\] Prefix: Unexpected error in file data reloader/)
+          end
+        end
+
         describe "a file that is found empty" do
           # The retry comes long enough after the change signal for a test to finish writing the
           # file before it, even on a slow machine.
diff --git a/spec/integrations/file_override_source_spec.rb b/spec/integrations/file_override_source_spec.rb
new file mode 100644
index 00000000..4df47d21
--- /dev/null
+++ b/spec/integrations/file_override_source_spec.rb
@@ -0,0 +1,705 @@
+# frozen_string_literal: true
+
+require "spec_helper"
+require "capturing_logger"
+require "in_place_writes"
+require "model_builders"
+require "override_test_components"
+require "tmpdir"
+require "ldclient-rb/integrations/file_data"
+
+module LaunchDarkly
+  module Integrations
+    describe "FileData.override_source" do
+      # Records every snapshot the source supplies.
+      class RecordingSink
+        include Interfaces::Overrides::OverrideSink
+
+        def initialize
+          @lock = Mutex.new
+          @snapshots = []
+        end
+
+        def set_overrides(flags, segments)
+          @lock.synchronize { @snapshots << [flags, segments] }
+        end
+
+        def snapshots
+          @lock.synchronize { @snapshots.dup }
+        end
+
+        def flag_values(index = -1)
+          snapshots[index][0].to_h { |flag| [flag.key, flag.variations[0]] }
+        end
+
+        def segment_keys(index = -1)
+          snapshots[index][1].map(&:key)
+        end
+      end
+
+      around do |example|
+        Dir.mktmpdir do |dir|
+          @dir = dir
+          example.run
+        end
+      end
+
+      let(:logger) { CapturingLogger.new }
+      let(:config) { Config.new(logger: logger) }
+
+      def path(name)
+        File.join(@dir, name)
+      end
+
+      # Applications rewrite override files in place, which is not atomic. The tests do too,
+      # deliberately slowly: a file that exists is truncated, and written 30 ms later, so a load can
+      # find it empty, as it would with an application's writes, instead of only by chance.
+      def write(name, content)
+        if File.exist?(path(name))
+          InPlaceWrites.rewrite(path(name), content, 0.03)
+        else
+          File.write(path(name), content)
+        end
+        path(name)
+      end
+
+      def yaml_values_doc(values)
+        "flagValues:\n" + values.map { |key, value| "  #{key}: #{value}\n" }.join
+      end
+
+      def values_doc(values)
+        { flagValues: values }.to_json
+      end
+
+      def wait_for(timeout = 5)
+        deadline = Time.now + timeout
+        until yield
+          return false if Time.now > deadline
+          sleep 0.02
+        end
+        true
+      end
+
+      def build(options)
+        FileData.override_source(options).build("sdk-key", config)
+      end
+
+      # Builds the source directly so that tests can use a short polling interval.
+      def make_source(paths, duplicate_keys_handling: :fail, change_detection: :polling, poll_interval: 0.05)
+        Impl::Integrations::FileOverrideSource.new(
+          paths: Impl::FileData.absolute_paths(paths),
+          duplicate_keys_handling: duplicate_keys_handling,
+          change_detection: change_detection,
+          poll_interval: poll_interval,
+          logger: logger
+        )
+      end
+
+      def with_source(paths, **options)
+        source = make_source(paths, **options)
+        sink = RecordingSink.new
+        source.start(sink)
+        begin
+          yield source, sink
+        ensure
+          source.stop
+        end
+      end
+
+      describe "builder" do
+        it "builds a polling source with the default options and absolute paths" do
+          source = build(paths: ["relative.json"])
+
+          expect(source).to be_a(Interfaces::Overrides::OverrideSource)
+          expect(source.paths).to eq [File.absolute_path("relative.json")]
+          expect(source.duplicate_keys_handling).to eq :fail
+          expect(source.change_detection).to eq :polling
+          expect(source.poll_interval).to eq 1
+        end
+
+        it "accepts a single path string and explicit options" do
+          source = build(paths: "one.json", duplicate_keys_handling: :ignore, poll_interval: 2.5)
+
+          expect(source.paths).to eq [File.absolute_path("one.json")]
+          expect(source.duplicate_keys_handling).to eq :ignore
+          expect(source.poll_interval).to eq 2.5
+        end
+
+        it "drops a path listed more than once with a warning" do
+          # Under the default fail handling the same file listed twice would make every one of its
+          # keys a duplicate of itself. The repeat is ignored and the first occurrence keeps its place.
+          source = build(paths: ["a.json", "b.json", "a.json"])
+
+          expect(source.paths).to eq [File.absolute_path("a.json"), File.absolute_path("b.json")]
+          expect(logger.output).to include("File paths list the same file more than once")
+        end
+
+        it "treats a path repeated in different case as a repeat on Windows, where file names ignore case" do
+          allow(Gem).to receive(:win_platform?).and_return(true)
+          source = build(paths: ["a.json", "A.json"])
+
+          expect(source.paths).to eq [File.absolute_path("a.json")]
+          expect(logger.output).to include("File paths list the same file more than once")
+        end
+
+        it "keeps paths that differ only in case elsewhere, where they name different files" do
+          allow(Gem).to receive(:win_platform?).and_return(false)
+          source = build(paths: ["a.json", "A.json"])
+
+          expect(source.paths).to eq [File.absolute_path("a.json"), File.absolute_path("A.json")]
+          expect(logger.output).not_to include("File paths list the same file more than once")
+        end
+
+        it "rejects a configuration with no file paths" do
+          expect { build({}) }.to raise_error(ArgumentError, /no file paths/)
+          expect { build(paths: []) }.to raise_error(ArgumentError, /no file paths/)
+          expect { build(paths: nil) }.to raise_error(ArgumentError, /no file paths/)
+        end
+
+        it "rejects a file path that is not a non-empty string" do
+          expect { build(paths: [""]) }.to raise_error(ArgumentError, /must be non-empty strings/)
+          expect { build(paths: ["a.json", nil]) }.to raise_error(ArgumentError, /must be non-empty strings/)
+          expect { build(paths: [:a]) }.to raise_error(ArgumentError, /must be non-empty strings/)
+        end
+
+        it "rejects options that are not a hash and unknown option keys" do
+          expect { FileData.override_source("x") }.to raise_error(ArgumentError, /must be a Hash/)
+          expect { build(paths: ["a.json"], path: "b.json") }.to raise_error(ArgumentError, /unknown options.*path/)
+        end
+
+        it "rejects an unrecognized duplicate keys handling" do
+          expect { build(paths: ["a.json"], duplicate_keys_handling: :first) }.to raise_error(ArgumentError, /duplicate keys handling/)
+          expect { build(paths: ["a.json"], duplicate_keys_handling: "fail") }.to raise_error(ArgumentError, /duplicate keys handling/)
+        end
+
+        it "rejects an unrecognized change detection mode" do
+          expect { build(paths: ["a.json"], change_detection: :notify) }.to raise_error(ArgumentError, /change detection/)
+        end
+
+        it "rejects watching when the listen gem is not available" do
+          allow(Impl::FileData::Watcher).to receive(:available?).and_return(false)
+
+          expect { build(paths: ["a.json"], change_detection: :watching) }.to raise_error(ArgumentError, /listen gem/)
+        end
+
+        it "builds a watching source when the listen gem is available" do
+          allow(Impl::FileData::Watcher).to receive(:available?).and_return(true)
+
+          expect(build(paths: ["a.json"], change_detection: :watching).change_detection).to eq :watching
+        end
+
+        it "rejects a poll interval that is not a number" do
+          expect { build(paths: ["a.json"], poll_interval: "1") }.to raise_error(ArgumentError, /poll interval/)
+        end
+
+        it "raises a poll interval below the minimum to the minimum with a warning" do
+          source = build(paths: ["a.json"], poll_interval: 0.1)
+
+          expect(source.poll_interval).to eq 1
+          expect(logger.output).to include("Poll interval 0.1s is below the minimum")
+        end
+
+        it "does not warn about the poll interval in watching mode" do
+          allow(Impl::FileData::Watcher).to receive(:available?).and_return(true)
+
+          build(paths: ["a.json"], change_detection: :watching, poll_interval: 0.1)
+
+          expect(logger.output).not_to include("below the minimum")
+        end
+      end
+
+      describe "initial load" do
+        it "supplies the merged files to the sink before start returns" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          b = write("b.json", { flags: { flag2: { key: "flag2", on: false, offVariation: 0, variations: ["b"] } },
+                                segments: { seg1: { key: "seg1", included: ["user"] } } }.to_json)
+
+          with_source([a, b]) do |_source, sink|
+            expect(sink.snapshots.length).to eq 1
+            expect(sink.flag_values).to eq({ "flag1" => "a", "flag2" => "b" })
+            expect(sink.segment_keys).to eq ["seg1"]
+          end
+        end
+
+        it "expands a flag value into a flag that is on and serves the value by fallthrough" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            flag = sink.snapshots[0][0][0]
+            expect(flag.on).to be true
+            expect(flag.off_variation).to be_nil
+            expect(flag.fallthrough.variation).to eq 0
+            expect(flag.variations).to eq ["a"]
+          end
+        end
+
+        it "reads YAML files" do
+          a = write("a.yaml", "flagValues:\n  yaml-flag: \"override-value\"\n")
+
+          with_source([a]) do |_source, sink|
+            expect(sink.flag_values).to eq({ "yaml-flag" => "override-value" })
+          end
+        end
+
+        it "treats a configured file that does not exist as contributing no overrides" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a, path("missing.json")]) do |_source, sink|
+            expect(sink.snapshots.length).to eq 1
+            expect(sink.flag_values).to eq({ "flag1" => "a" })
+            expect(logger.output).not_to include("ERROR")
+          end
+        end
+
+        it "supplies an empty snapshot when no configured file exists" do
+          with_source([path("missing.json")]) do |_source, sink|
+            expect(sink.snapshots).to eq [[[], []]]
+          end
+        end
+
+        it "fails the load and logs when a file cannot be parsed, leaving the sink untouched" do
+          a = write("a.json", '{"flagValues"')
+
+          with_source([a]) do |_source, sink|
+            expect(sink.snapshots).to be_empty
+            expect(logger.output).to include("ERROR")
+            expect(logger.output).to include("FileOverrideSource: Unable to load flags")
+          end
+        end
+
+        it "fails the load when a key appears in two files with the default handling" do
+          a = write("a.json", values_doc({ flag1: "first" }))
+          b = write("b.json", values_doc({ flag1: "second" }))
+
+          with_source([a, b]) do |_source, sink|
+            expect(sink.snapshots).to be_empty
+            expect(logger.output).to include("was used more than once")
+          end
+        end
+
+        it "keeps the first file's entry with ignore handling" do
+          a = write("a.json", values_doc({ flag1: "first" }))
+          b = write("b.json", values_doc({ flag1: "second", flag2: "other" }))
+
+          with_source([a, b], duplicate_keys_handling: :ignore) do |_source, sink|
+            expect(sink.flag_values).to eq({ "flag1" => "first", "flag2" => "other" })
+          end
+        end
+      end
+
+      describe "logging" do
+        it "logs the overrides in effect and what each file supplied" do
+          a = write("a.json", { flagValues: { flag1: "a", flag2: "b" }, segments: { seg1: { key: "seg1" } } }.to_json)
+          b = write("b.json", "{}")
+          missing = path("missing.json")
+
+          with_source([a, b, missing]) do |_source, _sink|
+            expect(logger.output).to include(
+              "INFO -- : [LDClient] FileOverrideSource: Flag overrides in effect: 2 flags, 1 segment " \
+              "(#{a}: 2 flags, 1 segment; #{b}: no entries; #{missing}: absent)"
+            )
+          end
+        end
+
+        it "logs when no overrides are in effect" do
+          a = write("a.json", "{}")
+
+          with_source([a]) do |_source, _sink|
+            expect(logger.output).to include("Flag overrides: none in effect (#{a}: no entries)")
+          end
+        end
+      end
+
+      describe "polling change detection" do
+        it "applies a changed file" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            sleep 0.1
+            write("a.json", values_doc({ flag1: "b" }))
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "b" })
+          end
+        end
+
+        it "applies a file that appears after start" do
+          with_source([path("later.json")]) do |_source, sink|
+            expect(sink.snapshots.length).to eq 1
+            write("later.json", values_doc({ flag1: "a" }))
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "a" })
+          end
+        end
+
+        it "removes the overrides of a deleted file" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            File.delete(a)
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.snapshots[-1]).to eq [[], []]
+          end
+        end
+
+        it "keeps the last good overrides through a malformed edit and recovers" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            write("a.json", '{"flagValues"')
+            sleep 0.5
+            expect(sink.snapshots.length).to eq 1
+            expect(logger.output).to include("Unable to load flags")
+
+            write("a.json", values_doc({ flag1: "c" }))
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "c" })
+          end
+        end
+
+        it "does not supply a snapshot for a rewrite with identical content" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            sleep 0.1
+            File.utime(Time.now + 5, Time.now + 5, a)
+            sleep 0.5
+
+            expect(sink.snapshots.length).to eq 1
+          end
+        end
+
+        it "stops detecting changes when stopped" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          source = make_source([a])
+          sink = RecordingSink.new
+          source.start(sink)
+          poller_threads = Thread.list.select { |t| t.name == "LD/FileDataPoller" }
+          expect(poller_threads.length).to eq 1
+
+          source.stop
+
+          expect(poller_threads[0].alive?).to be false
+          write("a.json", values_doc({ flag1: "b" }))
+          sleep 0.5
+          expect(sink.snapshots.length).to eq 1
+        end
+
+        it "does not start when stopped before start" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+          source = make_source([a])
+          sink = RecordingSink.new
+          source.stop
+          source.start(sink)
+
+          expect(sink.snapshots).to be_empty
+        end
+      end
+
+      describe "watching change detection" do
+        before do
+          skip "the listen gem is not installed" unless Impl::FileData::Watcher.available?
+        end
+
+        it "applies a changed file" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a], change_detection: :watching) do |_source, sink|
+            sleep 0.3
+            write("a.json", values_doc({ flag1: "b" }))
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "b" })
+          end
+        end
+
+        it "applies a file that appears after start" do
+          with_source([path("later.json")], change_detection: :watching) do |_source, sink|
+            sleep 0.3
+            write("later.json", values_doc({ flag1: "a" }))
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "a" })
+          end
+        end
+
+        it "removes the overrides of a deleted file" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a], change_detection: :watching) do |_source, sink|
+            sleep 0.3
+            File.delete(a)
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.snapshots[-1]).to eq [[], []]
+          end
+        end
+      end
+
+      describe "in-place rewrites" do
+        # A writer that rewrites a file in place truncates it, and writes the new content after this
+        # pause. A load in between finds the file empty.
+        let(:pause) { 0.3 }
+
+        before do
+          # The retry comes long enough after a load that finds the file empty for the rewrite to
+          # finish before it, even on a slow machine.
+          stub_const("LaunchDarkly::Impl::FileData::Reloader::DEFAULT_RETRY_DELAY", 1.5)
+        end
+
+        [:polling, :watching].each do |mode|
+          context "with #{mode} change detection" do
+            before do
+              skip "the listen gem is not installed" if mode == :watching && !Impl::FileData::Watcher.available?
+            end
+
+            it "never supplies a snapshot without the override while a JSON file is rewritten" do
+              a = write("a.json", values_doc({ flag1: "a" }))
+
+              with_source([a], change_detection: mode) do |_source, sink|
+                sleep 0.3 if mode == :watching
+                ["b", "c"].each do |value|
+                  InPlaceWrites.rewrite(a, values_doc({ flag1: value }), pause)
+                  expect(wait_for { sink.flag_values == { "flag1" => value } }).to be true
+                end
+
+                expect(sink.snapshots.map { |flags, _segments| flags.map(&:key) }).to all(eq(["flag1"]))
+                expect(logger.output).not_to include("ERROR")
+              end
+            end
+
+            it "never supplies a snapshot without the override while a YAML file is rewritten" do
+              a = write("a.yaml", yaml_values_doc({ flag1: "a" }))
+
+              with_source([a], change_detection: mode) do |_source, sink|
+                sleep 0.3 if mode == :watching
+                ["b", "c"].each do |value|
+                  InPlaceWrites.rewrite(a, yaml_values_doc({ flag1: value }), pause)
+                  expect(wait_for { sink.flag_values == { "flag1" => value } }).to be true
+                end
+
+                expect(sink.snapshots.map { |flags, _segments| flags.map(&:key) }).to all(eq(["flag1"]))
+              end
+            end
+
+            it "never supplies a snapshot without the override while a YAML file that starts with a comment is rewritten" do
+              # A load during the pause finds only the comment, which the YAML parser reads as nothing.
+              comment = "# overrides for the incident\n"
+              a = write("a.yaml", comment + yaml_values_doc({ flag1: "a" }))
+
+              with_source([a], change_detection: mode) do |_source, sink|
+                sleep 0.3 if mode == :watching
+                ["b", "c"].each do |value|
+                  InPlaceWrites.rewrite(a, comment + yaml_values_doc({ flag1: value }), pause, first: comment)
+                  expect(wait_for { sink.flag_values == { "flag1" => value } }).to be true
+                end
+
+                expect(sink.snapshots.map { |flags, _segments| flags.map(&:key) }).to all(eq(["flag1"]))
+              end
+            end
+          end
+        end
+
+        it "keeps the overrides in effect when a JSON file stays empty" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            File.write(a, "")
+
+            expect(wait_for { logger.output.include?("the file has no content, which is not a JSON document") }).to be true
+            expect(sink.snapshots.length).to eq 1
+            expect(sink.flag_values).to eq({ "flag1" => "a" })
+          end
+        end
+
+        it "removes the overrides of a YAML file that stays empty after the retry delay" do
+          a = write("a.yaml", yaml_values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            File.write(a, "")
+
+            expect(wait_for { sink.snapshots.length == 2 }).to be true
+            expect(sink.snapshots[-1]).to eq [[], []]
+            expect(logger.output).to include("reading it again before applying it")
+          end
+        end
+      end
+
+      describe "files that are not UTF-8" do
+        # Windows PowerShell 5.1 writes a file with `>` in UTF-16 with a byte order mark.
+        def utf16(text)
+          "\xFF\xFE".b + text.encode("UTF-16LE").b
+        end
+
+        it "keeps the overrides in effect when a file is rewritten as UTF-16, and applies the fix" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            File.binwrite(a, utf16(values_doc({ flag1: "b" })))
+            expect(wait_for { logger.output.include?("encoded in UTF-16 or UTF-32") }).to be true
+            expect(sink.snapshots.length).to eq 1
+
+            File.write(a, values_doc({ flag1: "c" }))
+            expect(wait_for { sink.flag_values == { "flag1" => "c" } }).to be true
+            expect(logger.output.scan("ERROR").length).to eq 1
+          end
+        end
+
+        it "starts with a UTF-16 file without its overrides, and applies the file once it is fixed" do
+          a = path("a.json")
+          File.binwrite(a, utf16(values_doc({ flag1: "a" })))
+
+          with_source([a]) do |_source, sink|
+            expect(sink.snapshots).to be_empty
+            expect(logger.output).to include("encoded in UTF-16 or UTF-32")
+
+            File.write(a, values_doc({ flag1: "b" }))
+            expect(wait_for { sink.snapshots.length == 1 }).to be true
+            expect(sink.flag_values).to eq({ "flag1" => "b" })
+          end
+        end
+
+        it "keeps the overrides in effect when a file is rewritten with content that is not UTF-8" do
+          a = write("a.json", values_doc({ flag1: "a" }))
+
+          with_source([a]) do |_source, sink|
+            File.binwrite(a, "{\"flagValues\": {\"flag1\": \"caf\xE9\"}}".b)
+            expect(wait_for { logger.output.include?("the file is not valid UTF-8") }).to be true
+            expect(sink.snapshots.length).to eq 1
+          end
+        end
+      end
+
+      describe "in-place rewrites with the default timing" do
+        it "never supplies a snapshot without the override while a YAML file is rewritten about once a second" do
+          # Polling once a second, a poll, or the retry, can find the second rewrite's truncation
+          # before any load has read the first rewrite's content.
+          a = write("a.yaml", yaml_values_doc({ flag1: "v0" }))
+          source = build(paths: [a])
+          sink = RecordingSink.new
+          source.start(sink)
+          begin
+            (1..5).each do |i|
+              InPlaceWrites.rewrite(a, yaml_values_doc({ flag1: "v#{i}" }), 0.3)
+              sleep 0.7
+            end
+            expect(wait_for { sink.flag_values == { "flag1" => "v5" } }).to be true
+            expect(sink.snapshots.map { |flags, _segments| flags.map(&:key) }).to all(eq(["flag1"]))
+          ensure
+            source.stop
+          end
+        end
+      end
+
+      describe "with a client" do
+        let(:context) { LDContext.create({ key: "user-key", kind: "user" }) }
+        let(:ld_flag) { { key: "flag", version: 100, on: false, offVariation: 0, variations: ["ld-value"] } }
+
+        def with_client(overrides)
+          data_system = DataSystem.custom
+            .initializers([TestDataInitializer.new(flags: { flag: ld_flag })])
+            .overrides(overrides)
+            .build
+          client = LDClient.new("sdk-key", Config.new(data_system_config: data_system, send_events: false, logger: logger), 5)
+          begin
+            yield client
+          ensure
+            client.close
+          end
+        end
+
+        it "serves overrides from the files and reloads them on a running client" do
+          a = write("a.json", values_doc({ flag: "override-value", other: true }))
+          overrides = FileData.override_source(paths: [a])
+
+          with_client(overrides) do |client|
+            detail = client.variation_detail("flag", context, "default")
+            expect(detail.value).to eq "override-value"
+            expect(detail.reason).to eq EvaluationReason.fallthrough.with_override_affected(true)
+            expect(client.variation("other", context, false)).to be true
+
+            write("a.json", "{}")
+            expect(wait_for { client.variation("flag", context, "default") == "ld-value" }).to be true
+            expect(client.variation("other", context, false)).to be false
+          end
+        end
+
+        it "leaves a flagValues entry out of a client-side state and keeps a definition with clientSide" do
+          # A flagValues entry is not available to client-side SDKs. A full definition that sets
+          # clientSide is, so that is how an operator overrides a flag that client-side SDKs use.
+          a = write("a.json", JSON.generate({
+            flagValues: { "value-flag": true },
+            flags: {
+              "full-flag": { key: "full-flag", version: 1, on: true, variations: [true],
+                             fallthrough: { variation: 0 }, clientSide: true },
+            },
+          }))
+
+          with_client(FileData.override_source(paths: [a])) do |client|
+            expect(client.all_flags_state(context).values_map.keys).to include("value-flag", "full-flag")
+            state = client.all_flags_state(context, client_side_only: true)
+            expect(state.valid?).to be true
+            expect(state.values_map).to eq({ "full-flag" => true })
+          end
+        end
+
+        [
+          ["JSON", "a.json", ->(value) { { flagValues: { flag: value } }.to_json }],
+          ["YAML", "a.yaml", ->(value) { "flagValues:\n  flag: #{value}\n" }],
+        ].each do |format, name, doc|
+          it "serves the override throughout an in-place rewrite of a #{format} file" do
+            stub_const("LaunchDarkly::Impl::FileData::Reloader::DEFAULT_RETRY_DELAY", 1.5)
+            a = write(name, doc.call("override-a"))
+            change_detection = Impl::FileData::Watcher.available? ? :watching : :polling
+            source = make_source([a], change_detection: change_detection)
+            builder = Object.new
+            builder.define_singleton_method(:build) { |_sdk_key, _config| source }
+
+            with_client(builder) do |client|
+              sleep 0.3 if change_detection == :watching
+              served = []
+              sampling = true
+              sampler = Thread.new do
+                while sampling
+                  value = client.variation("flag", context, "default")
+                  served << value if served.last != value
+                  sleep 0.002
+                end
+              end
+
+              InPlaceWrites.rewrite(a, doc.call("override-b"), 0.3)
+              expect(wait_for { served.last == "override-b" }).to be true
+              sampling = false
+              sampler.join
+
+              expect(served).to eq ["override-a", "override-b"]
+            end
+          end
+        end
+
+        it "serves a UTF-16 override file once it is fixed, after starting without it" do
+          a = path("a.json")
+          File.binwrite(a, "\xFF\xFE".b + { flagValues: { flag: "override-a" } }.to_json.encode("UTF-16LE").b)
+          source = make_source([a])
+          builder = Object.new
+          builder.define_singleton_method(:build) { |_sdk_key, _config| source }
+
+          with_client(builder) do |client|
+            expect(client.variation("flag", context, "default")).to eq "ld-value"
+            File.write(a, { flagValues: { flag: "override-b" } }.to_json)
+            expect(wait_for { client.variation("flag", context, "default") == "override-b" }).to be true
+          end
+        end
+
+        it "reports invalid options from client construction" do
+          data_system = DataSystem.custom.overrides(FileData.override_source(paths: [])).build
+
+          expect { LDClient.new("sdk-key", Config.new(data_system_config: data_system, send_events: false, logger: logger), 0) }
+            .to raise_error(ArgumentError, /no file paths/)
+        end
+      end
+    end
+  end
+end