diff --git a/.github/workflows/rust-test.yml b/.github/workflows/rust-test.yml
index ea9f216..8664c41 100644
--- a/.github/workflows/rust-test.yml
+++ b/.github/workflows/rust-test.yml
@@ -66,6 +66,12 @@ jobs:
           cargo test -p lance-context-wal --features lab --bin context-wal-lab
           cargo test -p lance-context-wal --features lab --test engine_cache
           cargo run -p lance-context-wal --features lab --bin context-wal-lab -- disk --directory "$RUNNER_TEMP/wal-disk-smoke" --payload-bytes 1024 --batch 4 --writers 2 --groups-per-writer 2 --rounds 8
+      - name: Optional Lance object-store metrics
+        run: |
+          cargo test -p lance-context-core --features lance-io-metrics --test lance_io_metrics
+          cargo test -p lance-context-core --all-features --lib latency_histograms_carry_only_bounded_labels
+          cargo check -p lance-context-server --features lance-io-metrics
+          cargo check -p lance-context-core --no-default-features
       - name: WAL placement, admission, record, contract and IPC decoding tests
         run: |
           cargo test -p lance-context-wal --features placement,admission,ipc,record,arrow-contract,vectored-io --lib
diff --git a/Cargo.lock b/Cargo.lock
index 546928e..ce284ad 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -4531,6 +4531,7 @@ dependencies = [
  "lance-core",
  "lance-namespace",
  "log",
+ "metrics",
  "moka",
  "object_store 0.14.2",
  "object_store_opendal",
diff --git a/crates/lance-context-core/Cargo.toml b/crates/lance-context-core/Cargo.toml
index 2780f4e..7abe169 100644
--- a/crates/lance-context-core/Cargo.toml
+++ b/crates/lance-context-core/Cargo.toml
@@ -16,6 +16,8 @@ default = ["metrics"]
 # Optional so downstream consumers embedding this library are not forced to take
 # the `metrics` dependency; call sites compile to nothing when disabled.
 metrics = ["dep:metrics"]
+# Opt-in storage diagnostics; share the application's existing metrics recorder.
+lance-io-metrics = ["metrics", "lance/metrics"]
 
 [dependencies]
 async-trait = "0.1.92"
diff --git a/crates/lance-context-core/README.md b/crates/lance-context-core/README.md
index a997321..cd06c5e 100644
--- a/crates/lance-context-core/README.md
+++ b/crates/lance-context-core/README.md
@@ -8,3 +8,28 @@ checks up to 1,024 IDs without fetching payloads or issuing a disk lookup per
 ID. Found rows are returned once in first-requested order, including flushed
 WAL updates. See [batch lookup](../../docs/benchmarks/batch-get.md) for the HTTP
 and Rust client APIs, visibility rules, and a reproducible benchmark.
+
+Enable the optional `lance-io-metrics` feature to record Lance 13 object-store
+request counts, bytes, durations, errors, in-flight requests and cloud retryable
+responses through the application's existing `metrics` recorder. Embedded users
+can call `metrics::describe_lance_io_metrics()` after installing their recorder.
+The REST server forwards this feature and registers descriptions automatically:
+
+```sh
+cargo build -p lance-context-server --features lance-io-metrics
+```
+
+Storage metrics then appear on the server's existing `/metrics` endpoint. Names
+start with `lance_object_store_`; request metrics have an `operation` label and
+use only the storage scheme as the default `base` label. Keep
+`LANCE_OBJECT_STORE_METRICS_LABEL=scheme` for bounded cardinality; Lance's `full`
+and `dataset` modes include storage identities and can create many more series.
+This feature is off by default. Cargo features are additive, so enabling it in
+one dependency enables Lance instrumentation throughout that binary.
+
+Request durations can overlap across uploads and include retries; their sum is
+neither elapsed ingestion time nor encoder CPU time. Local optimized writes
+measure the writer lifetime, including encoding between writes. Retryable HTTP
+response metrics cover native cloud stores, not every custom backend. These
+signals complement application phase timings; they do not alone establish
+network saturation or isolate encoding cost. See the [Lance metric catalog](https://docs.rs/lance/13.0.0/lance/metrics/index.html).
diff --git a/crates/lance-context-core/src/metrics.rs b/crates/lance-context-core/src/metrics.rs
index 0dffcf9..292059f 100644
--- a/crates/lance-context-core/src/metrics.rs
+++ b/crates/lance-context-core/src/metrics.rs
@@ -23,6 +23,13 @@
 //!    rate, so errors get a flat counter (1 series) and histograms measure the
 //!    success path only.
 
+/// Register Lance's object-store metric descriptions on the current recorder.
+/// Call after installing the application's recorder when `lance-io-metrics` is
+/// enabled. This does not install or replace a recorder. Lance's storage metrics
+/// use their own outcome and label conventions; scheme labels are the default.
+#[cfg(feature = "lance-io-metrics")]
+pub use lance_io::object_store::metrics::describe_metrics as describe_lance_io_metrics;
+
 /// Latency of one [`crate::RolloutStore::add`] — the durable WAL append only.
 /// Unlabelled: failures are counted by [`ROLLOUT_ADD_ERRORS`] instead.
 pub const ROLLOUT_ADD_DURATION: &str = "rollout_add_duration_seconds";
diff --git a/crates/lance-context-core/src/rollout_store.rs b/crates/lance-context-core/src/rollout_store.rs
index 40aafdc..fc46983 100644
--- a/crates/lance-context-core/src/rollout_store.rs
+++ b/crates/lance-context-core/src/rollout_store.rs
@@ -6941,10 +6941,37 @@ mod tests {
             ),
         ];
 
+        // Enabling Lance I/O metrics adds its own histogram to this recorder.
+        // Keep its label policy separate: application histograms must not gain
+        // operation/base labels just because a dependency uses them.
+        let lance_io_allowed: &[(&str, &[&str])] = &[
+            ("base", &["file"]),
+            (
+                "operation",
+                &[
+                    "get",
+                    "head",
+                    "put",
+                    "put_part",
+                    "complete_multipart",
+                    "abort_multipart",
+                    "list",
+                    "copy",
+                    "rename",
+                    "delete",
+                ],
+            ),
+        ];
+
         for (key, _, _, _) in snapshotter.snapshot().into_vec() {
             if key.kind() != MetricKind::Histogram {
                 continue;
             }
+            let allowed = if key.key().name() == "lance_object_store_request_duration_seconds" {
+                lance_io_allowed
+            } else {
+                allowed
+            };
             for label in key.key().labels() {
                 let (k, v) = (label.key(), label.value());
                 let allowed_values = allowed
diff --git a/crates/lance-context-core/src/store_base.rs b/crates/lance-context-core/src/store_base.rs
index 2c3daa0..679fa6c 100644
--- a/crates/lance-context-core/src/store_base.rs
+++ b/crates/lance-context-core/src/store_base.rs
@@ -747,16 +747,15 @@ impl StorageBase {
         // magnitude, and `noop` is by far the most common, so a blended histogram
         // would be dominated by near-zero samples. Failures are counted, not timed.
         match &result {
-            Ok(outcome) => {
-                let label = match outcome {
-                    FlushOutcome::Sealed => "sealed",
-                    FlushOutcome::Noop => "noop",
-                    FlushOutcome::Fenced => "fenced",
-                };
+            Ok(_outcome) => {
                 observe_duration!(
                     crate::metrics::ROLLOUT_FLUSH_DURATION,
                     timer_elapsed!(started),
-                    "outcome" => label,
+                    "outcome" => match _outcome {
+                        FlushOutcome::Sealed => "sealed",
+                        FlushOutcome::Noop => "noop",
+                        FlushOutcome::Fenced => "fenced",
+                    },
                 );
             }
             Err(_) => count!(crate::metrics::ROLLOUT_FLUSH_ERRORS),
@@ -2678,14 +2677,16 @@ pub(crate) fn align_batch_to_schema(
 /// is: a single store answered 503 for five hours today before anyone
 /// noticed. The label is the dataset directory name, bounded by the number
 /// of stores a worker touches.
-fn note_read_refused(uri: &str) {
-    let store = uri
-        .trim_end_matches('/')
-        .rsplit('/')
-        .next()
-        .unwrap_or("unknown")
-        .to_string();
-    metrics::counter!("rollout_reads_refused_pending_total", "store" => store).increment(1);
+fn note_read_refused(_uri: &str) {
+    count!(
+        "rollout_reads_refused_pending_total",
+        "store" => _uri
+            .trim_end_matches('/')
+            .rsplit('/')
+            .next()
+            .unwrap_or("unknown")
+            .to_string(),
+    );
 }
 
 /// Derive the MemWAL shard UUID a server instance writes to from its stable
diff --git a/crates/lance-context-core/tests/lance_io_metrics.rs b/crates/lance-context-core/tests/lance_io_metrics.rs
new file mode 100644
index 0000000..f5a42c0
--- /dev/null
+++ b/crates/lance-context-core/tests/lance_io_metrics.rs
@@ -0,0 +1,111 @@
+#![recursion_limit = "256"]
+#![cfg(feature = "lance-io-metrics")]
+
+use lance_context_core::{trace_schema, GenericStore, GenericStoreOptions};
+use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+use serde_json::json;
+
+/// This integration-test executable owns its recorder, so Lance's background
+/// tasks use the same recorder without depending on thread-local propagation.
+#[tokio::test]
+async fn real_lance_write_and_reopen_emit_storage_metrics() {
+    // One test in a separate process; set the label policy before opening stores.
+    std::env::set_var("LANCE_OBJECT_STORE_METRICS_LABEL", "scheme");
+    let recorder = DebuggingRecorder::new();
+    let snapshotter = recorder.snapshotter();
+    metrics::set_global_recorder(recorder).unwrap();
+    lance_context_core::metrics::describe_lance_io_metrics();
+
+    let dir = tempfile::tempdir().unwrap();
+    let uri = dir.path().join("records.lance");
+    let uri = uri.to_str().unwrap();
+    let mut store = GenericStore::open(
+        uri,
+        trace_schema(),
+        GenericStoreOptions {
+            seal_on_add: true,
+            ..Default::default()
+        },
+    )
+    .await
+    .unwrap();
+    let row = json!({"id":"record", "session_id":"session", "turn_id":0,
+        "role":"assistant", "content_type":"text/plain", "content":"retained payload"})
+    .as_object()
+    .unwrap()
+    .clone();
+    store.add(std::slice::from_ref(&row)).await.unwrap();
+    store.cleanup_wal().await.unwrap();
+    store.close().await.unwrap();
+    drop(store);
+    let store = GenericStore::open_existing(uri, Default::default())
+        .await
+        .unwrap();
+    let rows = store.list(None, None).await.unwrap();
+    assert_eq!(rows.len(), 1);
+    assert_eq!(rows[0]["content"], row["content"]);
+
+    let mut read_bytes = 0;
+    let mut written_bytes = 0;
+    let mut durations = 0;
+    let mut requests = 0;
+    for (key, _, _, value) in snapshotter.snapshot().into_vec() {
+        if !key.key().name().starts_with("lance_object_store_") {
+            continue;
+        }
+        let labels: Vec<_> = key.key().labels().collect();
+        assert_eq!(
+            labels.len(),
+            2,
+            "local I/O exposes only operation and scheme"
+        );
+        for label in &labels {
+            match label.key() {
+                "base" => assert_eq!(label.value(), "file"),
+                "operation" => assert!(matches!(
+                    label.value(),
+                    "get"
+                        | "head"
+                        | "put"
+                        | "put_part"
+                        | "complete_multipart"
+                        | "abort_multipart"
+                        | "list"
+                        | "copy"
+                        | "rename"
+                        | "delete"
+                )),
+                key => panic!("unexpected local I/O label {key}"),
+            }
+        }
+        let operation = labels
+            .iter()
+            .find(|label| label.key() == "operation")
+            .map(|label| label.value());
+        match (key.key().name(), value) {
+            ("lance_object_store_request_bytes_total", DebugValue::Counter(bytes)) => {
+                match operation {
+                    Some("get") => read_bytes += bytes,
+                    Some("put" | "put_part") => written_bytes += bytes,
+                    _ => {}
+                }
+            }
+            ("lance_object_store_requests_total", DebugValue::Counter(count)) => {
+                requests += count;
+            }
+            ("lance_object_store_request_duration_seconds", DebugValue::Histogram(samples)) => {
+                assert!(samples
+                    .iter()
+                    .all(|s| s.into_inner().is_finite() && s.into_inner() >= 0.0));
+                durations += samples.len();
+            }
+            _ => {}
+        }
+    }
+    assert!(read_bytes > 0, "reopen/read must record storage bytes");
+    assert!(
+        written_bytes > 0,
+        "actual Lance writes must record storage bytes"
+    );
+    assert!(requests > 0 && durations > 0);
+}
diff --git a/crates/lance-context-server/Cargo.toml b/crates/lance-context-server/Cargo.toml
index a3f6b07..f2077eb 100644
--- a/crates/lance-context-server/Cargo.toml
+++ b/crates/lance-context-server/Cargo.toml
@@ -12,6 +12,9 @@ keywords = ["context", "lance", "server", "api"]
 name = "lance-context-server"
 path = "src/main.rs"
 
+[features]
+lance-io-metrics = ["lance-context-core/lance-io-metrics"]
+
 [dependencies]
 lance-context-merge = { version = "0.1.0", path = "../lance-context-merge" }
 lance-context-core = { version = "0.6.7", path = "../lance-context-core" }
diff --git a/crates/lance-context-server/src/main.rs b/crates/lance-context-server/src/main.rs
index 5e8a5af..991b582 100644
--- a/crates/lance-context-server/src/main.rs
+++ b/crates/lance-context-server/src/main.rs
@@ -75,6 +75,8 @@ async fn main() {
 
     // Install the Prometheus recorder once, before any metrics are emitted.
     let metrics_handle = lance_context_metrics::install_recorder();
+    #[cfg(feature = "lance-io-metrics")]
+    lance_context_core::metrics::describe_lance_io_metrics();
 
     let app = routes::router()
         .with_state(state.clone())