diff --git a/docs/architecture.md b/docs/architecture.md
index c132910af..d7c4d7b79 100644
--- a/docs/architecture.md
+++ b/docs/architecture.md
@@ -342,6 +342,16 @@ Counters are incremented at the sites where resources are consumed:
 - **WAL latency**: Group commit fsync completion
 - **Maintenance CPU**: Lease acquisition / release
 - **Replication lag**: Raft follower log application
+- **Graph edge writes**: graph edge put handler, per applied edge in a single put or a batch put (`nodedb_graph_edges_written_total`, also the `graph_edges_written_total` row of `SHOW STATS`)
+- **Graph edge deletes**: graph edge delete handler, per live edge tombstoned in a single delete or a batch delete (`nodedb_graph_edges_deleted_total`, also the `graph_edges_deleted_total` row of `SHOW STATS`) — WIP, see below
+
+`nodedb_graph_edges_written_total` counts applied edge versions, so a put that rewrites a live edge increments it while the `nodedb_graph_edges` gauge stays flat. `nodedb_graph_edges_deleted_total` counts live edges removed, so a delete of an edge that was already absent writes a tombstone and increments neither counter — matching the affected count that statement reports.
+
+A cross-vShard edge is dual-homed: the same `EdgePut` or `EdgeDelete` runs on both endpoint homes. The write counter is recorded under `owns_logical_edge_stats` — the same ownership boundary `SHOW GRAPH STATS` counts logical edges with — so the source home counts the edge version once and the destination home adds nothing.
+
+`nodedb_graph_edges_deleted_total` is **WIP and over-reports on a multi-core cluster**. It counts each endpoint home that removes a live row. On one core the two dual-home participants share a store, so only the first finds the edge and the counter reads one; on two or more cores each home keeps its own copy, both find a live row, and one logical delete reads as two. Counting only the source home instead reads zero on one core, because the destination home performs the removal there. The deciding fact is the Control Plane's `single_home`, which the plan does not carry; distinguishing the cases needs a new `GraphOp::EdgeDelete` field threaded from `graph_ops/edge.rs`. Do not rely on this counter on a multi-core deployment until that lands.
+
+Both counters record client activity, so neither counts a WAL-replayed edge. Boot replay re-enters the same handlers to rebuild engine state, and the metrics are attached before replay runs; counting there would report pre-restart writes as new work on every restart. The handlers skip both counters while `CoreLoop::boot_replaying_wal()` holds, which covers the whole boot pass and is never set on a serving core. An online committed-redo apply reaches the same handlers and does count, because it is a real write.
 
 All metrics are dimensionalized by database and tenant to enable per-customer tracking and alerting.
 
diff --git a/nodedb/src/control/metrics/prometheus/engines.rs b/nodedb/src/control/metrics/prometheus/engines.rs
index b4b72aa55..dda405868 100644
--- a/nodedb/src/control/metrics/prometheus/engines.rs
+++ b/nodedb/src/control/metrics/prometheus/engines.rs
@@ -86,6 +86,20 @@ impl SystemMetrics {
             "Graph edges stored",
             self.graph_edges.load(Ordering::Relaxed),
         );
+        counter(
+            out,
+            "nodedb_graph_edges_written_total",
+            "Edge versions applied by a graph edge write",
+            self.graph_edges_written.load(Ordering::Relaxed),
+        );
+        counter(
+            out,
+            "nodedb_graph_edges_deleted_total",
+            "Live edges tombstoned by a graph edge delete (WIP: counted per \
+             endpoint home, so a cross-shard delete on a multi-core cluster \
+             reports one logical removal per home)",
+            self.graph_edges_deleted.load(Ordering::Relaxed),
+        );
 
         // ── Document engine ──
         counter(
diff --git a/nodedb/src/control/metrics/system/fields.rs b/nodedb/src/control/metrics/system/fields.rs
index a21da10da..21649b66f 100644
--- a/nodedb/src/control/metrics/system/fields.rs
+++ b/nodedb/src/control/metrics/system/fields.rs
@@ -77,6 +77,14 @@ pub struct SystemMetrics {
     pub graph_traversals: AtomicU64,
     pub graph_nodes: AtomicU64,
     pub graph_edges: AtomicU64,
+    /// Edge versions applied by a graph edge write (single and batch puts).
+    /// Distinct from [`Self::graph_edges`], a gauge of live edges: a put that
+    /// rewrites an edge leaves the gauge flat and still counts here.
+    pub graph_edges_written: AtomicU64,
+    /// Live edges tombstoned by a graph edge delete (single and batch deletes).
+    /// A delete of an edge that was already absent writes a tombstone but
+    /// removes no live edge, so it does not count.
+    pub graph_edges_deleted: AtomicU64,
 
     pub document_inserts: AtomicU64,
     pub document_reads: AtomicU64,
diff --git a/nodedb/src/control/metrics/system/record.rs b/nodedb/src/control/metrics/system/record.rs
index bb8d2d65e..514b36a1c 100644
--- a/nodedb/src/control/metrics/system/record.rs
+++ b/nodedb/src/control/metrics/system/record.rs
@@ -217,6 +217,23 @@ impl SystemMetrics {
         self.graph_edges.store(edges, Ordering::Relaxed);
     }
 
+    /// One edge version was applied by a graph edge write.
+    ///
+    /// Batch handlers call this once per edge as it is applied, never once per
+    /// batch: a batch that fails midway has already applied the edges before
+    /// the failure, and those writes are counted.
+    pub fn record_graph_edge_written(&self) {
+        self.graph_edges_written.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// One live edge was tombstoned by a graph edge delete.
+    ///
+    /// Callers pass the pre-image result: a delete whose pre-image was absent
+    /// removes no live edge, so it does not count.
+    pub fn record_graph_edge_deleted(&self) {
+        self.graph_edges_deleted.fetch_add(1, Ordering::Relaxed);
+    }
+
     // ── Document engine ──
 
     pub fn record_document_insert(&self) {
diff --git a/nodedb/src/control/server/shared/ddl/neutral/observability.rs b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
index 94cfa2a4f..b42a4c69f 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/observability.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/observability.rs
@@ -123,6 +123,14 @@ fn server_stats_rows(state: &SharedState) -> Vec<(String, String)> {
             "queries_graph".into(),
             sys.queries_graph.load(Ordering::Relaxed).to_string(),
         ));
+        rows.push((
+            "graph_edges_written_total".into(),
+            sys.graph_edges_written.load(Ordering::Relaxed).to_string(),
+        ));
+        rows.push((
+            "graph_edges_deleted_total".into(),
+            sys.graph_edges_deleted.load(Ordering::Relaxed).to_string(),
+        ));
         rows.push((
             "queries_document".into(),
             sys.queries_document.load(Ordering::Relaxed).to_string(),
@@ -254,3 +262,79 @@ pub fn show_memory(
         rows,
     ))])
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::control::metrics::SystemMetrics;
+    use std::sync::Arc;
+
+    /// Read one `(name, value)` row from a rendered row set.
+    fn row_value(rows: &[(String, String)], name: &str) -> Option<String> {
+        rows.iter()
+            .find(|(key, _)| key == name)
+            .map(|(_, value)| value.clone())
+    }
+
+    /// The graph write counters render on the Prometheus surface as counters
+    /// with their own help text, next to the live-edge gauge they must not be
+    /// confused with.
+    #[test]
+    fn the_graph_write_counters_render_on_the_prometheus_surface() {
+        let metrics = SystemMetrics::new();
+        metrics.record_graph_edge_written();
+        metrics.record_graph_edge_deleted();
+        let output = metrics.to_prometheus();
+
+        assert!(
+            output.contains("# TYPE nodedb_graph_edges_written_total counter"),
+            "the write counter renders as a counter: {output}"
+        );
+        assert!(
+            output.contains("nodedb_graph_edges_written_total 1"),
+            "the write counter carries its value: {output}"
+        );
+        assert!(
+            output.contains("# TYPE nodedb_graph_edges_deleted_total counter"),
+            "the delete counter renders as a counter: {output}"
+        );
+        assert!(
+            output.contains("nodedb_graph_edges_deleted_total 1"),
+            "the delete counter carries its value: {output}"
+        );
+    }
+
+    /// The row builder reads the counters off `SystemMetrics`, the same
+    /// instance the Data Plane records through, so `SHOW STATS` cannot drift
+    /// from `/metrics`. Rows carry the value as decimal text like every other
+    /// counter in the set.
+    #[test]
+    fn the_stats_rows_read_the_graph_write_counters_from_metrics() {
+        let directory = tempfile::tempdir().expect("tempdir");
+        let wal = Arc::new(
+            crate::wal::WalManager::open_for_testing(&directory.path().join("obs-stats.wal"))
+                .expect("open WAL"),
+        );
+        let (dispatcher, _data_sides) = crate::bridge::dispatch::Dispatcher::new(1, 64);
+        let state = SharedState::new(dispatcher, wal).expect("shared state");
+        let metrics = state
+            .system_metrics
+            .as_ref()
+            .expect("system metrics are wired")
+            .clone();
+        metrics.record_graph_edge_written();
+        metrics.record_graph_edge_written();
+        metrics.record_graph_edge_deleted();
+
+        let rows = server_stats_rows(&state);
+
+        assert_eq!(
+            row_value(&rows, "graph_edges_written_total").as_deref(),
+            Some("2"),
+        );
+        assert_eq!(
+            row_value(&rows, "graph_edges_deleted_total").as_deref(),
+            Some("1"),
+        );
+    }
+}
diff --git a/nodedb/src/data/executor/core_loop/open.rs b/nodedb/src/data/executor/core_loop/open.rs
index f13e93e8e..54123217b 100644
--- a/nodedb/src/data/executor/core_loop/open.rs
+++ b/nodedb/src/data/executor/core_loop/open.rs
@@ -197,6 +197,7 @@ impl CoreLoop {
             redo_apply:
                 crate::data::executor::handlers::transaction::redo_apply::RedoApplyState::new(),
             fail_stop: super::fail_stop::CoreFailStop::default(),
+            boot_replaying_wal: false,
         })
     }
 }
diff --git a/nodedb/src/data/executor/core_loop/state.rs b/nodedb/src/data/executor/core_loop/state.rs
index f948d83e2..75497e8c7 100644
--- a/nodedb/src/data/executor/core_loop/state.rs
+++ b/nodedb/src/data/executor/core_loop/state.rs
@@ -491,4 +491,9 @@ pub struct CoreLoop {
         crate::data::executor::handlers::transaction::redo_apply::RedoApplyState,
     /// Set once this core's state is unknown. It then refuses every request.
     pub(in crate::data::executor) fail_stop: super::fail_stop::CoreFailStop,
+
+    /// True while this core rebuilds engine state from the WAL at boot.
+    /// Boot replay re-enters the write handlers, so a handler that counts
+    /// client activity asks this first. See [`Self::boot_replaying_wal`].
+    pub(in crate::data::executor) boot_replaying_wal: bool,
 }
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs b/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
index 355a122fe..489766f58 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/delete.rs
@@ -12,7 +12,7 @@ use crate::types::TenantId;
 use crate::data::executor::handlers::transaction::undo::UndoEntry;
 use crate::data::executor::handlers::transaction::undo::edge_write::EdgeTarget;
 
-use super::shared::{EdgeDeleteParams, owns_logical_edge_stats};
+use super::shared::{EdgeDeleteParams, counts_logical_edge_delete, owns_logical_edge_stats};
 
 impl CoreLoop {
     pub(in crate::data::executor) fn execute_edge_delete(
@@ -129,6 +129,24 @@ impl CoreLoop {
                         properties: None,
                     },
                 );
+                // Only a live edge counts as deleted: the tombstone is
+                // written either way, but a delete of an absent edge removes
+                // nothing.
+                //
+                // WIP, not yet correct on a multi-core cluster. See
+                // `counts_logical_edge_delete` in `shared.rs`: on one core the
+                // two dual-home participants share a store and only the
+                // removal is observable, while on two or more cores each home
+                // holds its own copy and both see a live pre-image. No
+                // predicate available to this handler distinguishes the two,
+                // so this over-reports one logical delete as two on a cluster.
+                if existed
+                    && !self.boot_replaying_wal()
+                    && counts_logical_edge_delete(task, src_id, dst_id)
+                    && let Some(metrics) = &self.metrics
+                {
+                    metrics.record_graph_edge_deleted();
+                }
                 self.response_affected(task, u64::from(existed))
             }
             Err(e) => self.response_error(
@@ -143,14 +161,203 @@ impl CoreLoop {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
+
     use super::*;
     use crate::bridge::envelope::Status;
+    use crate::control::metrics::SystemMetrics;
     use crate::event::WriteOp;
     use crate::event::bus::create_event_bus_with_capacity;
     use nodedb_types::{RlsWriteCheck, Surrogate};
 
     use super::super::shared::EdgePutParams;
-    use super::super::shared::test_support::{affected_count, make_core, make_task_with_lsn};
+    use super::super::shared::test_support::{
+        affected_count, make_core, make_task_at_source, make_task_with_lsn,
+    };
+
+    /// A delete of a live edge counts one; a delete of an edge that was never
+    /// there writes a tombstone but removes nothing, so it counts zero — the
+    /// counter and the reported affected count stay in step.
+    #[test]
+    fn a_delete_counts_only_a_live_edge() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        let delete = |core: &mut super::CoreLoop, lsn: u64| {
+            let task = make_task_at_source(lsn, "a");
+            core.execute_edge_delete(
+                &task,
+                EdgeDeleteParams {
+                    tid: 1,
+                    collection: "knows",
+                    src_id: "a",
+                    label: "KNOWS",
+                    dst_id: "b",
+                    rls_write_check: &RlsWriteCheck::NoPolicyApplies,
+                },
+            )
+        };
+
+        let absent = delete(&mut h.core, 30);
+        assert_eq!(absent.status, Status::Ok);
+        assert_eq!(affected_count(&absent), 0);
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            0,
+            "an absent edge is nothing to remove"
+        );
+
+        let put_task = make_task_at_source(31, "a");
+        assert_eq!(
+            h.core
+                .execute_edge_put(
+                    &put_task,
+                    EdgePutParams {
+                        tid: 1,
+                        collection: "knows",
+                        src_id: "a",
+                        label: "KNOWS",
+                        dst_id: "b",
+                        properties: b"w=1",
+                        src_surrogate: Surrogate::new(1),
+                        dst_surrogate: Surrogate::new(2),
+                    },
+                )
+                .status,
+            Status::Ok
+        );
+
+        let live = delete(&mut h.core, 32);
+        assert_eq!(live.status, Status::Ok);
+        assert_eq!(affected_count(&live), 1);
+        assert_eq!(metrics.graph_edges_deleted.load(Ordering::Relaxed), 1);
+    }
+
+    /// A dual-homed delete is counted by the home that holds the live row,
+    /// whichever that is. This home is not the edge's source, and it still
+    /// counts the removal it performed; the other home then finds the edge
+    /// absent and counts nothing, so the pair counts one.
+    #[test]
+    fn a_delete_on_any_home_counts_the_live_row_it_removes() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+
+        // Seed the edge through its source home so it is live here too.
+        let put_task = make_task_at_source(33, "a");
+        assert_eq!(
+            h.core
+                .execute_edge_put(
+                    &put_task,
+                    EdgePutParams {
+                        tid: 1,
+                        collection: "knows",
+                        src_id: "a",
+                        label: "KNOWS",
+                        dst_id: "b",
+                        properties: b"w=1",
+                        src_surrogate: Surrogate::new(1),
+                        dst_surrogate: Surrogate::new(2),
+                    },
+                )
+                .status,
+            Status::Ok
+        );
+
+        // The task is homed on "b", the edge's destination home.
+        let task = make_task_at_source(34, "b");
+        let resp = h.core.execute_edge_delete(
+            &task,
+            EdgeDeleteParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "b",
+                rls_write_check: &RlsWriteCheck::NoPolicyApplies,
+            },
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(affected_count(&resp), 1, "a live row was removed");
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            1,
+            "the home that held the live row counts the removal"
+        );
+
+        // The other home, which now finds the edge absent, adds nothing.
+        let other = make_task_at_source(35, "a");
+        let resp = h.core.execute_edge_delete(
+            &other,
+            EdgeDeleteParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "b",
+                rls_write_check: &RlsWriteCheck::NoPolicyApplies,
+            },
+        );
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(affected_count(&resp), 0);
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            1,
+            "the second home removes nothing, so the logical edge counts once"
+        );
+    }
+
+    /// Replay re-enters this handler for edges a client removed before the
+    /// restart, so it must not count them as new removals.
+    #[test]
+    fn a_replayed_delete_counts_no_delete() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+
+        let put_task = make_task_at_source(35, "a");
+        assert_eq!(
+            h.core
+                .execute_edge_put(
+                    &put_task,
+                    EdgePutParams {
+                        tid: 1,
+                        collection: "knows",
+                        src_id: "a",
+                        label: "KNOWS",
+                        dst_id: "b",
+                        properties: b"w=1",
+                        src_surrogate: Surrogate::new(1),
+                        dst_surrogate: Surrogate::new(2),
+                    },
+                )
+                .status,
+            Status::Ok
+        );
+
+        h.core.boot_replaying_wal = true;
+        let task = make_task_at_source(36, "a");
+        let resp = h.core.execute_edge_delete(
+            &task,
+            EdgeDeleteParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "b",
+                rls_write_check: &RlsWriteCheck::NoPolicyApplies,
+            },
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            0,
+            "a replayed removal happened before the restart"
+        );
+    }
 
     /// Compiled filters equivalent to a `FOR WRITE` policy on `owner`.
     fn owner_write_check(owner: &str) -> Vec<u8> {
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs b/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
index 346c32fae..ef2e71465 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/delete_batch.rs
@@ -9,7 +9,7 @@ use crate::data::executor::core_loop::CoreLoop;
 use crate::data::executor::task::ExecutionTask;
 use crate::types::TenantId;
 
-use super::shared::owns_logical_edge_stats;
+use super::shared::{counts_logical_edge_delete, owns_logical_edge_stats};
 
 impl CoreLoop {
     /// Apply a batched edge delete in a single SPSC round-trip.
@@ -81,6 +81,18 @@ impl CoreLoop {
                 &edge.dst_id,
                 edge.collection.as_str(),
             );
+            // Counted per edge as it is tombstoned, never once for the batch:
+            // edges before a mid-batch failure are already removed.
+            //
+            // WIP, not yet correct on a multi-core cluster — see the note on
+            // `execute_edge_delete_with_undo`. Counted per home.
+            if existed
+                && !self.boot_replaying_wal()
+                && counts_logical_edge_delete(task, &edge.src_id, &edge.dst_id)
+                && let Some(metrics) = &self.metrics
+            {
+                metrics.record_graph_edge_deleted();
+            }
         }
         if !edges.is_empty() {
             self.checkpoint_coordinator
@@ -111,3 +123,135 @@ impl CoreLoop {
         self.response_affected(task, removed)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
+
+    use crate::bridge::envelope::Status;
+    use crate::control::metrics::SystemMetrics;
+    use crate::data::executor::handlers::graph::graph_edge_write::shared::EdgePutParams;
+    use crate::data::executor::handlers::graph::graph_edge_write::shared::test_support::{
+        affected_count, make_core, make_task_at_source,
+    };
+    use nodedb_physical::physical_plan::BatchEdge;
+    use nodedb_types::{DatabaseId, QualifiedCollection, Surrogate};
+
+    fn edge(src: &str, dst: &str) -> BatchEdge {
+        BatchEdge {
+            collection: QualifiedCollection::new(DatabaseId::DEFAULT, "knows"),
+            src_id: src.to_string(),
+            label: "KNOWS".to_string(),
+            dst_id: dst.to_string(),
+            src_surrogate: Surrogate::new(1),
+            dst_surrogate: Surrogate::new(2),
+        }
+    }
+
+    fn put(core: &mut crate::data::executor::core_loop::CoreLoop, src: &str, dst: &str, lsn: u64) {
+        assert_eq!(
+            core.execute_edge_put(
+                &make_task_at_source(lsn, src),
+                EdgePutParams {
+                    tid: 1,
+                    collection: "knows",
+                    src_id: src,
+                    label: "KNOWS",
+                    dst_id: dst,
+                    properties: b"w=1",
+                    src_surrogate: Surrogate::new(1),
+                    dst_surrogate: Surrogate::new(2),
+                },
+            )
+            .status,
+            Status::Ok
+        );
+    }
+
+    /// The batch counts the edges it actually removed: two live edges count
+    /// two, the absent third counts nothing, and the counter matches the
+    /// affected count the response carries.
+    #[test]
+    fn a_delete_batch_counts_only_the_live_edges() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        put(&mut h.core, "owner", "b", 40);
+        put(&mut h.core, "owner", "d", 41);
+
+        let batch = vec![edge("owner", "b"), edge("owner", "d"), edge("owner", "f")];
+        let resp = h
+            .core
+            .execute_edge_delete_batch(&make_task_at_source(42, "owner"), 1, &batch);
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(affected_count(&resp), 2);
+        assert_eq!(metrics.graph_edges_deleted.load(Ordering::Relaxed), 2);
+    }
+
+    /// A batch counts the live rows it removes on whichever home holds them,
+    /// and the other home of a dual-homed edge then finds them absent and adds
+    /// nothing — so the pair counts each logical edge once.
+    #[test]
+    fn a_delete_batch_counts_the_live_rows_any_home_removes() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        // Seed through the edge's source home so a live row exists.
+        put(&mut h.core, "owner", "b", 43);
+
+        // The task is homed on "b", not the edge's source home.
+        let resp = h.core.execute_edge_delete_batch(
+            &make_task_at_source(44, "b"),
+            1,
+            &[edge("owner", "b")],
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(affected_count(&resp), 1, "a live row was removed");
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            1,
+            "the home that held the live row counts the removal"
+        );
+
+        // The source home now finds the edge absent and adds nothing.
+        let resp = h.core.execute_edge_delete_batch(
+            &make_task_at_source(45, "owner"),
+            1,
+            &[edge("owner", "b")],
+        );
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(affected_count(&resp), 0);
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            1,
+            "the second home removes nothing, so the logical edge counts once"
+        );
+    }
+
+    /// Replay re-enters this handler for batches a client ran before the
+    /// restart, so it must not count them as new removals.
+    #[test]
+    fn a_replayed_delete_batch_counts_no_delete() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        put(&mut h.core, "owner", "b", 47);
+
+        h.core.boot_replaying_wal = true;
+        let resp = h.core.execute_edge_delete_batch(
+            &make_task_at_source(48, "owner"),
+            1,
+            &[edge("owner", "b")],
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(
+            metrics.graph_edges_deleted.load(Ordering::Relaxed),
+            0,
+            "a replayed removal happened before the restart"
+        );
+    }
+}
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/put.rs b/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
index 6aaaf6702..025193607 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/put.rs
@@ -33,7 +33,7 @@ impl CoreLoop {
     ///
     /// A put writes a new edge-store version and makes the CSR edge live with
     /// the weight in `properties`, so a successful put always reports exactly
-    /// one edge affected.
+    /// one edge affected and records exactly one edge version written.
     pub(in crate::data::executor) fn execute_edge_put_with_undo(
         &mut self,
         task: &ExecutionTask,
@@ -140,6 +140,15 @@ impl CoreLoop {
                                 properties: Some(properties),
                             },
                         );
+                        // A cross-vShard edge is dual-homed: the put runs on
+                        // both endpoint homes and only the source home owns
+                        // the logical edge, so only that home counts it.
+                        if !self.boot_replaying_wal()
+                            && owns_logical_edge_stats(task, src_id)
+                            && let Some(metrics) = &self.metrics
+                        {
+                            metrics.record_graph_edge_written();
+                        }
                         self.response_affected(task, 1)
                     }
                     Err(e) => self.response_error(
@@ -162,6 +171,8 @@ impl CoreLoop {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::atomic::Ordering;
+
     use super::*;
     use crate::bridge::envelope::Status;
     use crate::event::WriteOp;
@@ -169,7 +180,161 @@ mod tests {
     use crate::types::Lsn;
     use nodedb_types::Surrogate;
 
-    use super::super::shared::test_support::{affected_count, make_core, make_task_with_lsn};
+    use super::super::shared::test_support::{
+        affected_count, make_core, make_task_at_source, make_task_with_lsn,
+    };
+
+    /// A put records one edge version written, even when it rewrites an edge
+    /// that is already live: the gauge of live edges stays flat, so the write
+    /// count is the only place the second put shows up.
+    #[test]
+    fn a_put_counts_one_edge_version_per_write() {
+        use std::sync::Arc;
+
+        use crate::control::metrics::SystemMetrics;
+
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        let task = make_task_at_source(5, "a");
+        let params = || EdgePutParams {
+            tid: 1,
+            collection: "knows",
+            src_id: "a",
+            label: "KNOWS",
+            dst_id: "b",
+            properties: b"w=1",
+            src_surrogate: Surrogate::new(1),
+            dst_surrogate: Surrogate::new(2),
+        };
+
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "the core starts with a clean counter"
+        );
+
+        let first = h.core.execute_edge_put(&task, params());
+        assert_eq!(first.status, Status::Ok);
+        assert_eq!(metrics.graph_edges_written.load(Ordering::Relaxed), 1);
+
+        // The same edge again: a new version, still one write.
+        let second = h.core.execute_edge_put(&task, params());
+        assert_eq!(second.status, Status::Ok);
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            2,
+            "a rewrite is a write even though no new live edge appears"
+        );
+    }
+
+    /// A put refused before any version is written must not count.
+    #[test]
+    fn a_refused_put_counts_no_write() {
+        use std::sync::Arc;
+
+        use crate::control::metrics::SystemMetrics;
+
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        h.core
+            .mark_node_deleted(crate::types::DatabaseId::DEFAULT.as_u64(), 1, "gone");
+
+        let resp = h.core.execute_edge_put(
+            &make_task_at_source(6, "a"),
+            EdgePutParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "gone",
+                properties: b"w=1",
+                src_surrogate: Surrogate::new(1),
+                dst_surrogate: Surrogate::new(2),
+            },
+        );
+
+        assert_eq!(resp.status, Status::Error);
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "a refusal applies nothing, so it counts nothing"
+        );
+    }
+
+    /// A single put running on the destination home applies its replica of a
+    /// dual-homed edge without counting it. This is the single-edge form of
+    /// the double count the wire test caught, so it fails if the ownership
+    /// gate is dropped from `execute_edge_put`.
+    #[test]
+    fn a_put_on_a_foreign_home_counts_no_write() {
+        use std::sync::Arc;
+
+        use crate::control::metrics::SystemMetrics;
+
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+
+        // The task is homed on "b"; the put writes an edge whose source is
+        // "a", so this core is the destination home.
+        let resp = h.core.execute_edge_put(
+            &make_task_at_source(7, "b"),
+            EdgePutParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "b",
+                properties: b"w=1",
+                src_surrogate: Surrogate::new(1),
+                dst_surrogate: Surrogate::new(2),
+            },
+        );
+
+        assert_eq!(resp.status, Status::Ok, "the replica is still applied");
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "the destination home does not own the logical edge"
+        );
+    }
+
+    /// Replay re-enters this handler for edges a client wrote before the
+    /// restart, so it must not count them as new writes.
+    #[test]
+    fn a_replayed_put_counts_no_write() {
+        use std::sync::Arc;
+
+        use crate::control::metrics::SystemMetrics;
+
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        h.core.boot_replaying_wal = true;
+
+        let resp = h.core.execute_edge_put(
+            &make_task_at_source(8, "a"),
+            EdgePutParams {
+                tid: 1,
+                collection: "knows",
+                src_id: "a",
+                label: "KNOWS",
+                dst_id: "b",
+                properties: b"w=1",
+                src_surrogate: Surrogate::new(1),
+                dst_surrogate: Surrogate::new(2),
+            },
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "a replayed write happened before the restart"
+        );
+    }
 
     #[test]
     fn edge_put_emits_cdc_insert_on_its_collection() {
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs b/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
index dcaf26b5f..e4914b54c 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/put_batch.rs
@@ -74,6 +74,16 @@ impl CoreLoop {
                     }
                     partition.set_node_surrogate(&edge.src_id, edge.src_surrogate);
                     partition.set_node_surrogate(&edge.dst_id, edge.dst_surrogate);
+                    // Counted per edge as it is applied, not once for the
+                    // batch: edges before a mid-batch failure are still
+                    // written and must show up in the counter. A dual-homed
+                    // edge is counted by its source home only.
+                    if !self.boot_replaying_wal()
+                        && owns_logical_edge_stats(task, &edge.src_id)
+                        && let Some(metrics) = &self.metrics
+                    {
+                        metrics.record_graph_edge_written();
+                    }
                 }
                 Err(e) => {
                     return self.response_error(
@@ -119,12 +129,16 @@ impl CoreLoop {
 
 #[cfg(test)]
 mod tests {
+    use std::sync::Arc;
+    use std::sync::atomic::Ordering;
+
     use crate::bridge::envelope::{ErrorCode, Status};
+    use crate::control::metrics::SystemMetrics;
     use crate::types::TenantId;
     use nodedb_physical::physical_plan::BatchEdge;
     use nodedb_types::{DatabaseId, QualifiedCollection, Surrogate};
 
-    use super::super::shared::test_support::{make_core, make_task_with_lsn};
+    use super::super::shared::test_support::{make_core, make_task_at_source, make_task_with_lsn};
 
     fn edge(src: &str, dst: &str) -> BatchEdge {
         BatchEdge {
@@ -137,6 +151,92 @@ mod tests {
         }
     }
 
+    /// Every applied edge of a batch is one write, not one write for the
+    /// batch: the counter has to describe the edges that reached storage.
+    #[test]
+    fn a_batch_counts_one_write_per_applied_edge() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        // One source home owns every edge of the batch, so each applied edge
+        // is one counted write.
+        let edges = vec![edge("owner", "b"), edge("owner", "d"), edge("owner", "f")];
+
+        let resp = h
+            .core
+            .execute_edge_put_batch(&make_task_at_source(11, "owner"), 1, &edges);
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(metrics.graph_edges_written.load(Ordering::Relaxed), 3);
+    }
+
+    /// A dual-homed edge runs on both endpoint homes. Only the source home
+    /// owns the logical edge, so a batch running on a foreign home applies the
+    /// edge without counting it — otherwise one cross-shard insert would count
+    /// twice.
+    #[test]
+    fn a_batch_on_a_foreign_home_counts_no_write() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+
+        // The task is homed on "b", the batch writes an edge whose source is
+        // "a" — the destination-home replica of a dual-homed edge.
+        let resp =
+            h.core
+                .execute_edge_put_batch(&make_task_at_source(13, "b"), 1, &[edge("a", "b")]);
+
+        assert_eq!(resp.status, Status::Ok, "the replica is still applied");
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "the destination home does not own the logical edge"
+        );
+    }
+
+    /// Replay re-enters this handler for batches a client ran before the
+    /// restart, so it must not count them as new writes.
+    #[test]
+    fn a_replayed_batch_counts_no_write() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        h.core.boot_replaying_wal = true;
+
+        let resp = h.core.execute_edge_put_batch(
+            &make_task_at_source(14, "owner"),
+            1,
+            &[edge("owner", "b")],
+        );
+
+        assert_eq!(resp.status, Status::Ok);
+        assert_eq!(
+            metrics.graph_edges_written.load(Ordering::Relaxed),
+            0,
+            "a replayed write happened before the restart"
+        );
+    }
+
+    /// A dangling endpoint refuses the whole batch before any edge is applied,
+    /// so no write is counted — the counter and the affected count agree.
+    #[test]
+    fn a_refused_batch_counts_no_write() {
+        let mut h = make_core();
+        let metrics = Arc::new(SystemMetrics::new());
+        h.core.set_metrics(Arc::clone(&metrics));
+        h.core
+            .mark_node_deleted(DatabaseId::DEFAULT.as_u64(), 1, "gone");
+
+        let resp = h.core.execute_edge_put_batch(
+            &make_task_with_lsn(12),
+            1,
+            &[edge("a", "b"), edge("c", "gone")],
+        );
+
+        assert_eq!(resp.status, Status::Error);
+        assert_eq!(metrics.graph_edges_written.load(Ordering::Relaxed), 0);
+    }
+
     /// The funnel cancels the batch's record on a dangling refusal, so the
     /// refusal must leave no edge of the batch behind.
     #[test]
diff --git a/nodedb/src/data/executor/handlers/graph_edge_write/shared.rs b/nodedb/src/data/executor/handlers/graph_edge_write/shared.rs
index 5ce0ec219..d8432d4f7 100644
--- a/nodedb/src/data/executor/handlers/graph_edge_write/shared.rs
+++ b/nodedb/src/data/executor/handlers/graph_edge_write/shared.rs
@@ -16,6 +16,37 @@ pub(in crate::data::executor) fn owns_logical_edge_stats(
     task.request.vshard_id == VShardId::from_key(src_id.as_bytes())
 }
 
+/// Whether this participant is the one that counts a delete of
+/// `(src_id, dst_id)` once.
+///
+/// WIP — this cannot answer correctly from the Data Plane today, and exists so
+/// the limitation is stated in one place instead of implied by two call sites.
+///
+/// A delete removes the copy of the row its participant holds, so the count
+/// must follow the removal:
+///
+/// * One core, or two endpoints on one vShard (the Control Plane calls this
+///   `single_home`): one participant tombstones the forward and reverse rows
+///   together, so it must count even when it is not the source home.
+/// * Two or more cores: each home keeps its own copy in its own store, both
+///   see a live pre-image, and only the source home may count or one logical
+///   delete counts twice.
+///
+/// The deciding fact is `single_home`, which
+/// [`nodedb::control::server::shared::ddl::neutral::graph_ops::edge`] computes
+/// on the Control Plane and which the plan does not carry. Distinguishing the
+/// two cases here would need a new `GraphOp::EdgeDelete` field threaded from
+/// there, so this predicate currently reports the removal-follows rule and
+/// over-counts on a cluster. Do not rely on the counter on a multi-core
+/// deployment until that field exists.
+pub(in crate::data::executor) fn counts_logical_edge_delete(
+    _task: &ExecutionTask,
+    _src_id: &str,
+    _dst_id: &str,
+) -> bool {
+    true
+}
+
 /// Bundled arguments for [`CoreLoop::execute_edge_put`].
 pub(in crate::data::executor) struct EdgePutParams<'a> {
     pub tid: u64,
@@ -113,11 +144,32 @@ pub(super) mod test_support {
     /// it — the LSN the emitted CDC event then carries. The `plan` field is
     /// unused by the edge handlers (they take params directly).
     pub fn make_task_with_lsn(lsn: u64) -> crate::data::executor::task::ExecutionTask {
+        task_at_vshard(lsn, VShardId::new(0))
+    }
+
+    /// A task whose `vshard_id` is the home of `key`, the vShard the planner
+    /// routes a write on `key` to.
+    ///
+    /// The edge write handlers record persistent logical-edge statistics only
+    /// on the source home, so a test asserting those counters must build its
+    /// task the way the planner routes one.
+    pub fn make_task_at_source(
+        lsn: u64,
+        src_id: &str,
+    ) -> crate::data::executor::task::ExecutionTask {
+        task_at_vshard(lsn, VShardId::from_key(src_id.as_bytes()))
+    }
+
+    /// A task carrying `wal_lsn` on an explicit vShard home.
+    pub fn task_at_vshard(
+        lsn: u64,
+        vshard_id: VShardId,
+    ) -> crate::data::executor::task::ExecutionTask {
         crate::data::executor::task::ExecutionTask::new(Request {
             request_id: RequestId::new(1),
             tenant_id: TenantId::new(1),
             database_id: DatabaseId::DEFAULT,
-            vshard_id: VShardId::new(0),
+            vshard_id,
             plan: PhysicalPlan::Graph(GraphOp::Neighbors {
                 node_id: "x".to_string(),
                 edge_label: None,
diff --git a/nodedb/src/data/executor/replay_policy.rs b/nodedb/src/data/executor/replay_policy.rs
index 92151a4f4..d69b04e17 100644
--- a/nodedb/src/data/executor/replay_policy.rs
+++ b/nodedb/src/data/executor/replay_policy.rs
@@ -57,6 +57,20 @@ use crate::data::executor::handlers::transaction::undo::UndoEntry;
 use crate::data::executor::replay_abort::abort_replay;
 
 impl CoreLoop {
+    /// Whether this core is rebuilding engine state from the WAL at boot.
+    ///
+    /// Boot replay re-enters the ordinary write handlers, so a handler that
+    /// records client activity asks this first: every edge boot replay
+    /// re-applies was written by a client before the restart, and counting it
+    /// again reports the restart as new work. The flag covers the whole boot
+    /// pass and is never set on a serving core.
+    ///
+    /// An online committed-redo apply reaches the same handlers, but it does
+    /// not come through the boot entry, so it counts as the real write it is.
+    pub(in crate::data::executor) fn boot_replaying_wal(&self) -> bool {
+        self.boot_replaying_wal
+    }
+
     /// Whether a committed-redo apply is driving the replay arms.
     pub(in crate::data::executor) fn applying_committed_redo(&self) -> bool {
         self.redo_apply.scope.is_some()
diff --git a/nodedb/src/data/executor/wal_replay_all.rs b/nodedb/src/data/executor/wal_replay_all.rs
index 7a99db993..880405469 100644
--- a/nodedb/src/data/executor/wal_replay_all.rs
+++ b/nodedb/src/data/executor/wal_replay_all.rs
@@ -41,6 +41,13 @@ impl CoreLoop {
         }
         let core_id = self.core_id;
 
+        // Boot replay re-enters the ordinary write handlers to rebuild engine
+        // state, and the metrics are attached before recovery runs. Every edge
+        // re-applied here was written by a client before the restart, so the
+        // write counters must not see it: the flag is held across the whole
+        // pass and cleared on every exit below.
+        self.boot_replaying_wal = true;
+
         // Replay decides every record handed to it before this core serves a
         // request or writes a checkpoint: it applies the record, or a stamp,
         // a tombstone or an abort marker says it must not. Every one of them
@@ -101,6 +108,10 @@ impl CoreLoop {
                 std::process::exit(1);
             }
         }
+
+        // Replay is over: this core is about to serve, so client writes must
+        // count again.
+        self.boot_replaying_wal = false;
     }
 }
 
diff --git a/nodedb/tests/crash_recovery_overlays.rs b/nodedb/tests/crash_recovery_overlays.rs
index 201b9eb4c..6c0cf6b55 100644
--- a/nodedb/tests/crash_recovery_overlays.rs
+++ b/nodedb/tests/crash_recovery_overlays.rs
@@ -51,6 +51,83 @@ async fn graph_edges_survive_kill_9() {
     );
 }
 
+/// The graph write counters describe client activity, so a restart that
+/// re-applies an edge from the WAL must not report that edge as a new write.
+///
+/// Boot replay re-enters the same put/delete handlers, and the metrics are
+/// attached before recovery runs, so without a boot-replay guard the counters
+/// count every replayed edge again on every restart.
+#[tokio::test(flavor = "multi_thread")]
+async fn replay_does_not_count_graph_edges_as_new_writes() {
+    let mut h = CrashHarness::new();
+    h.spawn();
+    h.wait_ready();
+
+    h.exec("CREATE COLLECTION crash_counter_edges").await;
+    h.exec("GRAPH INSERT EDGE IN 'crash_counter_edges' FROM 'a' TO 'b' TYPE 'knows'")
+        .await;
+    h.exec("GRAPH INSERT EDGE IN 'crash_counter_edges' FROM 'a' TO 'c' TYPE 'knows'")
+        .await;
+
+    let written = counter_value(&h, "graph_edges_written_total").await;
+    assert!(
+        written >= 2,
+        "two client inserts must be counted before the crash, saw {written}"
+    );
+
+    h.kill_9();
+    h.reopen();
+
+    // The counters are per-process, so a restart begins at zero. What matters
+    // is that recovery's re-application of both edges, which re-enters the
+    // same put handler, does not count as client work: without the boot-replay
+    // guard this reads 2 (or 4, one per home) with no client write at all.
+    let after_replay = counter_value(&h, "graph_edges_written_total").await;
+    assert_eq!(
+        after_replay, 0,
+        "boot replay must not count a re-applied edge as a new write"
+    );
+
+    let recovered = h
+        .query_col(
+            "MATCH (x)-[:knows]->(y) IN 'crash_counter_edges' RETURN x, y",
+            "y",
+        )
+        .await;
+    assert_eq!(
+        recovered.len(),
+        2,
+        "both edges must still be present after recovery: {recovered:?}"
+    );
+
+    // A write issued after recovery is still counted normally.
+    h.exec("GRAPH INSERT EDGE IN 'crash_counter_edges' FROM 'a' TO 'd' TYPE 'knows'")
+        .await;
+    assert_eq!(
+        counter_value(&h, "graph_edges_written_total").await,
+        1,
+        "a live write after recovery is counted exactly once"
+    );
+}
+
+/// Read one `(name, value)` counter out of `SHOW STATS`.
+async fn counter_value(h: &CrashHarness, name: &str) -> u64 {
+    let names = h.query_col_idx("SHOW STATS", 0).await;
+    let values = h.query_col_idx("SHOW STATS", 1).await;
+    assert_eq!(
+        names.len(),
+        values.len(),
+        "SHOW STATS must return the same number of names and values"
+    );
+    let index = names
+        .iter()
+        .position(|n| n == name)
+        .unwrap_or_else(|| panic!("SHOW STATS must carry {name}, got {names:?}"));
+    values[index]
+        .parse::<u64>()
+        .unwrap_or_else(|_| panic!("{name} must be a decimal integer, got {:?}", values[index]))
+}
+
 #[tokio::test(flavor = "multi_thread")]
 async fn graph_node_labels_survive_kill_9() {
     let mut h = CrashHarness::new();
diff --git a/nodedb/tests/wire/cases/graph_edge_write_counters.rs b/nodedb/tests/wire/cases/graph_edge_write_counters.rs
new file mode 100644
index 000000000..21be7bb90
--- /dev/null
+++ b/nodedb/tests/wire/cases/graph_edge_write_counters.rs
@@ -0,0 +1,209 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! The graph edge write counters, end to end.
+//!
+//! `nodedb_graph_edges_written_total` and `nodedb_graph_edges_deleted_total`
+//! count edge versions applied and live edges tombstoned. They answer what the
+//! live-edge gauge cannot: a put that rewrites an existing edge leaves
+//! `nodedb_graph_edges` flat, and only the write counter moves.
+//!
+//! Both surfaces are asserted against one statement's outcome, so the SQL row
+//! and the Prometheus sample cannot drift apart.
+
+use crate::harness::TestServer;
+
+/// Read `GET /metrics` and return its body.
+async fn fetch_metrics(http_port: u16) -> String {
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
+    let mut stream = tokio::net::TcpStream::connect(("127.0.0.1", http_port))
+        .await
+        .expect("connect to /metrics");
+    let req = b"GET /metrics HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n";
+    stream.write_all(req).await.expect("write metrics request");
+    let mut body = String::new();
+    stream
+        .read_to_string(&mut body)
+        .await
+        .expect("read metrics response");
+    body
+}
+
+/// Read one `(name, value)` counter out of `SHOW STATS`, as a number.
+async fn stats_counter(server: &TestServer, name: &str) -> u64 {
+    let rows = server
+        .query_named_rows("SHOW STATS")
+        .await
+        .expect("SHOW STATS must succeed");
+    let row = rows
+        .iter()
+        .find(|r| r.get("name").map(|n| n == name).unwrap_or(false))
+        .unwrap_or_else(|| panic!("SHOW STATS must carry {name}; got {rows:?}"));
+    let value = row.get("value").expect("the counter row carries a value");
+    value
+        .parse::<u64>()
+        .unwrap_or_else(|_| panic!("{name} must be a decimal integer, got {value:?}"))
+}
+
+/// Read one sample value out of a `/metrics` body.
+fn metrics_sample(body: &str, name: &str) -> u64 {
+    let prefix = format!("{name} ");
+    body.lines()
+        .find_map(|line| line.strip_prefix(&prefix))
+        .unwrap_or_else(|| panic!("/metrics must export a {name} sample"))
+        .trim()
+        .parse::<u64>()
+        .unwrap_or_else(|_| panic!("{name} must be a decimal integer"))
+}
+
+/// Each `GRAPH INSERT EDGE` applies one edge version, so the write counter
+/// advances by one and both surfaces agree — including over a second insert of
+/// the same edge, which adds no live edge and still counts.
+#[tokio::test]
+async fn graph_insert_edge_advances_the_write_counter() {
+    let server = TestServer::start().await;
+    server.exec("CREATE COLLECTION edge_counter").await.unwrap();
+    assert_eq!(
+        stats_counter(&server, "graph_edges_written_total").await,
+        0,
+        "this server starts with clean counters"
+    );
+
+    server
+        .exec("GRAPH INSERT EDGE IN 'edge_counter' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("first edge insert");
+    server
+        .exec("GRAPH INSERT EDGE IN 'edge_counter' FROM 'a' TO 'c' TYPE 'knows'")
+        .await
+        .expect("second edge insert");
+
+    // Exact, not a bound: this test owns its server subprocess and its data
+    // dir, so the counter it reads describes only the statements above. Two
+    // cross-shard inserts must count two, not four — each is dual-homed, and
+    // the destination home must not count the edge it does not own.
+    assert_eq!(
+        stats_counter(&server, "graph_edges_written_total").await,
+        2,
+        "two inserts apply two edge versions"
+    );
+
+    let body = fetch_metrics(server.http_port).await;
+    assert!(
+        body.contains("# TYPE nodedb_graph_edges_written_total counter"),
+        "/metrics must declare nodedb_graph_edges_written_total as a counter"
+    );
+    assert_eq!(
+        metrics_sample(&body, "nodedb_graph_edges_written_total"),
+        2,
+        "the SQL row and the Prometheus sample describe the same writes"
+    );
+}
+
+/// A delete of a live edge counts one; a delete of an edge that was never
+/// there counts nothing, because the tombstone removes no live edge.
+#[tokio::test]
+async fn graph_delete_edge_counts_only_a_live_edge() {
+    let server = TestServer::start().await;
+    server
+        .exec("CREATE COLLECTION edge_counter_del")
+        .await
+        .unwrap();
+    server
+        .exec("GRAPH INSERT EDGE IN 'edge_counter_del' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("seed the edge");
+    assert_eq!(
+        stats_counter(&server, "graph_edges_deleted_total").await,
+        0,
+        "seeding an edge removes nothing"
+    );
+
+    server
+        .exec("GRAPH DELETE EDGE IN 'edge_counter_del' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("delete the live edge");
+    // Exact, not a bound: this test owns its server subprocess and its data
+    // dir, so only the statements above are counted.
+    assert_eq!(
+        stats_counter(&server, "graph_edges_deleted_total").await,
+        1,
+        "removing a live edge counts one"
+    );
+
+    server
+        .exec("GRAPH DELETE EDGE IN 'edge_counter_del' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("delete the same edge again");
+    assert_eq!(
+        stats_counter(&server, "graph_edges_deleted_total").await,
+        1,
+        "the second delete removes nothing, so it does not count"
+    );
+
+    let body = fetch_metrics(server.http_port).await;
+    assert!(
+        body.contains("# TYPE nodedb_graph_edges_deleted_total counter"),
+        "/metrics must declare nodedb_graph_edges_deleted_total as a counter"
+    );
+    assert_eq!(
+        metrics_sample(&body, "nodedb_graph_edges_deleted_total"),
+        1,
+        "the SQL row and the Prometheus sample describe the same removals"
+    );
+}
+
+/// The multi-core shape, which is where the two counters differ today.
+///
+/// A cross-shard edge is dual-homed: both endpoint homes run the statement,
+/// and each home keeps its own copy of the row in its own core's store. The
+/// write counter is ownership-gated, so it reports one logical insert as one.
+/// The delete counter counts each home that removes a live row, so on two
+/// cores it reports one logical delete as two — documented in the metric's
+/// HELP text and in `counts_logical_edge_delete`.
+#[tokio::test]
+async fn the_counters_on_a_multi_core_server() {
+    let server = TestServer::start_multicores(2).await;
+    server
+        .exec("CREATE COLLECTION edge_counter_cores")
+        .await
+        .unwrap();
+
+    server
+        .exec("GRAPH INSERT EDGE IN 'edge_counter_cores' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("insert the edge");
+    assert_eq!(
+        stats_counter(&server, "graph_edges_written_total").await,
+        1,
+        "one logical insert counts one write, not one per home"
+    );
+
+    server
+        .exec("GRAPH DELETE EDGE IN 'edge_counter_cores' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("delete the edge");
+    let deleted = stats_counter(&server, "graph_edges_deleted_total").await;
+    assert!(
+        deleted >= 1,
+        "the delete must be counted at least once, saw {deleted}"
+    );
+    // The exact value is the WIP limitation: 2 on two cores. Asserted as a
+    // bound so this test states the current behaviour without pinning the
+    // value the fix will change.
+    assert_eq!(
+        deleted, 2,
+        "WIP: each endpoint home counts its own removal on a multi-core server"
+    );
+
+    // A rewrite is still a write, and still counts once on two cores.
+    server
+        .exec("GRAPH INSERT EDGE IN 'edge_counter_cores' FROM 'a' TO 'b' TYPE 'knows'")
+        .await
+        .expect("rewrite the edge");
+    assert_eq!(
+        stats_counter(&server, "graph_edges_written_total").await,
+        2,
+        "the rewrite counts a second write and the gauge stays flat"
+    );
+}
diff --git a/nodedb/tests/wire/cases/mod.rs b/nodedb/tests/wire/cases/mod.rs
index f58a3ffcf..e7766ce9a 100644
--- a/nodedb/tests/wire/cases/mod.rs
+++ b/nodedb/tests/wire/cases/mod.rs
@@ -86,6 +86,7 @@ mod graph_drop_hides_edges;
 mod graph_dsl_algo;
 mod graph_dsl_argument_validation;
 mod graph_dsl_handlers;
+mod graph_edge_write_counters;
 mod graph_match_authorization;
 mod graph_rag_fusion;
 mod graph_timeseries_rls_probe;