diff --git a/nodedb-cluster-tests/tests/cluster_execute_request.rs b/nodedb-cluster-tests/tests/cluster_execute_request.rs
index 30706d11b..4690c5743 100644
--- a/nodedb-cluster-tests/tests/cluster_execute_request.rs
+++ b/nodedb-cluster-tests/tests/cluster_execute_request.rs
@@ -53,6 +53,7 @@ fn make_kv_put_request(
             collection: collection.into(),
             version: descriptor_version,
         }],
+        txn_id: None,
     }
 }
 
@@ -209,6 +210,7 @@ async fn execute_request_cross_node_dispatch() {
             collection: "cross_node_kv".into(),
             version: 0, // Accept any version (pre-B.1 sentinel bypass)
         }],
+        txn_id: None,
     };
 
     let resp = send_execute_request(sender_transport, target_addr, req).await;
diff --git a/nodedb-cluster-tests/tests/cross_node_ryow_probe.rs b/nodedb-cluster-tests/tests/cross_node_ryow_probe.rs
new file mode 100644
index 000000000..8b6fecf33
--- /dev/null
+++ b/nodedb-cluster-tests/tests/cross_node_ryow_probe.rs
@@ -0,0 +1,145 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! DEFERRED PROBE (design unit U4/U5) — CROSS-NODE in-transaction
+//! read-your-own-writes. `#[ignore]`d: cross-node in-txn RYOW is out of scope
+//! for U3, whose validation surface is single-node only (LOCKED design §8:
+//! "single-node-verifiable: U1, U2, U3 (RYOW), U7"; cross-node is "cluster-bound
+//! (stage-2): U4, U5, U6, U8"). Kept in-tree, ignored, as the executable
+//! specification for the U4/U5 cross-shard-transaction work.
+//!
+//! SINGLE-NODE in-txn RYOW works today (U3): an in-block read takes the
+//! NON-streaming dispatch path (`maybe_stream_select` bails on `InBlock`), and
+//! `dispatch_local` threads the session `txn_id` to the local leaseholder so the
+//! Data Plane merges this transaction's staging overlay. The native twin
+//! (`native_sql_ryow_probe.rs`) and the graph twin
+//! (`single_node_calvin_graph_txn.rs`) both cover that and stay GREEN.
+//!
+//! CONFIRMED ROOT CAUSE of the cross-node gap (diagnosed 2026-07, live trace):
+//! the whole per-transaction machinery — staging gate, COMMIT buffer, and the
+//! COMMIT/ROLLBACK overlay flush — is LOCAL-ONLY. On a NON-owner node the pgwire
+//! dispatcher takes the remote-leader forward short-circuit
+//! (`routing/execute.rs`: `should_forward_via_gateway` → `dispatch_tasks_via_gateway`)
+//! and RETURNS before `dispatch_task_loop`, so `route_in_tx_write` never runs.
+//! The in-txn write is therefore shipped to the owner as a PLAIN replicable
+//! write (Raft-proposed = committed at statement time, NOT staged into
+//! `txn_overlays[txn_id]`), and `dispatch_tasks_via_gateway` hardcodes
+//! `QueryContext.txn_id = None` (`routing/gateway_dispatch.rs`) so the in-txn
+//! read reaches the owner with no txn id and cannot merge the overlay. Net: a
+//! cross-node in-txn write bypasses the transaction entirely — it neither
+//! reads-its-own-write NOR honours ROLLBACK atomicity.
+//!
+//! The `ExecuteRequest.txn_id` wire field (+ `WIRE_VERSION` 2→3) and the
+//! globally-unique `TxnId::from_origin` packing landed on this branch are the
+//! necessary GROUNDWORK for the fix, but are NOT sufficient on their own: making
+//! this probe pass requires routing the txn staging / read / COMMIT / ROLLBACK /
+//! overlay-drop through the gateway to the owning node — i.e. the U4/U5
+//! cross-shard-transaction unit.
+//!
+//! This probe drives BEGIN / INSERT / SELECT(own write) / ROLLBACK on EVERY node
+//! of a 3-node cluster against a single-vShard-homed collection. Exactly one node
+//! leads that vShard (local RYOW — the GREEN control); the other two route the
+//! whole transaction to it over QUIC and so exercise the cross-node path, which
+//! MISSES today. Remove `#[ignore]` when U4/U5 lands to turn it back into a gate.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::{TestCluster, wait_for};
+
+/// `n` for `id` read on `client` in its current session. `None` when absent, the
+/// column did not parse as an integer, or the query itself errored (a cross-node
+/// failure is recorded by the caller as a RYOW miss, not a panic, so every node
+/// is exercised).
+async fn n_for(client: &tokio_postgres::Client, id: &str) -> Option<i64> {
+    let msgs = client
+        .simple_query(&format!("SELECT n FROM t WHERE id = '{id}'"))
+        .await
+        .ok()?;
+    msgs.iter().find_map(|m| match m {
+        tokio_postgres::SimpleQueryMessage::Row(r) => {
+            r.get("n").and_then(|s| s.parse::<i64>().ok())
+        }
+        _ => None,
+    })
+}
+
+/// A transaction reads its OWN staged write even when the collection's vShard is
+/// led by a DIFFERENT node (cross-node read-your-own-writes).
+///
+/// `#[ignore]`d: DEFERRED to U4/U5 (cross-shard transactions). U3's validation
+/// surface is single-node only (LOCKED design §8). See the module docstring for
+/// the confirmed root cause. Run explicitly with `--ignored` to reproduce the
+/// cross-node gap while iterating on U4/U5.
+#[ignore = "cross-node in-txn RYOW is U4/U5 (cross-shard txn); U3 is single-node only — see module docstring"]
+#[tokio::test(flavor = "multi_thread", worker_threads = 6)]
+async fn cross_node_in_txn_select_reads_own_write() {
+    let cluster = TestCluster::spawn_three().await.expect("3-node cluster");
+
+    cluster
+        .exec_ddl_on_any_leader(
+            "CREATE COLLECTION t \
+             (id TEXT PRIMARY KEY, n INT) \
+             WITH (engine='document_strict')",
+        )
+        .await
+        .expect("CREATE COLLECTION t");
+
+    wait_for(
+        "all 3 nodes see collection t",
+        Duration::from_secs(10),
+        Duration::from_millis(50),
+        || {
+            cluster
+                .nodes
+                .iter()
+                .all(|n| n.cached_collection_count() >= 1)
+        },
+    )
+    .await;
+
+    // Drive the whole transaction on each node in turn. The collection is
+    // single-vShard-homed, so every key lives on the ONE owning vShard: the node
+    // leading it does local RYOW (the control that passes today), the other two
+    // route the transaction to it over QUIC and exercise the cross-node path.
+    let mut failures: Vec<String> = Vec::new();
+    for idx in 0..cluster.nodes.len() {
+        let client = &cluster.nodes[idx].client;
+        let id = format!("k{idx}");
+
+        client.simple_query("BEGIN").await.expect("BEGIN");
+
+        // Single-shard write; on a non-owning node it stages to the remote
+        // leaseholder's overlay. An error here is itself a cross-node gap
+        // symptom — record it and move on rather than aborting the whole probe.
+        if let Err(e) = client
+            .simple_query(&format!("INSERT INTO t (id, n) VALUES ('{id}', 42)"))
+            .await
+        {
+            let _ = client.simple_query("ROLLBACK").await;
+            failures.push(format!("node {idx}: in-txn INSERT errored: {e}"));
+            continue;
+        }
+
+        // THE PROBE: the transaction's OWN read must observe its staged write.
+        let seen = n_for(client, &id).await;
+
+        // ROLLBACK before asserting so a miss never leaves the txn open and every
+        // node's staging overlay is torn down regardless of outcome.
+        client.simple_query("ROLLBACK").await.expect("ROLLBACK");
+
+        if seen != Some(42) {
+            failures.push(format!(
+                "node {idx}: RYOW MISS — in-txn SELECT saw {seen:?}, expected Some(42)"
+            ));
+        }
+    }
+
+    assert!(
+        failures.is_empty(),
+        "CROSS-NODE in-txn read-your-own-writes is broken (ExecuteRequest carries \
+         no txn_id, so the owning node stages/reads with txn_id=None): {failures:#?}"
+    );
+
+    cluster.shutdown().await;
+}
diff --git a/nodedb-cluster-tests/tests/native_sql_ryow_probe.rs b/nodedb-cluster-tests/tests/native_sql_ryow_probe.rs
new file mode 100644
index 000000000..9c86b46ef
--- /dev/null
+++ b/nodedb-cluster-tests/tests/native_sql_ryow_probe.rs
@@ -0,0 +1,161 @@
+// SPDX-License-Identifier: BUSL-1.1
+
+//! REGRESSION GUARDS for two native (MessagePack) explicit-transaction
+//! behaviours that were broken on `upstream/main` and fixed on this branch.
+//! Upstream had NO native explicit-txn wire coverage
+//! (`single_node_calvin_graph_txn.rs` drives BEGIN over pgwire), so these paths
+//! were unguarded; both tests assert the CORRECT post-fix behaviour.
+//!
+//! GAP A — first-frame BEGIN must buffer. Previously `handle_begin` ->
+//! `run_begin` -> `sessions.begin(addr)` used
+//! `write_session(addr,..).unwrap_or(Ok(()))`; with NO session yet for the addr
+//! it silently returned `Ok(())` WITHOUT setting `InBlock`, so a BEGIN issued
+//! before any SQL no-oped and the next INSERT autocommitted. `handle_begin` now
+//! calls `ensure_session` before `run_begin`, so the session transitions to
+//! `InBlock` even when BEGIN is the first post-auth frame.
+//!
+//! GAP B (U1) — in-txn SQL read-your-own-writes. Native SQL reads route through
+//! `sql_gateway.rs::dispatch_task_via_gateway`, which previously built a
+//! `GatewayQueryContext { tenant_id, trace_id, database_id }` with NO `txn_id`
+//! and called `gw.execute(&gw_ctx, plan)`, dropping the connection's `txn_id`
+//! before the Data Plane so the per-txn staging overlay-merge never fired. It
+//! now propagates `txn_id` into `GatewayQueryContext` so the Data Plane resolves
+//! this transaction's staging overlay. The pgwire twin
+//! (`nodedb/tests/sql_transactions_staged_point_writes.rs`) asserts `["42"]`.
+//!
+//! The first test guards GAP A; the second WARMS UP the session (a SQL read
+//! triggers `ensure_session`) so a subsequent BEGIN buffers, then guards the
+//! GAP B RYOW path. A failure here signals a REGRESSION of either fix.
+
+mod common;
+
+use std::time::Duration;
+
+use common::cluster_harness::TestClusterNode;
+use nodedb_client::native::connection::NativeConnection;
+use nodedb_types::protocol::AuthMethod;
+use tokio_postgres::SimpleQueryMessage;
+
+/// Single-node harness with collection `t` created and settled.
+async fn setup() -> TestClusterNode {
+    let node = TestClusterNode::spawn(1, vec![])
+        .await
+        .expect("spawn single-node cluster");
+    tokio::time::sleep(Duration::from_millis(300)).await;
+
+    node.exec(
+        "CREATE COLLECTION t \
+         (id STRING NOT NULL PRIMARY KEY, n INT) WITH (engine='document_strict')",
+    )
+    .await
+    .expect("CREATE COLLECTION t");
+    tokio::time::sleep(Duration::from_millis(150)).await;
+
+    node
+}
+
+/// One pinned native connection (explicit txns are peer-address-scoped, so the
+/// whole transaction must ride a single socket).
+async fn native_conn(node: &TestClusterNode) -> NativeConnection {
+    let mut conn = NativeConnection::connect(&format!("127.0.0.1:{}", node.native_port))
+        .await
+        .expect("connect native socket");
+    conn.authenticate(
+        AuthMethod::Trust {
+            username: "admin".into(),
+        },
+        None,
+    )
+    .await
+    .expect("native trust auth");
+    conn
+}
+
+/// `n` for `id` read over the INDEPENDENT pgwire session (durable state, not the
+/// txn connection's overlay). `None` when absent.
+async fn durable_n(node: &TestClusterNode, id: &str) -> Option<String> {
+    let msgs = node
+        .client
+        .simple_query(&format!("SELECT n FROM t WHERE id = '{id}'"))
+        .await
+        .expect("pgwire durable read");
+    msgs.iter().find_map(|m| match m {
+        SimpleQueryMessage::Row(r) => r.get("n").map(str::to_string),
+        _ => None,
+    })
+}
+
+/// GAP A: a BEGIN issued as the first post-auth frame does NOT buffer; the
+/// following INSERT autocommits (visible to an independent session pre-COMMIT).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_first_frame_begin_does_not_buffer_probe() {
+    let node = setup().await;
+    let mut conn = native_conn(&node).await;
+
+    // BEGIN is the very first post-auth frame -> no session exists yet.
+    conn.begin().await.expect("BEGIN");
+    conn.execute_sql("INSERT INTO t (id, n) VALUES ('ff', 7)")
+        .await
+        .expect("INSERT after first-frame BEGIN");
+
+    let leaked = durable_n(&node, "ff").await;
+    assert_eq!(
+        leaked, None,
+        "GAP A GUARD: first-frame BEGIN must buffer the write (invisible \
+         pre-COMMIT). leaked={leaked:?}. Some(\"7\") -> the INSERT \
+         autocommitted (first-frame BEGIN no-oped) -> GAP A REGRESSED."
+    );
+
+    node.shutdown().await;
+}
+
+/// GAP B (U1): with the session warmed up (so BEGIN buffers), the transaction's
+/// OWN native SQL read must observe the staged write (read-your-own-writes).
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn native_in_txn_sql_select_reads_own_write_probe() {
+    let node = setup().await;
+    let mut conn = native_conn(&node).await;
+
+    // WARM UP: a SQL statement triggers `ensure_session` (sql.rs), creating the
+    // session so the following BEGIN can transition it to InBlock.
+    conn.execute_sql("SELECT n FROM t WHERE id = 'warmup'")
+        .await
+        .expect("warm-up read (creates native session)");
+
+    conn.begin().await.expect("BEGIN");
+    conn.execute_sql("INSERT INTO t (id, n) VALUES ('a', 42)")
+        .await
+        .expect("in-tx INSERT");
+
+    // Precondition for a meaningful RYOW probe: the write must actually be
+    // buffered (invisible to an independent session). If this trips, staging is
+    // broken even with a warm session (a broader gap than U1).
+    assert_eq!(
+        durable_n(&node, "a").await,
+        None,
+        "warmed in-block write must be invisible to other sessions mid-transaction"
+    );
+
+    // THE U1 PROBE: the transaction's OWN native SQL read.
+    let res = conn
+        .execute_sql("SELECT n FROM t WHERE id = 'a'")
+        .await
+        .expect("in-tx read");
+    let seen = res
+        .rows
+        .first()
+        .and_then(|r| r.first())
+        .and_then(|v| v.as_i64());
+
+    assert_eq!(
+        seen,
+        Some(42),
+        "GAP B (U1) GUARD: native in-txn SQL SELECT must read-its-own-write \
+         (Some(42)); seen={seen:?}, rows={:?}. None -> native RYOW \
+         REGRESSED (gateway dropped txn_id).",
+        res.rows
+    );
+
+    conn.commit().await.expect("COMMIT");
+    node.shutdown().await;
+}
diff --git a/nodedb-cluster/src/rpc_codec/execute.rs b/nodedb-cluster/src/rpc_codec/execute.rs
index 5bd1fa68c..862897a79 100644
--- a/nodedb-cluster/src/rpc_codec/execute.rs
+++ b/nodedb-cluster/src/rpc_codec/execute.rs
@@ -38,6 +38,19 @@ pub struct ExecuteRequest {
     pub trace_id: [u8; 16],
     /// Caller's view of descriptor versions for every collection touched by the plan.
     pub descriptor_versions: Vec<DescriptorVersionEntry>,
+    /// Owning session transaction, when this plan executes inside an explicit
+    /// `BEGIN` block. `Some(txn_id)` directs the receiving leaseholder to
+    /// resolve this transaction's staging overlay so a cross-node in-transaction
+    /// read observes its own staged writes (read-your-own-writes); `None` is an
+    /// autocommit statement with no session transaction.
+    ///
+    /// Wire compatibility: this field was added in `WIRE_VERSION` 3. Because
+    /// `ExecuteRequest` is an rkyv fixed-layout message, the added field changes
+    /// its archived layout. The bumped envelope version means a v2 node rejects
+    /// a v3-stamped frame outright (`unwrap_bytes_versioned`) rather than
+    /// silently misdecoding it, so nodes must be upgraded together (see
+    /// `crate::wire`).
+    pub txn_id: Option<u64>,
 }
 
 /// Response to an `ExecuteRequest`.
@@ -238,11 +251,17 @@ mod tests {
                     version: 1,
                 },
             ],
+            txn_id: Some(0xABCD_1234),
         };
         let decoded = roundtrip_req(req.clone());
         assert_eq!(decoded.plan_bytes, req.plan_bytes);
         assert_eq!(decoded.tenant_id, 7);
         assert_eq!(decoded.deadline_remaining_ms, 5000);
+        assert_eq!(
+            decoded.txn_id,
+            Some(0xABCD_1234),
+            "txn_id roundtrips for an in-transaction plan"
+        );
         assert_eq!(
             decoded.trace_id, req.trace_id,
             "trace_id roundtrips correctly"
@@ -261,9 +280,11 @@ mod tests {
             deadline_remaining_ms: 1000,
             trace_id: [0u8; 16],
             descriptor_versions: vec![],
+            txn_id: None,
         };
         let decoded = roundtrip_req(req);
         assert!(decoded.descriptor_versions.is_empty());
+        assert_eq!(decoded.txn_id, None, "autocommit plan carries no txn_id");
     }
 
     #[test]
@@ -382,6 +403,7 @@ mod tests {
                 collection: "wide".into(),
                 version: 3,
             }],
+            txn_id: Some(99),
         };
         let rpc = RaftRpc::ExecuteStreamRequest(req.clone());
         let encoded = super::super::encode(&rpc).unwrap();
@@ -395,6 +417,7 @@ mod tests {
                 assert_eq!(r.descriptor_versions.len(), 1);
                 assert_eq!(r.descriptor_versions[0].collection, "wide");
                 assert_eq!(r.descriptor_versions[0].version, 3);
+                assert_eq!(r.txn_id, Some(99), "stream request carries txn_id");
             }
             other => panic!("expected ExecuteStreamRequest, got {other:?}"),
         }
diff --git a/nodedb-cluster/src/transport/client/tests.rs b/nodedb-cluster/src/transport/client/tests.rs
index e0d782fb1..b445d380b 100644
--- a/nodedb-cluster/src/transport/client/tests.rs
+++ b/nodedb-cluster/src/transport/client/tests.rs
@@ -233,6 +233,7 @@ async fn execute_stream_roundtrip() {
         deadline_remaining_ms: 5000,
         trace_id: [0u8; 16],
         descriptor_versions: vec![],
+        txn_id: None,
     });
 
     let stream = client.send_rpc_stream(1, req).await.unwrap();
diff --git a/nodedb-cluster/src/wire.rs b/nodedb-cluster/src/wire.rs
index 2aa4a973e..b0d6a0497 100644
--- a/nodedb-cluster/src/wire.rs
+++ b/nodedb-cluster/src/wire.rs
@@ -149,7 +149,14 @@ pub enum VShardMessageType {
 ///
 /// v2 widens `vshard_id` u16→u32 in the binary frame, increasing min header
 /// size from 26 to 28 bytes.
-pub const WIRE_VERSION: u16 = 2;
+///
+/// v3 adds the `txn_id` field to `ExecuteRequest` (cross-node in-transaction
+/// read-your-own-writes). `ExecuteRequest` is an rkyv fixed-layout message, so
+/// the added field changes its archived layout: a v2 node cannot decode a v3
+/// frame and vice versa. The matching `WireVersion::CURRENT` bump makes a v2
+/// node reject a v3-stamped envelope outright rather than silently misdecode —
+/// so a mixed v2/v3 cluster fails loudly and nodes must be upgraded together.
+pub const WIRE_VERSION: u16 = 3;
 
 impl VShardEnvelope {
     pub fn new(
@@ -265,8 +272,11 @@ mod tests {
     use super::*;
 
     #[test]
-    fn wire_version_is_2() {
-        assert_eq!(WIRE_VERSION, 2, "v2 widened vshard_id to u32");
+    fn wire_version_is_3() {
+        assert_eq!(
+            WIRE_VERSION, 3,
+            "v3 added ExecuteRequest.txn_id for cross-node in-transaction RYOW"
+        );
     }
 
     #[test]
diff --git a/nodedb-cluster/src/wire_version/types.rs b/nodedb-cluster/src/wire_version/types.rs
index da84ecba8..8578ff572 100644
--- a/nodedb-cluster/src/wire_version/types.rs
+++ b/nodedb-cluster/src/wire_version/types.rs
@@ -14,10 +14,16 @@ impl WireVersion {
     /// v1: legacy — no `Versioned<T>` envelope. Raw inner type bytes.
     pub const V1: WireVersion = WireVersion(1);
 
-    /// v2: first explicit envelope version. Introduced alongside this module.
-    /// `encode_versioned` always emits v2; `decode_versioned` falls back to v1
-    /// if the outer envelope is absent or unparseable.
-    pub const CURRENT: WireVersion = WireVersion(2);
+    /// The current envelope version this build emits. `encode_versioned`
+    /// always stamps `CURRENT`, and `decode_versioned` rejects any envelope
+    /// whose version exceeds it — so a newer peer's frames fail loudly on an
+    /// older node rather than being silently misdecoded.
+    ///
+    /// - v2: first explicit envelope version, introduced alongside this module.
+    /// - v3: bumped alongside `crate::wire::WIRE_VERSION` for the
+    ///   `ExecuteRequest.txn_id` rkyv layout change (cross-node in-transaction
+    ///   read-your-own-writes).
+    pub const CURRENT: WireVersion = WireVersion(3);
 }
 
 impl std::fmt::Display for WireVersion {
diff --git a/nodedb-types/src/id/txn.rs b/nodedb-types/src/id/txn.rs
index 524aec826..61b482189 100644
--- a/nodedb-types/src/id/txn.rs
+++ b/nodedb-types/src/id/txn.rs
@@ -5,7 +5,10 @@ use std::fmt;
 use serde::{Deserialize, Serialize};
 
 /// Identifies a session transaction block. Keys the per-transaction staging
-/// overlay. Unique across concurrent sessions on a shard.
+/// overlay. Globally unique across the cluster: the minting node's id is
+/// packed into the high bits (see [`TxnId::from_origin`]), so two coordinators
+/// that each mint the same per-node sequence never collide on the same
+/// staging-overlay key of a shared owner shard.
 #[derive(
     Debug,
     Clone,
@@ -21,10 +24,30 @@ use serde::{Deserialize, Serialize};
 pub struct TxnId(u64);
 
 impl TxnId {
+    /// Low bits reserved for the per-node monotonic sequence; the remaining
+    /// high bits carry the origin node id. 48 bits of sequence is ~281e12
+    /// transactions per process lifetime; 16 bits of node id is 65_536 nodes.
+    const SEQ_BITS: u32 = 48;
+    const SEQ_MASK: u64 = (1u64 << Self::SEQ_BITS) - 1;
+
     pub const fn new(id: u64) -> Self {
         Self(id)
     }
 
+    /// Mint a globally-unique id from the origin `node_id` and a per-node
+    /// monotonic `seq`uence. The node id occupies the high 16 bits, the
+    /// sequence the low 48.
+    ///
+    /// `node_id == 0` (single-node / no-cluster deployments, where
+    /// `SharedState.node_id` is never assigned) yields `TxnId(seq)` —
+    /// byte-identical to the pre-cluster scheme, so single-node behavior and
+    /// value-sensitive tests are unchanged. A non-zero origin disambiguates
+    /// otherwise-identical sequences from different coordinators, which is
+    /// what makes cross-node read-your-own-write staging collision-free.
+    pub const fn from_origin(node_id: u64, seq: u64) -> Self {
+        Self((node_id << Self::SEQ_BITS) | (seq & Self::SEQ_MASK))
+    }
+
     pub const fn as_u64(self) -> u64 {
         self.0
     }
@@ -35,3 +58,38 @@ impl fmt::Display for TxnId {
         write!(f, "txn:{}", self.0)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::TxnId;
+
+    #[test]
+    fn node_zero_is_identity() {
+        // Single-node / no-cluster: the id equals the bare sequence, keeping
+        // the pre-cluster scheme (and value-sensitive tests) byte-identical.
+        assert_eq!(TxnId::from_origin(0, 1).as_u64(), 1);
+        assert_eq!(TxnId::from_origin(0, 42).as_u64(), 42);
+    }
+
+    #[test]
+    fn distinct_origins_never_collide_on_equal_sequence() {
+        // The exact flaky cross-node RYOW collision: two coordinators each
+        // minting sequence 1. Packing the origin in keeps them distinct, so
+        // they never share a staging-overlay key on a common owner shard.
+        assert_ne!(TxnId::from_origin(1, 1), TxnId::from_origin(2, 1));
+        assert_ne!(TxnId::from_origin(1, 1), TxnId::from_origin(3, 1));
+    }
+
+    #[test]
+    fn same_origin_distinct_sequence_stays_unique() {
+        assert_ne!(TxnId::from_origin(3, 1), TxnId::from_origin(3, 2));
+    }
+
+    #[test]
+    fn origin_is_recoverable_from_high_bits() {
+        // Sanity on the layout: sequence in the low 48, origin in the high 16.
+        let id = TxnId::from_origin(7, 99).as_u64();
+        assert_eq!(id >> 48, 7);
+        assert_eq!(id & ((1u64 << 48) - 1), 99);
+    }
+}
diff --git a/nodedb/src/control/backup/orchestrator.rs b/nodedb/src/control/backup/orchestrator.rs
index 169373b43..db4cb6800 100644
--- a/nodedb/src/control/backup/orchestrator.rs
+++ b/nodedb/src/control/backup/orchestrator.rs
@@ -322,6 +322,8 @@ async fn snapshot_remote(
         deadline_remaining_ms: NODE_SNAPSHOT_TIMEOUT.as_millis() as u64,
         trace_id: TraceId::generate().0,
         descriptor_versions: Vec::new(),
+        // Snapshot scans committed state for backup, never an interactive txn.
+        txn_id: None,
     });
 
     let resp = transport
diff --git a/nodedb/src/control/backup/restore/remote.rs b/nodedb/src/control/backup/restore/remote.rs
index 1394dcfba..3ae35aebf 100644
--- a/nodedb/src/control/backup/restore/remote.rs
+++ b/nodedb/src/control/backup/restore/remote.rs
@@ -38,6 +38,8 @@ pub(super) async fn dispatch_remote(
         deadline_remaining_ms: NODE_RESTORE_TIMEOUT.as_millis() as u64,
         trace_id: TraceId::generate().0,
         descriptor_versions: Vec::new(),
+        // Restore streams committed backup data, never an interactive txn.
+        txn_id: None,
     });
     let resp = transport
         .send_rpc(node_id, req)
diff --git a/nodedb/src/control/exec_receiver/executor.rs b/nodedb/src/control/exec_receiver/executor.rs
index d5ff5cd3c..cfd665e10 100644
--- a/nodedb/src/control/exec_receiver/executor.rs
+++ b/nodedb/src/control/exec_receiver/executor.rs
@@ -226,6 +226,10 @@ impl LocalPlanExecutor {
 
         let tenant_id = crate::types::TenantId::new(req.tenant_id);
         let trace_id = nodedb_types::TraceId(req.trace_id);
+        // Owning interactive transaction (WIRE_VERSION 3+). Threaded into the
+        // all-cores fan so an in-transaction StageWrite / read resolves this
+        // transaction's staging overlay on the owning shard (cross-node RYOW).
+        let txn_id = req.txn_id.map(crate::types::TxnId::new);
 
         // ── Replicable write: drive through Raft, NOT local cores ─────────────
         //
@@ -281,7 +285,14 @@ impl LocalPlanExecutor {
 
         match tokio::time::timeout(
             deadline,
-            execute_plan_all_local_cores(&self.state, tenant_id, database_id, plan, trace_id),
+            execute_plan_all_local_cores(
+                &self.state,
+                tenant_id,
+                database_id,
+                plan,
+                trace_id,
+                txn_id,
+            ),
         )
         .await
         {
@@ -316,16 +327,17 @@ impl LocalPlanExecutor {
 
         let tenant_id = crate::types::TenantId::new(req.tenant_id);
         let trace_id = nodedb_types::TraceId(req.trace_id);
+        // Owning interactive transaction (WIRE_VERSION 3+): resolve this
+        // transaction's staging overlay for a streamed in-transaction read.
+        let txn_id = req.txn_id.map(crate::types::TxnId::new);
 
-        // Cluster RPC receiver (remote-node local execution): `ExecuteRequest`
-        // carries no session-transaction context yet, so `None`.
         let mut stream = match crate::control::server::exchange::gather::gather_all_cores_stream(
             &self.state,
             tenant_id,
             database_id,
             plan,
             trace_id,
-            None,
+            txn_id,
         ) {
             Ok(s) => s,
             Err(e) => {
diff --git a/nodedb/src/control/gateway/core.rs b/nodedb/src/control/gateway/core.rs
index dbf66f11d..b5ab5464c 100644
--- a/nodedb/src/control/gateway/core.rs
+++ b/nodedb/src/control/gateway/core.rs
@@ -27,10 +27,10 @@ use tracing::{Instrument, debug, info_span};
 use crate::Error;
 use crate::control::state::SharedState;
 use crate::control::trace_export::EmitSpanParams;
-use crate::types::{DatabaseId, TenantId, TraceId};
+use crate::types::{DatabaseId, TenantId, TraceId, TxnId};
 use nodedb_physical::physical_plan::PhysicalPlan;
 
-use super::dispatcher::{default_deadline_ms, dispatch_route};
+use super::dispatcher::{DispatchRouteParams, default_deadline_ms, dispatch_route};
 use super::fuser::fuse_payloads;
 use super::key_extractor::UnwiredKeyExtractor;
 use super::plan_cache::{PlanCache, PlanCacheKey, SqlKey, hash_placeholder_types, hash_sql};
@@ -48,6 +48,16 @@ pub struct QueryContext {
     /// catalog lookups. Single-database deployments pass
     /// [`DatabaseId::DEFAULT`].
     pub database_id: DatabaseId,
+    /// Owning interactive transaction, when this plan is dispatched inside a
+    /// `BEGIN`/`COMMIT` session. Threaded to the Data Plane hop so the
+    /// leaseholder resolves this transaction's staging overlay
+    /// (read-your-own-writes) instead of reading only committed state. `None`
+    /// for autocommit statements and all non-session callers.
+    ///
+    /// Forwarded on both the **local-leader** dispatch (`dispatch_local`) and
+    /// the remote `ExecuteRequest` (`WIRE_VERSION` 3 carries `txn_id`), so
+    /// cross-node in-transaction reads resolve the overlay on the owning shard.
+    pub txn_id: Option<TxnId>,
 }
 
 /// The gateway: routes, dispatches, retries, and caches physical plans.
@@ -243,6 +253,7 @@ impl Gateway {
                 let tenant_id = ctx.tenant_id;
                 let database_id = ctx.database_id;
                 let trace_id = ctx.trace_id;
+                let txn_id = ctx.txn_id;
                 let version_set = version_set_for_route.clone();
                 async move {
                     let decision = {
@@ -277,15 +288,16 @@ impl Gateway {
                         decision,
                         vshard_id: vshard_id_u32,
                     };
-                    dispatch_route(
+                    dispatch_route(DispatchRouteParams {
                         route,
-                        &shared,
+                        shared: &shared,
                         tenant_id,
                         database_id,
                         trace_id,
+                        txn_id,
                         deadline_ms,
-                        &version_set,
-                    )
+                        version_set: &version_set,
+                    })
                     .await
                 }
             })
diff --git a/nodedb/src/control/gateway/dispatcher.rs b/nodedb/src/control/gateway/dispatcher.rs
index 77df4f565..fe81cf56c 100644
--- a/nodedb/src/control/gateway/dispatcher.rs
+++ b/nodedb/src/control/gateway/dispatcher.rs
@@ -5,7 +5,8 @@
 //! The dispatcher takes a single [`TaskRoute`] and executes it:
 //!
 //! - `RouteDecision::Local` → dispatch through the SPSC bridge via
-//!   [`dispatch_to_data_plane`].
+//!   [`dispatch_to_data_plane_with_txn`] (threading the owning transaction so a
+//!   local-leader in-transaction read resolves its staging overlay).
 //! - `RouteDecision::Remote { node_id, .. }` → encode the plan as
 //!   [`ExecuteRequest`] bytes and send via [`NexarTransport::send_rpc`].
 //! - `RouteDecision::Broadcast { .. }` → each individual route in the
@@ -23,33 +24,51 @@ use nodedb_cluster::rpc_codec::{ExecuteRequest, RaftRpc, TypedClusterError};
 use tracing::debug;
 
 use crate::Error;
-use crate::control::server::dispatch_utils::dispatch_to_data_plane;
+use crate::control::server::dispatch_utils::dispatch_to_data_plane_with_txn;
 use crate::control::server::result_stream::{ResultStream, RowBatch};
 use crate::control::state::SharedState;
-use crate::types::{DatabaseId, Lsn, TenantId, TraceId, VShardId};
+use crate::types::{DatabaseId, Lsn, TenantId, TraceId, TxnId, VShardId};
 use nodedb_physical::physical_plan::wire as plan_wire;
 
 use super::route::{RouteDecision, TaskRoute};
 use super::version_set::GatewayVersionSet;
 
-/// Dispatch a single route and return the raw payload bytes.
+/// Parameters for [`dispatch_route`] (bundled to stay within clippy's
+/// `too_many_arguments` limit once the `txn_id` thread was added).
 ///
 /// `tenant_id` — the authenticated tenant for this query.
 /// `trace_id` — distributed trace ID propagated from the client request.
+/// `txn_id` — owning interactive transaction, threaded to both the **local**
+///   hop and the **remote** `ExecuteRequest` so the leaseholder (wherever it
+///   lives) resolves this transaction's staging overlay (read-your-own-writes).
 /// `deadline_ms` — remaining deadline in milliseconds.
 /// `version_set` — descriptor versions for the collections touched by the plan.
-pub async fn dispatch_route(
-    route: TaskRoute,
-    shared: &Arc<SharedState>,
-    tenant_id: TenantId,
-    database_id: DatabaseId,
-    trace_id: TraceId,
-    deadline_ms: u64,
-    version_set: &GatewayVersionSet,
-) -> Result<Vec<Vec<u8>>, Error> {
+pub struct DispatchRouteParams<'a> {
+    pub route: TaskRoute,
+    pub shared: &'a Arc<SharedState>,
+    pub tenant_id: TenantId,
+    pub database_id: DatabaseId,
+    pub trace_id: TraceId,
+    pub txn_id: Option<TxnId>,
+    pub deadline_ms: u64,
+    pub version_set: &'a GatewayVersionSet,
+}
+
+/// Dispatch a single route and return the raw payload bytes.
+pub async fn dispatch_route(params: DispatchRouteParams<'_>) -> Result<Vec<Vec<u8>>, Error> {
+    let DispatchRouteParams {
+        route,
+        shared,
+        tenant_id,
+        database_id,
+        trace_id,
+        txn_id,
+        deadline_ms,
+        version_set,
+    } = params;
     match route.decision {
         RouteDecision::Local => {
-            dispatch_local(route, shared, tenant_id, database_id, trace_id).await
+            dispatch_local(route, shared, tenant_id, database_id, trace_id, txn_id).await
         }
         RouteDecision::Remote { node_id, vshard_id } => {
             dispatch_remote(RemoteDispatchArgs {
@@ -60,6 +79,7 @@ pub async fn dispatch_route(
                 tenant_id,
                 database_id,
                 trace_id,
+                txn_id,
                 deadline_ms,
                 version_set,
             })
@@ -88,12 +108,21 @@ pub async fn dispatch_route(
 }
 
 /// Parameters for [`dispatch_route_stream`].
+///
+/// `txn_id` — owning interactive transaction, threaded to the local
+///   `gather_all_cores_stream` and the remote `ExecuteStreamRequest` so a
+///   streamed in-transaction read resolves this transaction's staging overlay.
+///   In practice in-transaction reads take the non-streaming path today
+///   (`maybe_stream_select` bails inside a block), so this is `None` for
+///   autocommit streaming scans; it is threaded for correctness if that gate
+///   ever changes.
 pub struct DispatchRouteStreamParams<'a> {
     pub route: TaskRoute,
     pub shared: &'a Arc<SharedState>,
     pub tenant_id: TenantId,
     pub database_id: DatabaseId,
     pub trace_id: TraceId,
+    pub txn_id: Option<TxnId>,
     pub deadline_ms: u64,
     pub version_set: &'a GatewayVersionSet,
 }
@@ -116,20 +145,20 @@ pub async fn dispatch_route_stream(
         tenant_id,
         database_id,
         trace_id,
+        txn_id,
         deadline_ms,
         version_set,
     } = args;
     match route.decision {
-        // Cluster gateway route dispatch: no session-transaction context
-        // crosses this boundary yet, so `None`. TRACKED: cross-node
-        // in-transaction reads are a known gap (see resolve/exchange.rs).
+        // Local streamed scan: thread the owning transaction so the all-cores
+        // gather resolves this transaction's staging overlay when present.
         RouteDecision::Local => crate::control::server::exchange::gather::gather_all_cores_stream(
             shared,
             tenant_id,
             database_id,
             route.plan,
             trace_id,
-            None,
+            txn_id,
         ),
         RouteDecision::Remote { node_id, vshard_id } => {
             dispatch_remote_stream(RemoteDispatchArgs {
@@ -140,6 +169,7 @@ pub async fn dispatch_route_stream(
                 tenant_id,
                 database_id,
                 trace_id,
+                txn_id,
                 deadline_ms,
                 version_set,
             })
@@ -164,21 +194,23 @@ async fn dispatch_local(
     tenant_id: TenantId,
     database_id: DatabaseId,
     trace_id: TraceId,
+    txn_id: Option<TxnId>,
 ) -> Result<Vec<Vec<u8>>, Error> {
     let vshard_id = VShardId::new(route.vshard_id);
-    let resp = dispatch_to_data_plane(
+    let resp = dispatch_to_data_plane_with_txn(
         shared,
         tenant_id,
         database_id,
         vshard_id,
         route.plan,
         trace_id,
+        txn_id,
     )
     .await?;
     Ok(vec![resp.payload.to_vec()])
 }
 
-/// Arguments for a remote dispatch call (bundles the 8 parameters to stay
+/// Arguments for a remote dispatch call (bundles the parameters to stay
 /// within clippy's `too_many_arguments` limit).
 struct RemoteDispatchArgs<'a> {
     plan: nodedb_physical::physical_plan::PhysicalPlan,
@@ -188,6 +220,10 @@ struct RemoteDispatchArgs<'a> {
     tenant_id: TenantId,
     database_id: DatabaseId,
     trace_id: TraceId,
+    /// Owning session transaction, stamped into the outgoing `ExecuteRequest`
+    /// so the remote leaseholder resolves this transaction's staging overlay
+    /// (cross-node read-your-own-writes). `None` for autocommit statements.
+    txn_id: Option<TxnId>,
     deadline_ms: u64,
     version_set: &'a GatewayVersionSet,
 }
@@ -202,6 +238,7 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result<Vec<Vec<u8>>, E
         tenant_id,
         database_id,
         trace_id,
+        txn_id,
         deadline_ms,
         version_set,
     } = args;
@@ -270,6 +307,7 @@ async fn dispatch_remote(args: RemoteDispatchArgs<'_>) -> Result<Vec<Vec<u8>>, E
         deadline_remaining_ms: deadline_ms,
         trace_id: trace_id.0,
         descriptor_versions,
+        txn_id: txn_id.map(|t| t.as_u64()),
     });
 
     debug!(
@@ -335,6 +373,7 @@ async fn dispatch_remote_stream(args: RemoteDispatchArgs<'_>) -> Result<ResultSt
         tenant_id,
         database_id,
         trace_id,
+        txn_id,
         deadline_ms,
         version_set,
     } = args;
@@ -390,6 +429,7 @@ async fn dispatch_remote_stream(args: RemoteDispatchArgs<'_>) -> Result<ResultSt
         deadline_remaining_ms: deadline_ms,
         trace_id: trace_id.0,
         descriptor_versions,
+        txn_id: txn_id.map(|t| t.as_u64()),
     });
 
     debug!(
diff --git a/nodedb/src/control/gateway/stream.rs b/nodedb/src/control/gateway/stream.rs
index 5421fcf51..70c1412ab 100644
--- a/nodedb/src/control/gateway/stream.rs
+++ b/nodedb/src/control/gateway/stream.rs
@@ -70,6 +70,7 @@ impl Gateway {
                 let tenant_id = ctx.tenant_id;
                 let database_id = ctx.database_id;
                 let trace_id = ctx.trace_id;
+                let txn_id = ctx.txn_id;
                 let version_set = version_set_for_route.clone();
                 async move {
                     let decision = {
@@ -110,6 +111,7 @@ impl Gateway {
                         tenant_id,
                         database_id,
                         trace_id,
+                        txn_id,
                         deadline_ms,
                         version_set: &version_set,
                     })
diff --git a/nodedb/src/control/planner/calvin/preexec.rs b/nodedb/src/control/planner/calvin/preexec.rs
index 159ffbaf4..f48c215ff 100644
--- a/nodedb/src/control/planner/calvin/preexec.rs
+++ b/nodedb/src/control/planner/calvin/preexec.rs
@@ -126,6 +126,8 @@ pub async fn run_preexec_scan(
             tenant_id,
             trace_id: TraceId::ZERO,
             database_id,
+            // Calvin preexecution uses its own deterministic transaction model.
+            txn_id: None,
         };
         let payloads =
             gateway
diff --git a/nodedb/src/control/scatter_gather.rs b/nodedb/src/control/scatter_gather.rs
index a0f3dd928..f1c263ffa 100644
--- a/nodedb/src/control/scatter_gather.rs
+++ b/nodedb/src/control/scatter_gather.rs
@@ -345,6 +345,9 @@ pub async fn coordinate_cross_shard_hop(
                     tenant_id: crate::types::TenantId::new(tenant_id_u64),
                     trace_id: TraceId::generate(),
                     database_id,
+                    // Cross-node graph hop; cross-node txn propagation (RPC
+                    // envelope) is tracked debt, so no txn_id here.
+                    txn_id: None,
                 };
 
                 // Build a fresh QueryContext per traversal using cloned inputs
diff --git a/nodedb/src/control/server/exchange/all_cores/dispatch.rs b/nodedb/src/control/server/exchange/all_cores/dispatch.rs
index 9a15190a6..426c83391 100644
--- a/nodedb/src/control/server/exchange/all_cores/dispatch.rs
+++ b/nodedb/src/control/server/exchange/all_cores/dispatch.rs
@@ -5,7 +5,7 @@
 //! bespoke single-blob merge (see `snapshot.rs`, `bsp.rs`, `wcc.rs`).
 
 use crate::control::state::SharedState;
-use crate::types::{DatabaseId, Lsn, TenantId, TraceId};
+use crate::types::{DatabaseId, Lsn, TenantId, TraceId, TxnId};
 use nodedb_physical::physical_plan::{GraphOp, MetaOp, PhysicalPlan};
 
 use super::bsp::fan_bsp_all_cores;
@@ -38,12 +38,22 @@ pub struct NodeLevelResult {
 ///
 /// At 1 core/node every branch is behaviour-identical to the prior single-core
 /// paths.
+///
+/// `txn_id` is the owning interactive transaction when this plan was dispatched
+/// inside a `BEGIN`/`COMMIT` session (threaded from the remote `ExecuteRequest`
+/// since `WIRE_VERSION` 3). It selects this transaction's staging overlay so a
+/// cross-node in-transaction read (or `StageWrite`) resolves the same overlay
+/// the owning shard's local session would — cross-node read-your-own-writes.
+/// `None` for autocommit statements. The MATCH branch is a deliberate
+/// exception: its overlay lives on the origin node, so cross-node MATCH RYOW is
+/// a separate unit and it always runs committed-only.
 pub async fn execute_plan_all_local_cores(
     state: &SharedState,
     tenant_id: TenantId,
     database_id: DatabaseId,
     plan: PhysicalPlan,
     trace_id: TraceId,
+    txn_id: Option<TxnId>,
 ) -> crate::Result<NodeLevelResult> {
     match &plan {
         PhysicalPlan::Graph(g) => match g {
@@ -111,7 +121,7 @@ pub async fn execute_plan_all_local_cores(
             | GraphOp::TemporalNeighbors { .. }
             | GraphOp::TemporalAlgorithm { .. }
             | GraphOp::Stats { .. } => {
-                generic_gather(state, tenant_id, database_id, plan, trace_id).await
+                generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await
             }
         },
 
@@ -192,7 +202,7 @@ pub async fn execute_plan_all_local_cores(
             | MetaOp::RecordCalvinWriteVersions { .. }
             | MetaOp::CalvinFlush { .. }
             | MetaOp::CalvinDrop { .. } => {
-                generic_gather(state, tenant_id, database_id, plan, trace_id).await
+                generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await
             }
         },
 
@@ -208,25 +218,27 @@ pub async fn execute_plan_all_local_cores(
         | PhysicalPlan::Query(_)
         | PhysicalPlan::Array(_)
         | PhysicalPlan::ClusterArray(_) => {
-            generic_gather(state, tenant_id, database_id, plan, trace_id).await
+            generic_gather(state, tenant_id, database_id, plan, trace_id, txn_id).await
         }
     }
 }
 
 /// Generic gather path: delegate to [`gather_all_cores`] and wrap.
+///
+/// `txn_id` is forwarded to the row gather so a cross-node in-transaction read
+/// (or `StageWrite`) resolves the owning transaction's staging overlay
+/// (read-your-own-writes). `None` for autocommit statements.
 async fn generic_gather(
     state: &SharedState,
     tenant_id: TenantId,
     database_id: DatabaseId,
     plan: PhysicalPlan,
     trace_id: TraceId,
+    txn_id: Option<TxnId>,
 ) -> crate::Result<NodeLevelResult> {
     use crate::control::server::exchange::gather::gather_all_cores;
 
-    // Cluster RPC receiver path (remote-node local execution): no session
-    // transaction context crosses the node boundary yet, so `txn_id` is
-    // `None` here. TRACKED: cross-node in-transaction reads are a known gap.
-    let outcome = gather_all_cores(state, tenant_id, database_id, plan, trace_id, None).await?;
+    let outcome = gather_all_cores(state, tenant_id, database_id, plan, trace_id, txn_id).await?;
     Ok(NodeLevelResult {
         payload: outcome.merged_array,
         watermark_lsn: outcome.watermark_lsn,
diff --git a/nodedb/src/control/server/exchange/gather.rs b/nodedb/src/control/server/exchange/gather.rs
index 90592b962..461be890f 100644
--- a/nodedb/src/control/server/exchange/gather.rs
+++ b/nodedb/src/control/server/exchange/gather.rs
@@ -370,6 +370,9 @@ pub async fn gather_all_vshards(
         tenant_id,
         trace_id,
         database_id,
+        // Carry the in-block transaction id into the single-vShard gateway
+        // dispatch so the owning vShard resolves this txn's staging overlay.
+        txn_id,
     };
 
     // `Box::pin` breaks an async-fn recursion cycle: the gateway dispatches the
diff --git a/nodedb/src/control/server/exchange/resolve/exchange.rs b/nodedb/src/control/server/exchange/resolve/exchange.rs
index be8a18ec2..a1761e47f 100644
--- a/nodedb/src/control/server/exchange/resolve/exchange.rs
+++ b/nodedb/src/control/server/exchange/resolve/exchange.rs
@@ -169,6 +169,9 @@ async fn resolve_exchange(
                         tenant_id,
                         trace_id,
                         database_id,
+                        // `None` here by the `txn_id.is_none()` guard above
+                        // (streaming is autocommit-only); threaded for honesty.
+                        txn_id,
                     };
                     // NOTE: cluster mode does not yet thread `txn_id` through
                     // `gateway.execute_stream` — cross-node in-transaction
diff --git a/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs b/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs
index 658b4000f..a550887c2 100644
--- a/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs
+++ b/nodedb/src/control/server/graph_dispatch/cluster_resolve.rs
@@ -12,7 +12,7 @@
 use std::sync::Arc;
 
 use crate::bridge::envelope::{Payload, PhysicalPlan};
-use crate::control::gateway::dispatcher::dispatch_route;
+use crate::control::gateway::dispatcher::{DispatchRouteParams, dispatch_route};
 use crate::control::gateway::router::resolve_decision;
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::gateway::{RouteDecision, TaskRoute};
@@ -93,6 +93,10 @@ pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_no
             database_id,
             plan,
             TraceId::ZERO,
+            // Cross-node MATCH resolve: the transaction's staged overlay lives
+            // on the origin node and is unreachable here, so run committed-only.
+            // Cross-node MATCH read-your-own-writes is a separate unit.
+            None,
         )
         .await?;
         Ok(Payload::from_vec(node_result.payload))
@@ -106,15 +110,18 @@ pub(in crate::control::server::graph_dispatch) async fn dispatch_superstep_to_no
             },
             vshard_id: route_vshard,
         };
-        let payloads = dispatch_route(
+        let payloads = dispatch_route(DispatchRouteParams {
             route,
-            shared_arc,
+            shared: shared_arc,
             tenant_id,
             database_id,
-            TraceId::ZERO,
+            trace_id: TraceId::ZERO,
+            // Cross-node graph superstep read; no owning interactive txn and a
+            // remote route cannot carry txn_id regardless.
+            txn_id: None,
             deadline_ms,
             version_set,
-        )
+        })
         .await?;
         payloads
             .into_iter()
diff --git a/nodedb/src/control/server/graph_dispatch/hop.rs b/nodedb/src/control/server/graph_dispatch/hop.rs
index 74bfa6563..a132beae6 100644
--- a/nodedb/src/control/server/graph_dispatch/hop.rs
+++ b/nodedb/src/control/server/graph_dispatch/hop.rs
@@ -36,7 +36,9 @@ use std::sync::Arc;
 use futures::future::join_all;
 
 use crate::bridge::envelope::PhysicalPlan;
-use crate::control::gateway::dispatcher::{default_deadline_ms, dispatch_route};
+use crate::control::gateway::dispatcher::{
+    DispatchRouteParams, default_deadline_ms, dispatch_route,
+};
 use crate::control::gateway::router::resolve_decision;
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::gateway::{RouteDecision, TaskRoute};
@@ -332,15 +334,18 @@ async fn expand_remote(
         // Box::pin keeps the heterogeneous async dispatch futures uniform for
         // `join_all` and guards against any future async-recursion concerns.
         Box::pin(async move {
-            dispatch_route(
+            dispatch_route(DispatchRouteParams {
                 route,
-                shared_arc,
+                shared: shared_arc,
                 tenant_id,
                 database_id,
-                TraceId::ZERO,
+                trace_id: TraceId::ZERO,
+                // Graph traversal hops are cross-node reads with no owning
+                // interactive transaction; remote routes cannot carry txn_id.
+                txn_id: None,
                 deadline_ms,
-                &version_set,
-            )
+                version_set: &version_set,
+            })
             .await
         })
     });
diff --git a/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs b/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs
index b65c6df75..c19d0df71 100644
--- a/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs
+++ b/nodedb/src/control/server/graph_dispatch/match_scatter/round_loop.rs
@@ -9,7 +9,7 @@ use std::sync::Arc;
 use futures::future::join_all;
 
 use crate::bridge::envelope::PhysicalPlan;
-use crate::control::gateway::dispatcher::dispatch_route;
+use crate::control::gateway::dispatcher::{DispatchRouteParams, dispatch_route};
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::gateway::{RouteDecision, TaskRoute};
 use crate::control::server::graph_dispatch::match_broadcast::broadcast_match_to_all_cores;
@@ -90,15 +90,18 @@ fn push_dispatch_fut<'f>(
             let shared_arc = ctx.shared_arc;
             let version_set = ctx.version_set.clone();
             futs.push(Box::pin(async move {
-                let payloads = dispatch_route(
+                let payloads = dispatch_route(DispatchRouteParams {
                     route,
-                    shared_arc,
+                    shared: shared_arc,
                     tenant_id,
                     database_id,
-                    TraceId::ZERO,
+                    trace_id: TraceId::ZERO,
+                    // Cross-node pattern-match scatter round; remote route, no
+                    // owning interactive transaction.
+                    txn_id: None,
                     deadline_ms,
-                    &version_set,
-                )
+                    version_set: &version_set,
+                })
                 .await?;
                 collect_remote_envelopes(node_id, payloads)
             }));
diff --git a/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs b/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs
index 2c18942e8..3db4c111d 100644
--- a/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs
+++ b/nodedb/src/control/server/graph_dispatch/match_scatter/round_zero.rs
@@ -8,7 +8,7 @@ use std::collections::HashSet;
 use futures::future::join_all;
 
 use crate::bridge::envelope::{Payload, PhysicalPlan};
-use crate::control::gateway::dispatcher::dispatch_route;
+use crate::control::gateway::dispatcher::{DispatchRouteParams, dispatch_route};
 use crate::control::gateway::version_set::GatewayVersionSet;
 use crate::control::gateway::{RouteDecision, TaskRoute};
 use crate::control::server::graph_dispatch::cluster_resolve::gateway_shared;
@@ -77,15 +77,18 @@ pub(super) async fn scatter_round_zero(
         let version_set = version_set.clone();
         let node_id = owner.node_id;
         Box::pin(async move {
-            let payloads = dispatch_route(
+            let payloads = dispatch_route(DispatchRouteParams {
                 route,
-                shared_arc,
+                shared: shared_arc,
                 tenant_id,
                 database_id,
-                TraceId::ZERO,
+                trace_id: TraceId::ZERO,
+                // Round-zero scatter is a cross-node remote read; no owning
+                // interactive transaction to propagate.
+                txn_id: None,
                 deadline_ms,
-                &version_set,
-            )
+                version_set: &version_set,
+            })
             .await?;
             collect_remote_envelopes(node_id, payloads)
         })
diff --git a/nodedb/src/control/server/http/routes/promql/remote.rs b/nodedb/src/control/server/http/routes/promql/remote.rs
index 0e4d500d9..a4fb3143c 100644
--- a/nodedb/src/control/server/http/routes/promql/remote.rs
+++ b/nodedb/src/control/server/http/routes/promql/remote.rs
@@ -90,6 +90,8 @@ pub async fn remote_write(
                     tenant_id,
                     trace_id: TraceId::generate(),
                     database_id: nodedb_types::id::DatabaseId::DEFAULT,
+                    // PromQL remote-read is a stateless autocommit query.
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, plan).await
             }
diff --git a/nodedb/src/control/server/http/routes/query.rs b/nodedb/src/control/server/http/routes/query.rs
index 441e2f17b..2cc834cb7 100644
--- a/nodedb/src/control/server/http/routes/query.rs
+++ b/nodedb/src/control/server/http/routes/query.rs
@@ -202,6 +202,8 @@ pub async fn query(
                         tenant_id: task.tenant_id,
                         trace_id,
                         database_id,
+                        // HTTP query endpoint is stateless autocommit.
+                        txn_id: None,
                     };
                     gw.execute(&gw_ctx, task.plan).await.map_err(|e| {
                         let (status, msg) = GatewayErrorMap::to_http(&e);
@@ -404,6 +406,8 @@ pub async fn query_ndjson(
                     tenant_id: task.tenant_id,
                     trace_id,
                     database_id,
+                    // HTTP query endpoint is stateless autocommit.
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, task.plan).await
             }
diff --git a/nodedb/src/control/server/http/routes/query_stream.rs b/nodedb/src/control/server/http/routes/query_stream.rs
index 775b82990..5ee075bea 100644
--- a/nodedb/src/control/server/http/routes/query_stream.rs
+++ b/nodedb/src/control/server/http/routes/query_stream.rs
@@ -61,6 +61,8 @@ pub(super) async fn try_open_stream(
             tenant_id: task.tenant_id,
             trace_id,
             database_id,
+            // HTTP streaming query endpoint is autocommit.
+            txn_id: None,
         };
         gw.execute_stream(&ctx, child_plan).await
     } else {
diff --git a/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs b/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs
index 725bd07bd..bcf472794 100644
--- a/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs
+++ b/nodedb/src/control/server/http/routes/ws_rpc/execute_sql.rs
@@ -35,6 +35,8 @@ pub async fn execute_sql(
                     tenant_id: task.tenant_id,
                     trace_id,
                     database_id: nodedb_types::id::DatabaseId::DEFAULT,
+                    // WebSocket RPC SQL exec is autocommit; no session txn block.
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, task.plan).await
             }
diff --git a/nodedb/src/control/server/ilp_batch.rs b/nodedb/src/control/server/ilp_batch.rs
index 047c24614..d4bfdf251 100644
--- a/nodedb/src/control/server/ilp_batch.rs
+++ b/nodedb/src/control/server/ilp_batch.rs
@@ -151,6 +151,8 @@ async fn flush_ilp_batch_inner(
                     tenant_id,
                     trace_id: TraceId::generate(),
                     database_id: nodedb_types::id::DatabaseId::DEFAULT,
+                    // ILP line-protocol ingest is autocommit; no interactive txn.
+                    txn_id: None,
                 };
                 gw.execute(&gw_ctx, plan)
                     .await
diff --git a/nodedb/src/control/server/native/dispatch/direct_ops.rs b/nodedb/src/control/server/native/dispatch/direct_ops.rs
index f8a3c03e2..c15dba92f 100644
--- a/nodedb/src/control/server/native/dispatch/direct_ops.rs
+++ b/nodedb/src/control/server/native/dispatch/direct_ops.rs
@@ -402,6 +402,10 @@ async fn dispatch_single_task_raw(
                 tenant_id,
                 trace_id: TraceId::generate(),
                 database_id: ctx.database_id(),
+                // Propagate the active transaction id so direct-op reads resolve
+                // this txn's staging overlay (read-your-own-writes), mirroring
+                // the SQL path in `sql_gateway.rs`.
+                txn_id,
             };
             match gw.execute(&gw_ctx, plan).await {
                 Ok(payloads) => Ok(gateway_payloads_to_response(payloads)),
diff --git a/nodedb/src/control/server/native/dispatch/sql_gateway.rs b/nodedb/src/control/server/native/dispatch/sql_gateway.rs
index 0f3993c37..fd7083d3c 100644
--- a/nodedb/src/control/server/native/dispatch/sql_gateway.rs
+++ b/nodedb/src/control/server/native/dispatch/sql_gateway.rs
@@ -39,6 +39,11 @@ pub(super) async fn dispatch_task_via_gateway(
                 tenant_id,
                 trace_id: TraceId::generate(),
                 database_id,
+                // Propagate the in-block transaction id so gateway dispatch
+                // resolves the per-txn staging overlay (read-your-own-writes).
+                // `route_in_tx_write` stamps this for in-block reads; the
+                // gateway branch previously dropped it, defeating native RYOW.
+                txn_id,
             };
             gw.execute(&gw_ctx, plan)
                 .await
diff --git a/nodedb/src/control/server/native/dispatch/streaming.rs b/nodedb/src/control/server/native/dispatch/streaming.rs
index 7e12d2e0d..bfb007df4 100644
--- a/nodedb/src/control/server/native/dispatch/streaming.rs
+++ b/nodedb/src/control/server/native/dispatch/streaming.rs
@@ -106,6 +106,10 @@ pub(crate) async fn try_open_sql_stream(
             tenant_id: task.tenant_id,
             trace_id: crate::types::TraceId::ZERO,
             database_id,
+            // In-block transactions are excluded above (streaming fast-path is
+            // autocommit-only), so this is `None` today; propagate the task's
+            // own id regardless to stay correct if that guard ever relaxes.
+            txn_id: task.txn_id,
         };
         gw.execute_stream(&gw_ctx, child_plan).await?
     } else {
diff --git a/nodedb/src/control/server/native/dispatch/transaction.rs b/nodedb/src/control/server/native/dispatch/transaction.rs
index 106bec153..a50801ea4 100644
--- a/nodedb/src/control/server/native/dispatch/transaction.rs
+++ b/nodedb/src/control/server/native/dispatch/transaction.rs
@@ -53,6 +53,11 @@ impl TxnDataPlane for NativeTxnDp<'_> {
                         tenant_id: task.tenant_id,
                         trace_id: TraceId::generate(),
                         database_id: task.database_id,
+                        // COMMIT applies the buffered writes as an atomic
+                        // TransactionBatch (self-describing plan); it is a
+                        // write-apply, not an overlay-resolving read, so no
+                        // `txn_id` is threaded.
+                        txn_id: None,
                     };
                     let payloads = gw.execute(&gw_ctx, task.plan).await?;
                     Ok(Response {
@@ -91,6 +96,14 @@ impl TxnDataPlane for NativeTxnDp<'_> {
 }
 
 pub(crate) fn handle_begin(ctx: &DispatchCtx<'_>, seq: u64) -> NativeResponse {
+    // Ensure a session exists before BEGIN. pgwire creates the session at
+    // connection startup (`pgwire/factory.rs`), but the native transport only
+    // lazily ensures one when an SQL statement arrives (`sql.rs`). A native
+    // client that sends BEGIN as its first frame would otherwise reach
+    // `run_begin` with no session, where `SessionStore::begin` silently
+    // no-ops (never setting `InBlock`) — so the first in-block write would
+    // autocommit instead of buffering into the staging overlay.
+    ctx.sessions.ensure_session(*ctx.peer_addr);
     match lifecycle::run_begin(ctx.sessions, ctx.peer_addr, ctx.state) {
         Ok(()) => NativeResponse::status_row(seq, "BEGIN"),
         Err(e) => {
diff --git a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs
index c3aef9864..b8f0f3e0c 100644
--- a/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs
+++ b/nodedb/src/control/server/pgwire/handler/routing/gateway_dispatch.rs
@@ -103,6 +103,21 @@ impl NodeDbPgHandler {
             tenant_id,
             trace_id: TraceId::generate(),
             database_id,
+            // This is the remote-leader forward path (`should_forward_via_gateway`
+            // requires a non-local leader). Local in-block reads take
+            // `dispatch_task_loop`, which stamps `task.txn_id` onto the Request
+            // directly and runs the in-txn staging gate (`route_in_tx_write`).
+            //
+            // Cross-node in-transaction statements are DEFERRED to U4/U5
+            // (cross-shard transactions): this short-circuit runs BEFORE the
+            // staging gate, so a non-owner's in-txn write ships to the owner as a
+            // plain committed write (not a `StageWrite` into `txn_overlays`) and
+            // its read carries no txn id. Carrying `txn_id` here is necessary but
+            // not sufficient — the owner would also have to STAGE (not commit) the
+            // write and flush/rollback the overlay at COMMIT/ROLLBACK. U3 is
+            // single-node only (LOCKED design §8); the ignored
+            // `cross_node_ryow_probe.rs` is the executable spec for the U4/U5 fix.
+            txn_id: None,
         };
 
         let mut responses: Vec<Response> = Vec::with_capacity(tasks.len());
diff --git a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs
index f1592faee..55779ec1f 100644
--- a/nodedb/src/control/server/pgwire/handler/routing/streaming.rs
+++ b/nodedb/src/control/server/pgwire/handler/routing/streaming.rs
@@ -82,6 +82,9 @@ impl NodeDbPgHandler {
                 tenant_id: task.tenant_id,
                 trace_id: crate::types::TraceId::ZERO,
                 database_id: task.database_id,
+                // Autocommit-only fast path (`maybe_stream_select` bails on
+                // InBlock), so this is `None` today; thread the task id honestly.
+                txn_id: task.txn_id,
             };
             gw.execute_stream(&ctx, child_plan).await
         } else {
diff --git a/nodedb/src/control/server/resp/gateway_dispatch.rs b/nodedb/src/control/server/resp/gateway_dispatch.rs
index 2c4c52be9..1c52576bc 100644
--- a/nodedb/src/control/server/resp/gateway_dispatch.rs
+++ b/nodedb/src/control/server/resp/gateway_dispatch.rs
@@ -38,6 +38,8 @@ pub(super) async fn dispatch_kv(
                 tenant_id: session.tenant_id,
                 trace_id: TraceId::generate(),
                 database_id: DatabaseId::DEFAULT,
+                // RESP (Redis) protocol has no SQL transaction blocks.
+                txn_id: None,
             };
             gw.execute(&gw_ctx, plan)
                 .await
@@ -86,6 +88,8 @@ pub(super) async fn dispatch_kv_write(
                 tenant_id: session.tenant_id,
                 trace_id: TraceId::generate(),
                 database_id: DatabaseId::DEFAULT,
+                // RESP (Redis) protocol has no SQL transaction blocks.
+                txn_id: None,
             };
             gw.execute(&gw_ctx, plan)
                 .await
diff --git a/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs b/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs
index 8fb845f47..fa58eb6ec 100644
--- a/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs
+++ b/nodedb/src/control/server/shared/ddl/neutral/collection/index_fanout.rs
@@ -130,6 +130,8 @@ pub(super) async fn backfill_on_peers(
             deadline_remaining_ms: deadline_ms,
             trace_id: trace_id.0,
             descriptor_versions: Vec::new(),
+            // Index backfill fan-out is a DDL side effect, not an interactive txn.
+            txn_id: None,
         });
         joins.push(tokio::spawn(async move {
             let outcome = transport.send_rpc(node_id, req).await;
diff --git a/nodedb/src/control/server/shared/session/lifecycle.rs b/nodedb/src/control/server/shared/session/lifecycle.rs
index bbe6a1087..1fa46f6b4 100644
--- a/nodedb/src/control/server/shared/session/lifecycle.rs
+++ b/nodedb/src/control/server/shared/session/lifecycle.rs
@@ -36,7 +36,7 @@ pub fn run_begin(
         .load(std::sync::atomic::Ordering::Acquire);
     ddl_buffer::activate();
     sessions
-        .begin(addr, snapshot_lsn, snapshot_epoch)
+        .begin(addr, snapshot_lsn, snapshot_epoch, state.node_id)
         .map_err(|msg| crate::Error::BadRequest {
             detail: msg.to_owned(),
         })
diff --git a/nodedb/src/control/server/shared/session/read_set.rs b/nodedb/src/control/server/shared/session/read_set.rs
index 171d52167..eb9cd955d 100644
--- a/nodedb/src/control/server/shared/session/read_set.rs
+++ b/nodedb/src/control/server/shared/session/read_set.rs
@@ -144,7 +144,7 @@ mod tests {
         let sessions = SessionStore::new();
         let a = addr();
         sessions.ensure_session(a);
-        sessions.begin(&a, Lsn::new(5), 0).expect("begin");
+        sessions.begin(&a, Lsn::new(5), 0, 0).expect("begin");
         (sessions, a)
     }
 
diff --git a/nodedb/src/control/server/shared/session/tests.rs b/nodedb/src/control/server/shared/session/tests.rs
index e297a70a7..61b1c62e0 100644
--- a/nodedb/src/control/server/shared/session/tests.rs
+++ b/nodedb/src/control/server/shared/session/tests.rs
@@ -13,13 +13,13 @@ fn transaction_lifecycle() {
 
     assert_eq!(store.transaction_state(&addr), TransactionState::Idle);
 
-    store.begin(&addr, crate::types::Lsn::new(1), 0).unwrap();
+    store.begin(&addr, crate::types::Lsn::new(1), 0, 0).unwrap();
     assert_eq!(store.transaction_state(&addr), TransactionState::InBlock);
 
     store.commit(&addr).unwrap();
     assert_eq!(store.transaction_state(&addr), TransactionState::Idle);
 
-    store.begin(&addr, crate::types::Lsn::new(1), 0).unwrap();
+    store.begin(&addr, crate::types::Lsn::new(1), 0, 0).unwrap();
     store.fail_transaction(&addr);
     assert_eq!(store.transaction_state(&addr), TransactionState::Failed);
 
@@ -321,7 +321,7 @@ async fn multi_vshard_rollback_to_savepoint_rewinds_each_vshard() {
     let store = SessionStore::new();
     let addr: std::net::SocketAddr = "127.0.0.1:5201".parse().unwrap();
     store.ensure_session(addr);
-    store.begin(&addr, Lsn::new(1), 0).unwrap();
+    store.begin(&addr, Lsn::new(1), 0, 0).unwrap();
     let tenant = TenantId::new(1);
     let dp = RecordingDp::default();
 
diff --git a/nodedb/src/control/server/shared/session/transaction.rs b/nodedb/src/control/server/shared/session/transaction.rs
index b9d5ee795..fe6422f65 100644
--- a/nodedb/src/control/server/shared/session/transaction.rs
+++ b/nodedb/src/control/server/shared/session/transaction.rs
@@ -13,10 +13,12 @@ use super::read_set::ReadSetEntry;
 use super::state::{SavepointEntry, TransactionState};
 use super::store::SessionStore;
 
-/// Global monotonic counter minting `TxnId`s across all sessions on this
-/// shard. Unique per shard for the lifetime of the process — sufficient
-/// for keying the per-transaction staging overlay, which is scoped to a
-/// single shard's in-memory state.
+/// Per-process monotonic sequence feeding the low bits of every minted
+/// `TxnId`. On its own it is only unique within this process; combined with
+/// the origin node id via [`TxnId::from_origin`] it becomes cluster-globally
+/// unique, so two coordinators staging onto the same owner shard never collide
+/// on the same overlay key. Starts at 1 so a single-node deployment (node id 0)
+/// still mints 1, 2, 3, … exactly as before.
 static NEXT_TXN_ID: AtomicU64 = AtomicU64::new(1);
 
 impl SessionStore {
@@ -32,11 +34,18 @@ impl SessionStore {
     /// fast path) and the last globally-applied Calvin `snapshot_epoch` as the
     /// cross-shard-valid version anchor. All reads within this transaction see
     /// data as of this LSN.
+    ///
+    /// `node_id` is this coordinator's cluster node id (`SharedState.node_id`,
+    /// 0 in single-node deployments). It is packed into the minted [`TxnId`] so
+    /// the id is globally unique — required for cross-node read-your-own-write:
+    /// two coordinators staging onto a shared owner shard must not collide on
+    /// the same `txn_overlays` key.
     pub fn begin(
         &self,
         addr: &SocketAddr,
         current_lsn: Lsn,
         snapshot_epoch: u64,
+        node_id: u64,
     ) -> Result<(), &'static str> {
         self.write_session(addr, |session| match session.tx_state {
             TransactionState::Idle => {
@@ -44,7 +53,10 @@ impl SessionStore {
                 session.tx_snapshot_lsn = Some(current_lsn);
                 session.tx_snapshot_epoch = Some(snapshot_epoch);
                 session.tx_read_set.clear();
-                session.tx_id = Some(TxnId::new(NEXT_TXN_ID.fetch_add(1, Ordering::Relaxed)));
+                session.tx_id = Some(TxnId::from_origin(
+                    node_id,
+                    NEXT_TXN_ID.fetch_add(1, Ordering::Relaxed),
+                ));
                 session.tx_vshards.clear();
                 Ok(())
             }
diff --git a/nodedb/src/control/server/shuffle/producer_hook.rs b/nodedb/src/control/server/shuffle/producer_hook.rs
index 02ad97130..9bf7cefc7 100644
--- a/nodedb/src/control/server/shuffle/producer_hook.rs
+++ b/nodedb/src/control/server/shuffle/producer_hook.rs
@@ -106,6 +106,10 @@ impl RegistryShuffleProducer {
                     version: d.version,
                 })
                 .collect(),
+            // Shuffle produce is the cross-node Exchange path; in-transaction
+            // RYOW across a distributed shuffle is a separate (deferred) unit,
+            // and `ShuffleProduceRequest` carries no `txn_id`, so this is `None`.
+            txn_id: None,
         };
 
         let executor = LocalPlanExecutor::new(Arc::clone(&self.state));
diff --git a/nodedb/src/event/cdc/consume.rs b/nodedb/src/event/cdc/consume.rs
index 3980621c2..b9dfef7ea 100644
--- a/nodedb/src/event/cdc/consume.rs
+++ b/nodedb/src/event/cdc/consume.rs
@@ -250,6 +250,8 @@ pub async fn consume_remote(
         tenant_id: crate::types::TenantId::new(tenant_id),
         trace_id: nodedb_types::TraceId::generate(),
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        // CDC consumer dispatch is internal, not session-transaction-scoped.
+        txn_id: None,
     };
 
     let query_ctx = crate::control::planner::context::QueryContext::for_state(state);
diff --git a/nodedb/src/event/topic/publish.rs b/nodedb/src/event/topic/publish.rs
index a2d582a1b..671aaead7 100644
--- a/nodedb/src/event/topic/publish.rs
+++ b/nodedb/src/event/topic/publish.rs
@@ -162,6 +162,8 @@ pub async fn publish_remote(
         tenant_id: crate::types::TenantId::new(tenant_id),
         trace_id: nodedb_types::TraceId::generate(),
         database_id: nodedb_types::id::DatabaseId::DEFAULT,
+        // Topic publish dispatch is internal, not session-transaction-scoped.
+        txn_id: None,
     };
 
     let query_ctx = crate::control::planner::context::QueryContext::for_state(state);