fix(observed): correct enrichment and test-harness behavior

crates/fetch_winhttp_impl/src/telemetry.rs

@@ -185,8 +185,7 @@ mod tests {
             ExpectedEvent::new("fetch.winhttp.session.initialization.failure", Severity::Error)
                 .body("WinHTTP transport initialization failed")
                 .dimension("winhttp.error_code", 1234_u32)
-                .dimension("winhttp.operation", "assured_non_blocking_callbacks")
-                .log(),
+                .dimension("winhttp.operation", "assured_non_blocking_callbacks"),
         );
         assert_eq!(
             events[1],
@@ -196,7 +195,6 @@ mod tests {
             events[2],
             ExpectedEvent::new("fetch.winhttp.request.error", Severity::Error)
                 .body("WinHTTP transport request failed")
-                .log()
                 .metric(),
         );
         assert!(events[1].dimensions().is_empty());
@@ -216,7 +214,6 @@ mod tests {
                 .body("WinHTTP transport request failed")
                 .dimension("winhttp.connect.duration", 0.25_f64)
                 .dimension("winhttp.connection.fresh", true)
-                .log()
                 .metric(),
         );
 

crates/fetch_winhttp_impl/src/transport.rs

@@ -483,7 +483,6 @@ mod tests {
                     .body("WinHTTP transport request failed")
                     .dimension("winhttp.connect.duration", 0.25_f64)
                     .dimension("winhttp.connection.fresh", true)
-                    .log()
                     .metric(),
             ]
         );

crates/observed/benches/observed_benchmarks.rs

@@ -7,7 +7,7 @@
 //! - Full emit pipeline (event -> log record via `OTel` provider)
 //! - Enrichment resolution (context lookup + Vec building)
 //! - Metric dimension building
-//! - Context operations (enrich)
+//! - Sink operations (construction, emit through a no-op sink)
 //!
 //! Run with:
 //! ```sh
@@ -277,11 +277,11 @@ fn entrypoint(c: &mut Criterion) {
         group.finish();
     }
 
-    // --- Context operation benchmarks ---
+    // --- Sink operation benchmarks ---
     {
-        let mut group = c.benchmark_group("emit_context");
-        bench_attach_emitter(&mut group, &allocs, &time);
-        bench_emit_event_direct(&mut group, &allocs, &time);
+        let mut group = c.benchmark_group("sink_operations");
+        bench_construct_processor_free_sink(&mut group, &allocs, &time);
+        bench_emit_to_noop_sink(&mut group, &allocs, &time);
         group.finish();
     }
 
@@ -468,11 +468,15 @@ fn bench_enrich_push_pop(group: &mut BenchmarkGroup<'_, WallTime>, allocs: &allo
 }
 
 // ---------------------------------------------------------------------------
-// Context operation benchmarks
+// Sink operation benchmarks
 // ---------------------------------------------------------------------------
 
-fn bench_attach_emitter(group: &mut BenchmarkGroup<'_, WallTime>, allocs: &alloc_tracker::Session, time: &all_the_time::Session) {
-    const ID: &str = "attach_emitter";
+fn bench_construct_processor_free_sink(
+    group: &mut BenchmarkGroup<'_, WallTime>,
+    allocs: &alloc_tracker::Session,
+    time: &all_the_time::Session,
+) {
+    const ID: &str = "construct_processor_free_sink";
     const EMPTY_PROCESSORS: Vec<Arc<dyn observed::processing::EventProcessor>> = Vec::new();
 
     bench_with_tracking(group, allocs, time, ID, || {
@@ -481,11 +485,10 @@ fn bench_attach_emitter(group: &mut BenchmarkGroup<'_, WallTime>, allocs: &alloc
     });
 }
 
-fn bench_emit_event_direct(group: &mut BenchmarkGroup<'_, WallTime>, allocs: &alloc_tracker::Session, time: &all_the_time::Session) {
-    const ID: &str = "emit_event_no_emitter";
+fn bench_emit_to_noop_sink(group: &mut BenchmarkGroup<'_, WallTime>, allocs: &alloc_tracker::Session, time: &all_the_time::Session) {
+    const ID: &str = "emit_to_noop_sink";
 
-    // Benchmark the emit path when no sink is registered - measures the
-    // cost of the context lookup + early return.
+    // Benchmark the emit path through an explicit no-op sink.
     let noop = Sink::noop();
     bench_with_tracking(group, allocs, time, ID, || {
         observed::emit!(&noop, SimpleLogEvent { status: 200, retries: 0 });

crates/observed/examples/layered_app/token_issuer.rs

@@ -14,7 +14,7 @@ use observed::{Enrichment, Sink, emit, event};
 
 use crate::taxonomy::MicrosoftEnterpriseDataTaxonomy;
 
-const DC: DataClass = DataClass::new("microsoft", "PublicNonPersonalData");
+const DC: DataClass = MicrosoftEnterpriseDataTaxonomy::PublicNonPersonalData.data_class();
 
 // ---------------------------------------------------------------------------
 // Classified newtypes
@@ -69,16 +69,11 @@ struct TokenValidated {
 
 /// A failed token validation attempt.
 ///
-/// Records each failure as an up-down counter metric (increments on failure).
+/// Records each failure as a monotonic counter metric.
 #[event("token.validation_failed")]
 #[warning("Token validation failed")]
-#[updown_counter(failure_count, name = "token.validation.failures")]
+#[counter(name = "token.validation.failures")]
 struct TokenValidationFailed {
-    /// Failure count - recorded as an up-down counter metric.
-    // TODO: replace #[unredacted] with classified type once metric fields support non-numeric Values
-    #[unredacted]
-    failure_count: i64,
-
     /// Error code identifying the failure reason.
     #[data_class(DC)]
     error_code: i64,
@@ -129,7 +124,6 @@ pub(crate) fn validate_token(sink: &Sink, valid: bool) {
             emit!(
                 sink,
                 TokenValidationFailed {
-                    failure_count: 1,
                     error_code: 401, // expired
                 }
             );

crates/observed/examples/support/otel.rs

@@ -31,9 +31,6 @@ const CODE_FILE_PATH: &str = "code.file.path";
 /// `OTel` attribute key for the source line a call site came from.
 const CODE_LINE_NUMBER: &str = "code.line.number";
 
-/// `OTel` attribute key for the crate a call site came from.
-const CODE_NAMESPACE: &str = "code.namespace";
-
 /// Converts a [`Text`] into an `OTel` string, preserving the borrowed-versus-
 /// shared distinction so neither representation copies.
 ///
@@ -133,6 +130,11 @@ pub(crate) fn populate_log_record(record: &mut impl LogRecord, event: &EventView
         ControlFlow::Continue(())
     });
 
+    // File and line use stable OpenTelemetry code attributes. The emitting
+    // crate is deliberately not exported: `code.namespace` is deprecated with
+    // no standalone replacement, and its successor `code.function.name` needs a
+    // fully qualified function name that the event model does not capture.
+    // https://opentelemetry.io/docs/specs/semconv/registry/attributes/code/
     if let Some(file) = event.source_file() {
         record.add_attribute(opentelemetry::Key::from_static_str(CODE_FILE_PATH), AnyValue::String(file.into()));
     }
@@ -142,12 +144,5 @@ pub(crate) fn populate_log_record(record: &mut impl LogRecord, event: &EventView
             AnyValue::Int(i64::from(line)),
         );
     }
-    if let Some(crate_name) = event.source_crate() {
-        record.add_attribute(
-            opentelemetry::Key::from_static_str(CODE_NAMESPACE),
-            AnyValue::String(crate_name.into()),
-        );
-    }
-
     true
 }

crates/observed/src/enrichment/slot.rs

@@ -68,7 +68,15 @@ impl Slot {
     }
 
     /// Pushes entries onto the enrichment chain and returns a guard.
+    ///
+    /// An empty layer installs no chain node and records no restoration, so the
+    /// storage boundary holds for callers that reach it without passing through
+    /// [`Sink::push_enrichment`](crate::Sink).
     pub(crate) fn push(&self, entries: Arc<[EnrichmentEntry]>) -> Guard {
+        if entries.is_empty() {
+            return Guard::empty();
+        }
+
         let prev = {
             let cell = self.0.get_or(|| RefCell::new(None));
             let next = Arc::new(EnrichmentNode {
@@ -100,6 +108,10 @@ pub(crate) struct Guard {
 }
 
 impl Guard {
+    pub(crate) fn empty() -> Self {
+        Self { slots: SmallVec::new() }
+    }
+
     /// Flattens several guards into one. Each input guard is consumed and its
     /// `Drop` is disarmed (`mem::take` empties its slots), so the merged guard
     /// owns the restoration responsibility.
@@ -170,21 +182,30 @@ impl EnrichmentTransfer {
     /// Pushes an additional enrichment node onto every captured chain.
     /// Broadcast within the transfer's known slots; transfers with no
     /// captured slots are left unchanged.
+    ///
+    /// Emptiness is tested while the entries are still an owned `Vec`, so an
+    /// empty layer skips the shared-slice allocation as well as the chain
+    /// nodes. [`push_entries`](Self::push_entries) relies on this and on the
+    /// same check in [`push_for`](Self::push_for) for its non-empty input.
     pub(crate) fn push(&mut self, additional_enrichment: impl Enrichment) {
-        self.push_entries(&Arc::from(additional_enrichment.into_entries()));
+        let entries = additional_enrichment.into_entries();
+        if entries.is_empty() {
+            return;
+        }
+
+        self.push_entries(&Arc::from(entries));
     }
 
     /// Same as [`push`](Self::push), but marks every entry as targeted at
     /// `target`, so only that sink observes it. Mirrors the `with_target`
     /// mapping that `EnrichFutureExt::enrich_for` applies.
     pub(crate) fn push_for(&mut self, target: SinkId, additional_enrichment: impl Enrichment) {
-        self.push_entries(
-            &additional_enrichment
-                .into_entries()
-                .into_iter()
-                .map(|entry| entry.with_target(target))
-                .collect(),
-        );
+        let entries = additional_enrichment.into_entries();
+        if entries.is_empty() {
+            return;
+        }
+
+        self.push_entries(&entries.into_iter().map(|entry| entry.with_target(target)).collect());
     }
 
     /// Layers `entries` onto every captured chain as a single new node.
@@ -221,6 +242,14 @@ mod coverage_tests {
         }
     }
 
+    struct EmptyEnrichment;
+
+    impl Enrichment for EmptyEnrichment {
+        fn into_entries(self) -> Vec<EnrichmentEntry> {
+            Vec::new()
+        }
+    }
+
     /// Flattens a chain into its entries, innermost node first.
     fn chain_entries(head: &OptEnrichmentNode) -> Vec<EnrichmentEntry> {
         let mut entries = Vec::new();
@@ -255,6 +284,20 @@ mod coverage_tests {
         assert!(slot.current().is_none(), "dropping the guard must pop the pushed node");
     }
 
+    #[test]
+    fn empty_push_returns_noop_guard_without_touching_the_slot() {
+        let slot = Slot::new();
+        let empty_guard = slot.push(Arc::from(Vec::new()));
+        assert!(slot.current().is_none());
+
+        let real_guard = slot.push(Arc::from(vec![EnrichmentEntry::unclassified("k", 1_i64)]));
+        let merged = Guard::merge([empty_guard, real_guard]);
+        assert!(slot.current().is_some());
+
+        drop(merged);
+        assert!(slot.current().is_none());
+    }
+
     #[test]
     fn transfer_captures_a_slot_and_layers_further_enrichment_onto_it() {
         let slot = Slot::new();
@@ -281,6 +324,22 @@ mod coverage_tests {
         assert!(slot.current().is_none());
     }
 
+    #[test]
+    fn transfer_ignores_empty_enrichment_layers() {
+        let slot = Slot::new();
+        let mut transfer = EnrichmentTransfer::default();
+        transfer.add_slot(&slot);
+
+        transfer.push(EmptyEnrichment);
+        transfer.push_for(SinkId::new("target"), EmptyEnrichment);
+
+        let guard = transfer.apply();
+        assert!(slot.current().is_none());
+
+        drop(guard);
+        assert!(slot.current().is_none());
+    }
+
     #[test]
     fn debug_impls_and_relocate() {
         let mut slot = Slot::new();

crates/observed/src/sink/core.rs

@@ -283,9 +283,16 @@ impl Sink {
     ///
     /// This is the entry point used by the `.enrich(&sink, ...)` API in
     /// [`EnrichFutureExt`](crate::enrichment::EnrichFutureExt) and
-    /// [`EnrichFnExt`](crate::enrichment::EnrichFnExt). Composites with zero children
-    /// return a no-op guard.
+    /// [`EnrichFnExt`](crate::enrichment::EnrichFnExt).
+    ///
+    /// An empty layer returns early, which spares composite fan-out and every
+    /// downstream slot write. The entry slice is built by the caller, so this
+    /// does not avoid the slice's own allocation.
     pub(crate) fn push_enrichment(&self, entries: Arc<[EnrichmentEntry]>) -> Guard {
+        if entries.is_empty() {
+            return Guard::empty();
+        }
+
         match &*self.inner {
             SinkInner::Single(state) => state.enrichment.push(entries),
             SinkInner::Composite { children } => Guard::merge(children.iter().map(|c| c.enrichment.push(Arc::clone(&entries)))),

crates/observed/tests/otel_log_record.rs

@@ -257,10 +257,9 @@ fn metric_only_event_exports_no_log_record() {
 }
 
 /// The exported record name comes from the log signal rather than the event
-/// name, and the emitting crate is recorded as `code.namespace`. The two names
-/// differ here, so a mapping that fell back to the event name is caught.
+/// name, and source attributes omit deprecated crate namespace metadata.
 #[test]
-fn maps_log_name_and_source_crate() {
+fn maps_log_name_without_deprecated_source_crate_attribute() {
     let (sink, provider, exporter) = otel_emitter();
 
     emit!(sink, CacheLookup { hits: 3 });
@@ -272,10 +271,7 @@ fn maps_log_name_and_source_crate() {
     assert_eq!(record.event_name(), Some("cache.lookup.completed"));
 
     let attrs: Vec<_> = record.attributes_iter().map(|(k, v)| (k.clone(), v.clone())).collect();
-    assert!(matches!(
-        find_attr(&attrs, "code.namespace"),
-        Some(AnyValue::String(s)) if s.as_ref() == env!("CARGO_PKG_NAME")
-    ));
+    assert!(find_attr(&attrs, "code.namespace").is_none());
 }
 
 #[test]

crates/observed_testing/src/lib.rs

@@ -25,9 +25,7 @@
 //! emit!(sink, MyEvent { count: 42 });
 //!
 //! let event = processor.single_event();
-//! let expected = ExpectedEvent::new("my.event", Severity::Info)
-//!     .dimension("count", 42i64)
-//!     .log();
+//! let expected = ExpectedEvent::new("my.event", Severity::Info).dimension("count", 42i64);
 //! assert_eq!(event, expected);
 //! ```