feat(miner): observe structured-body size via a histogram (RFC 0037 §3.2)

crates/ourios-miner/src/cluster.rs

@@ -1972,6 +1972,13 @@ impl MinerCluster {
         // rather than swallowing a `Result` we never inspect.
         let bytes = ourios_core::otlp::canonical::encode_any_value(any_value)
             .expect("RFC 0005 §3.3 encoder is infallible for any spec-compliant AnyValue");
+        // RFC 0037 §3.2 (hazard #2 guard): observe the canonical-JSON body
+        // size before `bytes` is moved into the record. Structured bodies are
+        // never capped, so this histogram is the only guard against oversized
+        // payloads — it makes the size visible per service without discarding
+        // the operator's payload.
+        self.metrics
+            .record_structured_body_bytes(&record.tenant_id, service, bytes.len() as u64);
         let mut rec = Self::record_envelope(record, BodyKind::Structured);
         rec.template_id = template_id;
         rec.template_version = 1;
@@ -2459,7 +2466,7 @@ fn positions_to_u16(positions: &[usize]) -> Vec<u16> {
 mod tests {
     use super::*;
     use ourios_core::audit::SharedAuditSink;
-    use ourios_core::otlp::{AnyValue, any_value::Value as AvValue};
+    use ourios_core::otlp::{AnyValue, ArrayValue, any_value::Value as AvValue};
     use ourios_core::record::SharedRecordSink;
     use proptest::prelude::*;
 
@@ -2570,6 +2577,51 @@ mod tests {
         }
     }
 
+    /// RFC0037.3 (unit) — the structured-body branch retains the body's
+    /// canonical JSON byte-for-byte and never flags it lossy (§3.2 fidelity),
+    /// colocated with `ingest_structured`. The per-service metric emission is
+    /// covered end-to-end in `tests/rfc0037_structured_body.rs`.
+    #[test]
+    fn rfc0037_3_structured_body_retained_byte_for_byte() {
+        let tenant = TenantId::new("t");
+        let sink = SharedRecordSink::new();
+        let mut cluster =
+            MinerCluster::new(MinerConfig::default()).with_record_sink(Box::new(sink.clone()));
+
+        let body_av = AnyValue {
+            value: Some(AvValue::ArrayValue(ArrayValue {
+                values: vec![
+                    AnyValue {
+                        value: Some(AvValue::StringValue("user turn".to_string())),
+                    },
+                    AnyValue {
+                        value: Some(AvValue::StringValue("assistant turn".to_string())),
+                    },
+                ],
+            })),
+        };
+        let expected = String::from_utf8(
+            ourios_core::otlp::canonical::encode_any_value(&body_av)
+                .expect("canonical encode is infallible"),
+        )
+        .expect("canonical JSON is UTF-8");
+
+        let mut record = structured_record(&tenant, 9, Some("lib.agent"));
+        record.event_name = Some("gen_ai.client.inference.operation.details".to_string());
+        record.body = Some(Body::Structured(body_av));
+        cluster.ingest(&record);
+
+        let mined = sink.drain();
+        assert_eq!(mined.len(), 1);
+        assert_eq!(mined[0].body_kind, BodyKind::Structured);
+        assert_eq!(
+            mined[0].body.as_deref(),
+            Some(expected.as_str()),
+            "the structured body is retained as canonical JSON, byte-for-byte"
+        );
+        assert!(!mined[0].lossy_flag, "a structured body is never lossy");
+    }
+
     /// Test helper — build a cluster wired to a [`SharedAuditSink`]
     /// and return both so the test can inspect emissions.
     fn cluster_with_observable_sink() -> (MinerCluster, SharedAuditSink) {

crates/ourios-miner/src/metrics.rs

@@ -310,6 +310,7 @@ pub(crate) struct MinerMetrics {
     template_version_changes_total: Counter<u64>,
     confidence: Histogram<f64>,
     miner_duration: Histogram<f64>,
+    structured_body_size: Histogram<u64>,
     /// The observable gauges are held for the [`MinerMetrics`]'s
     /// lifetime so their collection callbacks stay registered with
     /// the meter — dropping a handle deregisters its callback, after
@@ -363,6 +364,10 @@ impl MinerMetrics {
             .f64_histogram(semconv::OURIOS_MINER_DURATION)
             .with_unit("s")
             .build();
+        let structured_body_size = meter
+            .u64_histogram(semconv::OURIOS_MINER_STRUCTURED_BODY_SIZE)
+            .with_unit("By")
+            .build();
 
         let observable_gauges = Self::register_observable_gauges(&meter, &state);
 
@@ -374,6 +379,7 @@ impl MinerMetrics {
             template_version_changes_total,
             confidence,
             miner_duration,
+            structured_body_size,
             _observable_gauges: observable_gauges,
         }
     }
@@ -545,6 +551,23 @@ impl MinerMetrics {
         );
     }
 
+    /// Observe one structured (non-string) body's canonical-JSON byte
+    /// length (RFC 0037 §3.2, `ourios.miner.structured_body.size`).
+    /// Structured bodies are retained whole and never capped; this
+    /// histogram is the hazard-#2 guard — an operator watches its
+    /// per-`(tenant, service)` distribution to spot services emitting
+    /// oversized payloads (the fix belongs at the emitter, not a
+    /// store-side truncation).
+    pub(crate) fn record_structured_body_bytes(
+        &self,
+        tenant: &TenantId,
+        service: Option<&str>,
+        bytes: u64,
+    ) {
+        self.structured_body_size
+            .record(bytes, &service_attrs(tenant, service));
+    }
+
     /// Record `count` per-parameter overflow events on one line for
     /// `(tenant, service)`: bumps the `params_overflow_total`
     /// counter and the per-service overflow-line numerator (the

crates/ourios-miner/tests/rfc0037_structured_body.rs

@@ -0,0 +1,143 @@
+//! RFC 0037 §3.2 acceptance — **RFC0037.3** (unbounded fidelity +
+//! observability). A structured (non-string) log body is retained whole,
+//! never truncated (`lossy_flag = false`), and its canonical-JSON byte
+//! length is observed on the `ourios.miner.structured_body.size` histogram,
+//! dimensioned by service. This is the hazard-#2 guard: no cap, observation
+//! instead.
+
+use ourios_config::MinerConfig;
+use ourios_core::otlp::{
+    AnyValue, ArrayValue, Body, KeyValue as OtlpKeyValue, OtlpLogRecord, any_value,
+};
+use ourios_core::record::SharedRecordSink;
+use ourios_core::tenant::TenantId;
+use ourios_miner::cluster::MinerCluster;
+
+fn string_av(s: &str) -> AnyValue {
+    AnyValue {
+        value: Some(any_value::Value::StringValue(s.to_string())),
+    }
+}
+
+fn service_attrs(service: &str) -> Vec<OtlpKeyValue> {
+    vec![OtlpKeyValue {
+        key: "service.name".to_string(),
+        value: Some(string_av(service)),
+        ..Default::default()
+    }]
+}
+
+/// A large structured body: an array of `n` string elements, canonically
+/// encoding to many kilobytes — the `gen_ai.input.messages` shape at scale.
+fn big_structured_body(n: usize) -> AnyValue {
+    let values: Vec<AnyValue> = (0..n)
+        .map(|i| {
+            string_av(&format!(
+                "chat message part number {i} carrying several words of content"
+            ))
+        })
+        .collect();
+    AnyValue {
+        value: Some(any_value::Value::ArrayValue(ArrayValue { values })),
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+async fn rfc0037_3_structured_body_unbounded_fidelity_and_observability() {
+    use opentelemetry_sdk::metrics::data::{
+        AggregatedMetrics, MetricData, ResourceMetrics, ScopeMetrics,
+    };
+
+    let (guard, exporter) = ourios_telemetry::init_in_memory("ourios-test");
+
+    let sink = SharedRecordSink::new();
+    let mut cluster =
+        MinerCluster::new(MinerConfig::default()).with_record_sink(Box::new(sink.clone()));
+
+    let tenant = TenantId::new("genai-tenant");
+    let body_av = big_structured_body(200);
+    // The exact canonical JSON the record must retain, and its byte length —
+    // what the metric must observe.
+    let expected_body = String::from_utf8(
+        ourios_core::otlp::canonical::encode_any_value(&body_av)
+            .expect("canonical encode is infallible"),
+    )
+    .expect("canonical JSON is UTF-8");
+    let expected_bytes = expected_body.len() as u64;
+
+    let record = OtlpLogRecord {
+        tenant_id: tenant.clone(),
+        severity_number: 9,
+        scope_name: Some("lib.agent".to_string()),
+        event_name: Some("gen_ai.client.inference.operation.details".to_string()),
+        resource_attributes: service_attrs("checkout"),
+        body: Some(Body::Structured(body_av)),
+        ..Default::default()
+    };
+    cluster.ingest(&record);
+    guard.force_flush().expect("force_flush succeeds");
+
+    // Fidelity — the structured body is retained whole (byte-for-byte), never
+    // truncated, and never flagged lossy.
+    let mined = sink.drain();
+    assert_eq!(mined.len(), 1, "one record emitted");
+    let rec = &mined[0];
+    assert_eq!(
+        rec.body.as_deref(),
+        Some(expected_body.as_str()),
+        "the structured body must be retained byte-for-byte, never truncated"
+    );
+    assert!(
+        !rec.lossy_flag,
+        "a structured body is never lossy (RFC 0001 §6.1)"
+    );
+    assert!(
+        expected_bytes > 8_000,
+        "sanity: the fixture body is genuinely large ({expected_bytes} B)"
+    );
+
+    // Observability — the histogram recorded that byte length under
+    // ourios.service = checkout.
+    let rms = exporter.get_finished_metrics().expect("metrics exported");
+    let data = rms
+        .iter()
+        .flat_map(ResourceMetrics::scope_metrics)
+        .flat_map(ScopeMetrics::metrics)
+        .find(|m| m.name() == ourios_semconv::OURIOS_MINER_STRUCTURED_BODY_SIZE)
+        .expect("structured_body.size missing from the exported stream")
+        .data();
+    let AggregatedMetrics::U64(MetricData::Histogram(hist)) = data else {
+        panic!("structured_body.size should be a u64 histogram");
+    };
+    let point = hist
+        .data_points()
+        .find(|dp| {
+            dp.attributes().any(|kv| {
+                kv.key.as_str() == ourios_semconv::OURIOS_SERVICE && kv.value.as_str() == "checkout"
+            })
+        })
+        .expect("a data point carrying ourios.service = checkout");
+    // Validate the full required attribute set: the registry marks
+    // ourios.tenant `required` and ourios.service `recommended`, so both must
+    // ride the data point — asserting only service would still pass if tenant
+    // were accidentally dropped.
+    let has_attr = |key: &str, value: &str| {
+        point
+            .attributes()
+            .any(|kv| kv.key.as_str() == key && kv.value.as_str() == value)
+    };
+    assert!(
+        has_attr(ourios_semconv::OURIOS_TENANT, "genai-tenant"),
+        "the required ourios.tenant attribute must ride the data point"
+    );
+    assert!(
+        has_attr(ourios_semconv::OURIOS_SERVICE, "checkout"),
+        "the ourios.service attribute must ride the data point"
+    );
+    assert_eq!(point.count(), 1, "exactly one structured body observed");
+    assert_eq!(
+        point.sum(),
+        expected_bytes,
+        "histogram sum equals the canonical-JSON byte length"
+    );
+}

crates/ourios-semconv/src/lib.rs

@@ -109,6 +109,9 @@ pub const OURIOS_MINER_PARAMS_OVERFLOW_UTILIZATION: &str =
 /// `ourios.miner.parse_failures` (counter, unit `{failure}`).
 pub const OURIOS_MINER_PARSE_FAILURES: &str = "ourios.miner.parse_failures";
 
+/// `ourios.miner.structured_body.size` (histogram, unit `By`).
+pub const OURIOS_MINER_STRUCTURED_BODY_SIZE: &str = "ourios.miner.structured_body.size";
+
 /// `ourios.miner.template.count` (gauge, unit `{template}`).
 pub const OURIOS_MINER_TEMPLATE_COUNT: &str = "ourios.miner.template.count";
 

semconv/registry/metrics.yaml

@@ -164,6 +164,24 @@ groups:
       - ref: ourios.service
         requirement_level: recommended
 
+  - id: metric.ourios.miner.structured_body.size
+    type: metric
+    metric_name: ourios.miner.structured_body.size
+    stability: development
+    brief: >-
+      Canonical-JSON byte length of each structured (non-string) log body
+      (RFC 0037 §3.2). Structured bodies are retained whole and never capped;
+      this histogram is the hazard-#2 guard — an operator watches its
+      per-service distribution to spot services emitting oversized payloads,
+      the fix belonging at the emitter rather than a store-side truncation.
+    instrument: histogram
+    unit: "By"
+    attributes:
+      - ref: ourios.tenant
+        requirement_level: required
+      - ref: ourios.service
+        requirement_level: recommended
+
   - id: metric.ourios.miner.duration
     type: metric
     metric_name: ourios.miner.duration