Persist custom and OOTB log fields in state

bd-api/src/api.rs

@@ -587,6 +587,8 @@ impl Api {
       StateScope::FEATURE_FLAG => Some(bd_state::Scope::FeatureFlagExposure),
       StateScope::GLOBAL_STATE => Some(bd_state::Scope::GlobalState),
       StateScope::SYSTEM => Some(bd_state::Scope::System),
+      // Log-field scopes are owned by the SDK metadata collector. Server-pushed state updates
+      // must not overwrite their virtual matcher/extractor view independently of live metadata.
       StateScope::CUSTOM_FIELDS | StateScope::OOTB_FIELDS | StateScope::UNSPECIFIED => None,
     }
   }

bd-api/src/api_test.rs

@@ -55,6 +55,7 @@ use bd_proto::protos::client::api::{
 };
 use bd_proto::protos::logging::payload::LogType;
 use bd_proto::protos::logging::payload::data::Data_type;
+use bd_proto::protos::state::scope::StateScope;
 use bd_proto::protos::workflow::workflow::workflow::action::action_flush_buffers;
 use bd_runtime::runtime::{ConfigLoader, FeatureFlag};
 use bd_state::StateReader;
@@ -725,6 +726,12 @@ fn make_server_pushed_feature_flag_update(flag: &str, value: &str) -> ClientStat
   update
 }
 
+#[test]
+fn server_state_updates_exclude_sdk_owned_log_field_scopes() {
+  assert_eq!(Api::map_state_scope(StateScope::CUSTOM_FIELDS.into()), None);
+  assert_eq!(Api::map_state_scope(StateScope::OOTB_FIELDS.into()), None);
+}
+
 fn make_server_pushed_feature_flag_clear(flag: &str) -> ClientStateUpdate {
   let mut update = ClientStateUpdate::new();
   update.set_clear_client_state(client_state_update::ClearClientState {

bd-crash-handler/src/lib.rs

@@ -140,7 +140,7 @@ pub trait CrashReportHook: Send + Sync {
 #[derive(Clone)]
 pub struct Monitor {
   report_directory: PathBuf,
-  previous_run_state: bd_versioned_kv::ScopedMaps,
+  previous_run_state: Arc<bd_versioned_kv::ScopedMaps>,
   state: bd_state::Store,
   artifact_client: Arc<dyn bd_artifact_upload::Client>,
 
@@ -158,7 +158,7 @@ impl Monitor {
     session: Arc<bd_session::Strategy>,
     init_lifecycle: &InitLifecycleState,
     state: bd_state::Store,
-    previous_run_state: bd_versioned_kv::ScopedMaps,
+    previous_run_state: Arc<bd_versioned_kv::ScopedMaps>,
     emit_log: impl Fn(CrashLog) -> anyhow::Result<()> + Send + Sync + 'static,
     crash_report_hook: Option<Arc<dyn CrashReportHook>>,
   ) -> Self {

bd-crash-handler/src/monitor_test.rs

@@ -322,7 +322,7 @@ impl Setup {
       session,
       &InitLifecycleState::new(),
       (*state).clone(),
-      previous_run_state,
+      Arc::new(previous_run_state),
       emit_log,
       crash_report_hook,
     );

bd-log-filter/src/lib.rs

@@ -88,6 +88,17 @@ impl FilterChain {
     (Self { filters }, failures_count)
   }
 
+  /// Applies matching filters and their inline-log transforms.
+  ///
+  /// Matchers may read state-backed custom and OOTB fields through `state`, but transforms never
+  /// materialize, redact, replace, or remove those values. Later state-aware consumers resolve
+  /// the same state again, so transforms apply only to fields concretely present on `log`.
+  ///
+  /// TODO(snowp): Before eliding these fields from logs, preserve filter semantics with a per-log
+  /// state overlay. The overlay must carry rewritten and removed state values through every
+  /// downstream consumer, including buffer/workflow matching and tailing, without mutating the
+  /// global state. Snapshot uploads need the corresponding resolved values, not the current
+  /// state, because a snapshot can outlive the filter configuration that transformed its log.
   pub fn process(&self, log: &mut Log, state: &dyn bd_state::StateReader) {
     for filter in &self.filters {
       let fields_ref = FieldsRef::new(&log.fields, &log.matching_fields);