adapter: resolve scoped feature flags, gated, with create-time evaluation

The resolution layer that consumes the durable scoped overrides, behind the
`enable_scoped_system_parameters` dyncfg (off by default).

* The sync loop evaluates the `cluster` and `replica` LaunchDarkly contexts and
  records an override only when the scoped value differs from the environment-
  wide value, comparing in canonical encoding so bool spellings match. The whole
  evaluation is skipped while the gate is off, and the working copy cleared.
* Resolution boundaries: cluster-coherent overrides feed plan-time
  `OptimizerFeatureOverrides` (with a lenient bool decode so `on`/`off` cannot
  panic the optimizer); replica-local overrides feed the compute controller's
  per-replica dyncfg push.
* Create-time evaluation resolves a freshly created cluster or replica through a
  shared `SystemParameterFrontend` so new objects get their overrides without
  waiting for the next sync tick.
* Tests: canonical bool bridging, lenient optimizer decode, frontend context
  construction, the LaunchDarkly end-to-end cases, and workload-harness
  registration of the new gate.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: antiguru

misc/python/materialize/mzcompose/__init__.py

@@ -228,6 +228,11 @@ def get_variable_system_parameters(
             "true" if version >= MzVersion.parse_mz("v26.18.0-dev") else "false",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_scoped_system_parameters",
+            "false",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",

misc/python/materialize/parallel_workload/action.py

@@ -1518,6 +1518,7 @@ def __init__(
         )
         self.flags_with_values["enable_eager_delta_joins"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["enable_public_metrics_endpoint"] = BOOLEAN_FLAG_VALUES
+        self.flags_with_values["enable_scoped_system_parameters"] = BOOLEAN_FLAG_VALUES
         self.flags_with_values["persist_batch_structured_key_lower_len"] = [
             "0",
             "1",

src/adapter-types/src/dyncfgs.rs

@@ -267,6 +267,17 @@ pub const CATALOG_INFO_METRICS_RECONCILE_INTERVAL: Config<Duration> = Config::ne
     "How frequently to reconcile the catalog `*_info` metrics with the catalog. A zero duration disables reconciliation.",
 );
 
+/// Whether per-cluster and per-replica scoped system parameters are evaluated.
+/// Off by default: the parameter sync loop evaluates no cluster/replica
+/// contexts and resolution falls back to the environment-wide value everywhere
+/// (the pre-scoped behavior). Enabling it (e.g. from LaunchDarkly) turns on
+/// scoped evaluation without a deploy.
+pub const ENABLE_SCOPED_SYSTEM_PARAMETERS: Config<bool> = Config::new(
+    "enable_scoped_system_parameters",
+    false,
+    "Whether per-cluster and per-replica scoped system parameters are evaluated and applied.",
+);
+
 /// Adds the full set of all adapter `Config`s.
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
@@ -304,4 +315,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&ARRANGEMENT_SIZE_HISTORY_COLLECTION_INTERVAL)
         .add(&ARRANGEMENT_SIZE_HISTORY_RETENTION_PERIOD)
         .add(&CATALOG_INFO_METRICS_RECONCILE_INTERVAL)
+        .add(&ENABLE_SCOPED_SYSTEM_PARAMETERS)
 }

src/adapter/src/client.rs

@@ -56,6 +56,7 @@ use crate::command::{
     CatalogDump, CatalogSnapshot, Command, CopyFromStdinWriter, ExecuteResponse, Response,
     SASLChallengeResponse, SASLVerifyProofResponse, SuperuserAttribute,
 };
+use crate::config::{ScopedParameters, SystemParameterFrontend};
 use crate::coord::{Coordinator, ExecuteContextGuard};
 use crate::error::AdapterError;
 use crate::metrics::{self, Metrics};
@@ -555,6 +556,31 @@ Issue a SQL query to get started. Need help?
         rx.await.expect("coordinator unexpectedly gone")
     }
 
+    /// Returns a snapshot of the catalog.
+    pub async fn catalog_snapshot(&self) -> Arc<Catalog> {
+        let (tx, rx) = oneshot::channel();
+        self.send(Command::CatalogSnapshot { tx });
+        let CatalogSnapshot { catalog } = rx.await.expect("coordinator unexpectedly gone");
+        catalog
+    }
+
+    /// Replaces the scoped feature-flag overrides (the complete desired state).
+    /// Used by the system-parameter sync loop to reconcile the coordinator's
+    /// scoped-parameter working copy from continuous LaunchDarkly evaluation.
+    pub async fn update_scoped_system_parameters(&self, overrides: ScopedParameters) {
+        let (tx, rx) = oneshot::channel();
+        self.send(Command::UpdateScopedSystemParameters { overrides, tx });
+        let _ = rx.await;
+    }
+
+    /// Installs (or replaces) the shared system-parameter frontend on the
+    /// coordinator, letting the create-cluster / create-replica paths resolve a
+    /// new object's scoped overrides synchronously. Sent by the sync loop each
+    /// time it (re)initializes the frontend. Fire-and-forget.
+    pub fn install_scoped_system_parameter_frontend(&self, frontend: Arc<SystemParameterFrontend>) {
+        self.send(Command::InstallScopedSystemParameterFrontend { frontend });
+    }
+
     #[instrument(level = "debug")]
     pub(crate) fn send(&self, cmd: Command) {
         self.inner_cmd_tx
@@ -1320,6 +1346,8 @@ impl SessionClient {
                 | Command::PrivilegedCancelRequest { .. }
                 | Command::GetSystemVars { .. }
                 | Command::SetSystemVars { .. }
+                | Command::UpdateScopedSystemParameters { .. }
+                | Command::InstallScopedSystemParameterFrontend { .. }
                 | Command::Terminate { .. }
                 | Command::RetireExecute { .. }
                 | Command::CheckConsistency { .. }

src/adapter/src/command.rs

@@ -46,6 +46,7 @@ use tokio::sync::{mpsc, oneshot};
 use uuid::Uuid;
 
 use crate::catalog::Catalog;
+use crate::config::{ScopedParameters, SystemParameterFrontend};
 use crate::coord::appends::BuiltinTableAppendNotify;
 use crate::coord::consistency::CoordinatorInconsistencies;
 use crate::coord::peek::{PeekDataflowPlan, PeekResponseUnary};
@@ -168,6 +169,27 @@ pub enum Command {
         tx: oneshot::Sender<Result<(), AdapterError>>,
     },
 
+    /// Replace the scoped feature-flag overrides (the complete desired state).
+    /// Computed by the system-parameter sync loop from continuous LaunchDarkly
+    /// evaluation. The coordinator stores this working copy and reconciles it
+    /// into the per-scope resolution boundaries (the compute controller's
+    /// per-replica dyncfg layer for `replica`-scoped parameters). See the
+    /// scoped feature flags design.
+    UpdateScopedSystemParameters {
+        overrides: ScopedParameters,
+        tx: oneshot::Sender<()>,
+    },
+
+    /// Install (or replace) the shared system-parameter frontend on the
+    /// coordinator, so the create-cluster / create-replica paths can resolve a
+    /// new object's scoped overrides synchronously, before the controller
+    /// installs it or its first dataflow is planned, rather than waiting for
+    /// the next sync tick. Sent by the sync loop whenever it (re)initializes the
+    /// frontend. See the scoped feature flags design.
+    InstallScopedSystemParameterFrontend {
+        frontend: Arc<SystemParameterFrontend>,
+    },
+
     InjectAuditEvents {
         events: Vec<crate::catalog::InjectedAuditEvent>,
         conn_id: ConnectionId,
@@ -369,6 +391,8 @@ impl Command {
             | Command::Terminate { .. }
             | Command::GetSystemVars { .. }
             | Command::SetSystemVars { .. }
+            | Command::UpdateScopedSystemParameters { .. }
+            | Command::InstallScopedSystemParameterFrontend { .. }
             | Command::RetireExecute { .. }
             | Command::CheckConsistency { .. }
             | Command::Dump { .. }
@@ -407,6 +431,8 @@ impl Command {
             | Command::Terminate { .. }
             | Command::GetSystemVars { .. }
             | Command::SetSystemVars { .. }
+            | Command::UpdateScopedSystemParameters { .. }
+            | Command::InstallScopedSystemParameterFrontend { .. }
             | Command::RetireExecute { .. }
             | Command::CheckConsistency { .. }
             | Command::Dump { .. }

src/adapter/src/config.rs

@@ -25,8 +25,12 @@ mod params;
 mod sync;
 
 pub use backend::SystemParameterBackend;
-pub use frontend::SystemParameterFrontend;
+pub use frontend::{
+    ClusterEvalContext, ClusterScopeContext, ReplicaEvalContext, ReplicaScopeContext,
+    SystemParameterFrontend,
+};
 pub use params::{ModifiedParameter, SynchronizedParameters};
+pub(crate) use sync::evaluate_scoped_parameters;
 pub use sync::system_parameter_sync;
 
 /// Scoped (per-cluster and per-replica) system-parameter overrides, keyed by