feat: Add the FDv2 data system and expose it through configuration (#310)

## What this adds

The piece that ties the FDv2 building blocks together and turns them on:
`FDv2DataSystem`, the manager routing that applies FDv2 payloads, and
the configuration that opts in. This completes the FDv2 work in
`common_client` — everything below the public Flutter API.

- **`FDv2DataSystem`** composes the data source factories the
`DataSourceManager` consumes. It owns the state that must outlive any
single orchestrator: the current selector and the context it belongs to.
A fresh orchestrator is built per connection-mode switch and per
identify; the selector survives mode switches (initializers are skipped
when a selector is already held) but resets when the context changes,
since a selector is specific to one context. `buildFactories()` returns
factories for streaming, polling, and background — offline has no data
source and is handled by the manager directly. Custom connection modes
from config replace the built-in definitions by name.
- **`DataSourceManager` payload routing.** The `PayloadEvent` case — a
no-op placeholder in the orchestrator PR — now applies the change set
through `handlePayload` and completes the pending identify, the same way
`DataEvent` does for FDv1.
- **Configuration & exposure.** `DataSystemConfig` (providing it, even
empty, opts into FDv2); `LDCommonConfig.dataSystem`; `LDCommonClient`
constructs an `FDv2DataSystem` and installs its factories when
`dataSystem` is configured, otherwise the FDv1 sources run unchanged;
and the new public types are exported.

When `dataSystem` is absent the FDv1 path is byte-for-byte unchanged, so
existing behavior is unaffected.

## Testing

- `DataSourceManager` test: a `PayloadEvent` is applied through
`handlePayload`, marks the source valid, and completes identify (a
dropped/no-op payload would leave identify hanging — this distinguishes
the real routing from the placeholder).
- `FDv2DataSystem` tests: `buildFactories` exposes
streaming/polling/background and not offline; each factory builds a
fresh data source per call; a custom connection mode replaces the
built-in definition.
- `DataSystemConfig` default (no custom modes).
- Full `common_client` suite passes; `flutter_client_sdk` (the dependent
package) analyzes and tests clean against these changes.
- End-to-end behavior of the wired data system is exercised by the FDv2
contract tests, which land with the contract-service changes in a later
PR (and pass on the integration branch today).

SDK-2186

<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **High Risk**
> Touches core flag acquisition, identify completion,
connection-mode/offline behavior, and persistence read paths; incorrect
timing could leave identifies hanging or serve stale flags, though FDv1
remains unchanged when `dataSystem` is unset.
> 
> **Overview**
> Opts the SDK into **FDv2** when `LDCommonConfig.dataSystem` is set
(even empty), exports `DataSystemConfig` / `ConnectionModeId` and
mode-definition types, and keeps the FDv1 path when `dataSystem` is
absent.
> 
> **`FDv2DataSystem`** builds per-mode orchestrator factories
(streaming, polling, background, **offline** as a real cache-only
pipeline), holds the **selector** across mode switches, and applies
`connectionModes` overrides. **`FDv2DataManager`** clears the selector
on each identify and maps `waitForNetworkResults` to cached vs fresh
availability; FDv1 still loads cache at identify via
**`FDv1DataManager`**.
> 
> **`DataSourceManager`** applies **`PayloadEvent`** through
`handlePayload`, sets **valid** on applied online payloads (including
no-change restores), does not promote valid in offline mode, and
completes identify on first applied data (cached) or
**`InitializedEvent`** (fresh). **`InitializedEvent`** and orchestrator
init semantics separate cache hits from network-ready state. Cache loads
use **`readCached`** without touching the store until the pipeline
applies payloads.
> 
> Adds an **FDv1 fallback** synchronizer (FDv1 poll → FDv2 change sets)
with loop guards when already on fallback.
> 
> <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit
194eae7. Bugbot is set up for automated
code reviews on this repo. Configure
[here](https://www.cursor.com/dashboard/bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: kinyoklion

packages/common_client/lib/launchdarkly_common_client.dart

@@ -6,6 +6,21 @@ export 'src/ld_common_config.dart'
         AutoEnvAttributes,
         PollingConfig;
 
+export 'src/config/data_system_config.dart'
+    show DataSystemConfig, ConnectionModeId;
+export 'src/data_sources/fdv2/mode_definition.dart'
+    show
+        ModeDefinition,
+        EndpointConfig,
+        InitializerEntry,
+        SynchronizerEntry,
+        CacheInitializer,
+        PollingInitializer,
+        StreamingInitializer,
+        PollingSynchronizer,
+        StreamingSynchronizer,
+        Fdv1FallbackConfig;
+
 export 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart'
     show
         LDContext,

packages/common_client/lib/src/config/data_system_config.dart

@@ -0,0 +1,79 @@
+import '../data_sources/fdv2/mode_definition.dart';
+
+// Maintainer note (not public API): ConnectionModeId is a sealed
+// hierarchy rather than an enum so a custom-mode variant can be added
+// later without changing this surface. The planned extension is a custom
+// variant constructed as `ConnectionModeId.custom('my-mode')`:
+//
+//   factory ConnectionModeId.custom(String name) = _CustomConnectionMode;
+//   final class _CustomConnectionMode extends ConnectionModeId {
+//     final String name;
+//     const _CustomConnectionMode(this.name);
+//     // value equality on name so it works as an override-map key
+//   }
+//
+// A custom mode is a distinct type from a built-in, so the two share no
+// namespace: a custom id never equals a built-in id (even with the same
+// name), and so cannot collide with a current or future built-in. The
+// type is the namespace -- no name prefix is needed. This holds only
+// while custom modes stay typed; if one is ever reduced to a bare string
+// (logs, persistence) that reintroduces a shared string space where a
+// prefix would matter again.
+//
+// Equality split: the built-in values are const singletons relying on
+// canonical-instance identity, which lets a connectionModes map of only
+// built-in keys be a const map. A runtime-constructed custom variant must
+// carry value equality, so an override map holding a custom key would be
+// non-const. The built-in variant therefore must not override
+// `==`/`hashCode`.
+
+/// Identifies a built-in connection mode whose data-source pipeline can be
+/// overridden through [DataSystemConfig.connectionModes]: [streaming],
+/// [polling], [background], or [offline].
+sealed class ConnectionModeId {
+  const ConnectionModeId();
+
+  /// The built-in streaming mode.
+  static const ConnectionModeId streaming = _BuiltInConnectionMode('streaming');
+
+  /// The built-in polling mode.
+  static const ConnectionModeId polling = _BuiltInConnectionMode('polling');
+
+  /// The built-in background mode.
+  static const ConnectionModeId background =
+      _BuiltInConnectionMode('background');
+
+  /// The built-in offline mode. Its pipeline loads cached flags and runs
+  /// no synchronizer, so overriding it customizes how the SDK behaves
+  /// while offline (for example, the cache initializer it uses).
+  static const ConnectionModeId offline = _BuiltInConnectionMode('offline');
+}
+
+final class _BuiltInConnectionMode extends ConnectionModeId {
+  final String name;
+
+  const _BuiltInConnectionMode(this.name);
+
+  @override
+  String toString() => 'ConnectionModeId.$name';
+}
+
+/// Configuration for the FDv2 data system.
+///
+/// Providing a [DataSystemConfig] (even an empty one) opts the SDK into
+/// the FDv2 data acquisition protocol. When absent the SDK uses the
+/// FDv1 data sources.
+///
+/// This feature is not stable, and not subject to any backwards
+/// compatibility guarantees or semantic versioning. It is in early
+/// access. If you want access to this feature please join the EAP.
+final class DataSystemConfig {
+  /// Overrides for built-in connection modes. A definition given here
+  /// replaces the built-in pipeline for that mode; modes not present keep
+  /// their built-in definition.
+  final Map<ConnectionModeId, ModeDefinition> connectionModes;
+
+  const DataSystemConfig({
+    this.connectionModes = const {},
+  });
+}

packages/common_client/lib/src/data_sources/data_manager.dart

@@ -0,0 +1,61 @@
+import 'dart:async';
+
+import 'package:launchdarkly_dart_common/launchdarkly_dart_common.dart'
+    show LDContext;
+
+import '../flag_manager/flag_manager.dart';
+import 'data_source_manager.dart';
+
+/// Owns the data-acquisition strategy for an identify: how the cache is
+/// loaded and when the identify resolves. The FDv1 and FDv2 protocols
+/// diverge here, so each has its own implementation; everything else
+/// (connection lifecycle, mode switching, event routing) is shared in the
+/// [DataSourceManager] that both delegate to.
+abstract interface class DataManager {
+  /// Brings the SDK to a usable state for [context], resolving when the
+  /// manager's data-availability strategy is satisfied.
+  ///
+  /// When [waitForNetworkResults] is true the returned future resolves
+  /// only once network (or otherwise fresh) data has arrived; otherwise it
+  /// may resolve as soon as cached data is available.
+  Future<void> identify(LDContext context,
+      {required bool waitForNetworkResults});
+}
+
+final class FDv1DataManager implements DataManager {
+  final DataSourceManager _dataSourceManager;
+  final FlagManager _flagManager;
+
+  FDv1DataManager(this._dataSourceManager, this._flagManager);
+
+  @override
+  Future<void> identify(LDContext context,
+      {required bool waitForNetworkResults}) async {
+    final completer = Completer<void>();
+    final loadedFromCache = await _flagManager.loadCached(context);
+    _dataSourceManager.identify(context, completer);
+    if (loadedFromCache && !waitForNetworkResults) {
+      return;
+    }
+    return completer.future;
+  }
+}
+
+final class FDv2DataManager implements DataManager {
+  final DataSourceManager _dataSourceManager;
+  final void Function() _clearSelector;
+
+  FDv2DataManager(this._dataSourceManager, this._clearSelector);
+
+  @override
+  Future<void> identify(LDContext context,
+      {required bool waitForNetworkResults}) {
+    _clearSelector();
+    final completer = Completer<void>();
+    _dataSourceManager.identify(context, completer,
+        minimumDataAvailability: waitForNetworkResults
+            ? DataAvailability.fresh
+            : DataAvailability.cached);
+    return completer.future;
+  }
+}

packages/common_client/lib/src/data_sources/data_source.dart

@@ -31,6 +31,13 @@ final class StatusEvent implements DataSourceEvent {
       {this.shutdown = false});
 }
 
+/// Emitted once by the FDv2 orchestrator when initialization is complete:
+/// a selector-bearing payload arrived, the initializer chain was exhausted
+/// (with cached data or in a cache-only system), or the first synchronizer
+/// delivered a change set. The manager resolves a wait-for-network identify
+/// on this; a cached identify resolves earlier, on the first applied payload.
+final class InitializedEvent implements DataSourceEvent {}
+
 abstract interface class DataSource {
   Stream<DataSourceEvent> get events;
 

packages/common_client/lib/src/data_sources/data_source_event_handler.dart

@@ -101,14 +101,13 @@ final class DataSourceEventHandler {
   ///
   /// Full change sets replace the stored flags, partial change sets apply
   /// each update, and a change set of type none confirms the SDK is up to
-  /// date without changing data. All three mark the data source valid.
+  /// date without changing data.
   Future<MessageStatus> handlePayload(LDContext context, ChangeSet changeSet,
       {String? environmentId}) async {
     try {
       await _flagManager.applyChanges(
           context, changeSet.updates, changeSet.type,
           environmentId: environmentId);
-      _statusManager.setValid();
       return MessageStatus.messageHandled;
     } catch (err) {
       _logger.error('Failed to apply an FDv2 change set: ${err.runtimeType}');

packages/common_client/lib/src/data_sources/data_source_manager.dart

@@ -13,6 +13,18 @@ import 'data_source_status_manager.dart';
 
 typedef DataSourceFactory = DataSource Function(LDContext context);
 
+/// The minimum data availability an identify must reach before it
+/// completes, mapped from the caller's wait-for-network-results
+/// preference (`false` -> [cached], `true` -> [fresh]).
+enum DataAvailability {
+  /// Resolve as soon as any data is applied, including a cache load.
+  cached,
+
+  /// Wait for fresh network data (the orchestrator's InitializedEvent);
+  /// a cache load alone does not satisfy it.
+  fresh,
+}
+
 /// The data source manager controls which data source is connected to
 /// the data source status as well as the data source event handler.
 final class DataSourceManager {
@@ -38,6 +50,11 @@ final class DataSourceManager {
 
   Completer<void>? _identifyCompleter;
 
+  /// The minimum data availability the active identify must reach before
+  /// it resolves. Set per identify from the caller's
+  /// wait-for-network-results preference.
+  DataAvailability _minimumDataAvailability = DataAvailability.cached;
+
   DataSourceManager({
     ConnectionMode startingMode = ConnectionMode.streaming,
     required DataSourceStatusManager statusManager,
@@ -61,8 +78,10 @@ final class DataSourceManager {
     _dataSourceFactories.addAll(factories);
   }
 
-  void identify(LDContext context, Completer<void> completer) {
+  void identify(LDContext context, Completer<void> completer,
+      {DataAvailability minimumDataAvailability = DataAvailability.cached}) {
     _identifyCompleter = completer;
+    _minimumDataAvailability = minimumDataAvailability;
     _activeContext = context;
 
     _setupConnection();
@@ -92,6 +111,19 @@ final class DataSourceManager {
     _activeDataSource = null;
   }
 
+  /// Resolves the pending identify, if any. Idempotent: only the first call
+  /// completes it.
+  void _maybeCompleteIdentify() {
+    final completer = _identifyCompleter;
+    if (completer == null) {
+      return;
+    }
+    if (!completer.isCompleted) {
+      completer.complete();
+    }
+    _identifyCompleter = null;
+  }
+
   DataSource? _createDataSource(FDv2ConnectionMode mode) {
     if (_activeContext != null) {
       if (_dataSourceFactories[mode] == null) {
@@ -126,7 +158,6 @@ final class DataSourceManager {
           case OfflineBackgroundDisabled():
             _statusManager.setBackgroundDisabled();
         }
-        return;
       case FDv2Streaming():
       case FDv2Polling():
       case FDv2Background():
@@ -146,22 +177,35 @@ final class DataSourceManager {
           var handled = await _dataSourceEventHandler.handleMessage(
               _activeContext!, event.type, event.data,
               environmentId: event.environmentId);
-          if (handled == MessageStatus.messageHandled &&
-              _identifyCompleter != null) {
-            if (_identifyCompleter!.isCompleted) {
-              _logger.error('Identify was already complete before receiving '
-                  'data. This could represent an issue with SDK logic. Please'
-                  'make a bug report if you encounter this situation.');
-            } else {
-              _identifyCompleter!.complete();
-            }
+          if (handled == MessageStatus.messageHandled) {
+            _maybeCompleteIdentify();
           }
-          // Only need to complete this the first time.
-          _identifyCompleter = null;
           return handled;
         case PayloadEvent():
-          // The FDv1 data sources this manager runs never produce FDv2
-          // payload events.
+          var handled = await _dataSourceEventHandler.handlePayload(
+              _activeContext!, event.changeSet,
+              environmentId: event.environmentId);
+          if (handled == MessageStatus.messageHandled) {
+            // Applying any change set from a live source marks it valid --
+            // including a no-change response, which restores valid after an
+            // interruption.
+            if (_activeConnectionMode is! FDv2Offline) {
+              _statusManager.setValid();
+            }
+            // A 'cached' identify resolves on any applied data; a 'fresh'
+            // identify waits for the orchestrator's InitializedEvent
+            // instead.
+            if (_minimumDataAvailability == DataAvailability.cached) {
+              _maybeCompleteIdentify();
+            }
+          }
+          return handled;
+        case InitializedEvent():
+          // Initialization is complete (network basis, initializer
+          // exhaustion, or the first synchronizer change set). Resolves a
+          // wait-for-network identify; a cached identify has usually resolved
+          // already on earlier data.
+          _maybeCompleteIdentify();
           return MessageStatus.messageHandled;
         case StatusEvent():
           if (_identifyCompleter != null && !_identifyCompleter!.isCompleted) {