feat(sdk): send a raw request to a chosen replica (#4639) A client could not read the state of one particular node. Each node keeps its own stream, topic and system stats, but every official SDK sends a raw request, and every typed command other than the five partition commands, to the metadata primary. The client discovers the roster and keeps a link per replica, yet the choice of link stays internal. Tools, dashboards and tests could read only the metadata primary, and a client could not spread reads that any node answers. Add one function per SDK that sends a raw request to the replica id that cluster metadata names. Rust gains `IggyClient::send_binary_request_to_replica` and `BinaryTransport::send_raw_to_replica,` Node.js `sendBinaryRequestToReplica` and a replica option on `sendCommand`, Go `SendBinaryRequestToReplica`, Java `sendBinaryRequestToReplica` on the async, blocking and HTTP clients, C# `SendBinaryRequestToReplicaAsync`, Swift `sendRawRequest(code:payload:replica:)`, and the Python, C++ and PHP bindings expose the Rust method. The request travels on that replica's link with the session's metadata watermark, so the replica answers after it applied the metadata the client saw. Replicated commands, polls and session control fail with `InvalidCommand` before anything is sent. An id the roster does not name after one roster read fails with `InvalidIdentifier`. A lost link or `TransientNotAccepted` retries the same replica until the deadline, and `NotPrimary` is returned rather than re-routed, because the caller chose the node. HTTP has no raw binary path and returns `FeatureUnavailable`, as the existing raw request does. A new function keeps every existing caller compiling, and the trait method has a default so external `BinaryTransport` implementations do too. Reusing `InvalidIdentifier` avoids a new code in every SDK's error table. No command code, wire format or server behavior changes.

diff --git a/Cargo.lock b/Cargo.lock
index ad0b607..cefd525 100644
--- a/Cargo.lock
+++ b/Cargo.lock

@@ -7089,7 +7089,7 @@
 
 [[package]]
 name = "iggy"
-version = "0.11.1-edge.3"
+version = "0.11.1-edge.4"
 dependencies = [
  "async-broadcast",
  "async-dropper",
@@ -7369,7 +7369,7 @@
 
 [[package]]
 name = "iggy_common"
-version = "0.11.1-edge.3"
+version = "0.11.1-edge.4"
 dependencies = [
  "aes-gcm 0.11.1",
  "aes-gcm-siv",

diff --git a/Cargo.toml b/Cargo.toml
index f726d0c..f9528eb 100644
--- a/Cargo.toml
+++ b/Cargo.toml

@@ -213,11 +213,11 @@
 iceberg = "0.10.1"
 iceberg-catalog-rest = "0.10.1"
 iceberg-storage-opendal = "0.10.1"
-iggy = { path = "core/sdk", version = "0.11.1-edge.3" }
+iggy = { path = "core/sdk", version = "0.11.1-edge.4" }
 iggy-cli = { path = "core/cli", version = "0.14.1-edge.3" }
 iggy-gateway-kafka = { path = "gateways/kafka" }
 iggy_binary_protocol = { path = "core/binary_protocol", version = "0.11.1-edge.3" }
-iggy_common = { path = "core/common", version = "0.11.1-edge.3" }
+iggy_common = { path = "core/common", version = "0.11.1-edge.4" }
 iggy_connector_doris_sink = { path = "core/connectors/sinks/doris_sink" }
 iggy_connector_sdk = { path = "core/connectors/sdk", version = "0.5.1-edge.2" }
 indexmap = "2.14.2"

diff --git a/bdd/cpp/features/step_definitions/raw_command_steps.cpp b/bdd/cpp/features/step_definitions/raw_command_steps.cpp
index 3957754..deb1407 100644
--- a/bdd/cpp/features/step_definitions/raw_command_steps.cpp
+++ b/bdd/cpp/features/step_definitions/raw_command_steps.cpp

@@ -44,6 +44,23 @@
     }
 }
 
+WHEN("^I send a raw command with code ([0-9]+) and an empty payload to replica ([0-9]+)$") {
+    REGEX_PARAM(std::uint32_t, code);
+    REGEX_PARAM(unsigned int, replica);
+    cucumber::ScenarioScope<bdd::GlobalContext> context;
+
+    context->raw_response.clear();
+    context->raw_error.clear();
+    try {
+        iggy::ffi::ErrorInfo error_info{};
+        const auto response = context->client->send_binary_request_to_replica(
+            code, rust::Vec<std::uint8_t>(), static_cast<std::uint8_t>(replica), error_info);
+        context->raw_response.assign(response.begin(), response.end());
+    } catch (const std::exception &error) {
+        context->raw_error = error.what();
+    }
+}
+
 THEN("^the raw command should succeed with an empty response$") {
     cucumber::ScenarioScope<bdd::GlobalContext> context;
 
@@ -64,3 +81,10 @@
     EXPECT_TRUE(context->raw_response.empty());
     EXPECT_NE(context->raw_error.find("Invalid command"), std::string::npos);
 }
+
+THEN("^the raw command should fail with an invalid identifier error$") {
+    cucumber::ScenarioScope<bdd::GlobalContext> context;
+
+    EXPECT_TRUE(context->raw_response.empty());
+    EXPECT_NE(context->raw_error.find("Invalid identifier"), std::string::npos);
+}

diff --git a/bdd/go/tests/raw_command.go b/bdd/go/tests/raw_command.go
index 290dfe3..429ec55 100644
--- a/bdd/go/tests/raw_command.go
+++ b/bdd/go/tests/raw_command.go

@@ -73,6 +73,12 @@
 	return nil
 }
 
+func (rawCommandSteps) whenSendRawCommandToReplica(ctx context.Context, code uint32, replica uint8) error {
+	state := getRawCommandCtx(ctx)
+	state.lastResponse, state.lastError = state.client.SendBinaryRequestToReplica(ctx, code, nil, replica)
+	return nil
+}
+
 func (rawCommandSteps) thenEmptyResponse(ctx context.Context) error {
 	state := getRawCommandCtx(ctx)
 	if state.lastError != nil {
@@ -102,6 +108,13 @@
 	return nil
 }
 
+func (rawCommandSteps) thenInvalidIdentifier(ctx context.Context) error {
+	if !errors.Is(getRawCommandCtx(ctx).lastError, ierror.ErrInvalidIdentifier) {
+		return fmt.Errorf("expected invalid identifier error, got %v", getRawCommandCtx(ctx).lastError)
+	}
+	return nil
+}
+
 func initRawCommandScenario(sc *godog.ScenarioContext) {
 	sc.Before(func(context.Context, *godog.Scenario) (context.Context, error) {
 		return context.WithValue(context.Background(), rawCommandCtxKey{}, &rawCommandCtx{}), nil
@@ -110,9 +123,11 @@
 	sc.Step(`I have a running Iggy server`, steps.givenRunningServer)
 	sc.Step(`I am authenticated as the root user`, steps.givenAuthenticationAsRoot)
 	sc.Step(`^I send a raw command with code (\d+) and an empty payload$`, steps.whenSendRawCommand)
+	sc.Step(`^I send a raw command with code (\d+) and an empty payload to replica (\d+)$`, steps.whenSendRawCommandToReplica)
 	sc.Step(`the raw command should succeed with an empty response`, steps.thenEmptyResponse)
 	sc.Step(`the raw command should succeed with a non-empty response`, steps.thenNonEmptyResponse)
 	sc.Step(`the raw command should fail with an invalid command error`, steps.thenInvalidCommand)
+	sc.Step(`the raw command should fail with an invalid identifier error`, steps.thenInvalidIdentifier)
 	sc.After(func(ctx context.Context, _ *godog.Scenario, scenarioError error) (context.Context, error) {
 		state := getRawCommandCtx(ctx)
 		if state.client != nil {

diff --git a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java
index 823ec2d..d52efc9 100644
--- a/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java
+++ b/bdd/java/src/test/java/org/apache/iggy/bdd/BasicMessagingSteps.java

@@ -274,6 +274,17 @@
         }
     }
 
+    @When("I send a raw command with code {int} and an empty payload to replica {int}")
+    public void sendRawCommandToReplica(int code, int replica) {
+        try {
+            context.lastRawResponse = getClient().sendBinaryRequestToReplica(code, new byte[0], replica);
+            context.lastRawError = null;
+        } catch (RuntimeException error) {
+            context.lastRawResponse = null;
+            context.lastRawError = error;
+        }
+    }
+
     @Then("the raw command should succeed with an empty response")
     public void rawCommandSucceedsWithEmptyResponse() {
         assertNull(context.lastRawError, "Raw command should succeed");
@@ -296,6 +307,14 @@
         assertEquals(IggyErrorCode.INVALID_COMMAND, error.getErrorCode());
     }
 
+    @Then("the raw command should fail with an invalid identifier error")
+    public void rawCommandFailsWithInvalidIdentifier() {
+        assertNull(context.lastRawResponse, "Raw response should be absent");
+        assertTrue(context.lastRawError instanceof IggyServerException, "Expected an Iggy server error");
+        IggyServerException error = (IggyServerException) context.lastRawError;
+        assertEquals(IggyErrorCode.INVALID_IDENTIFIER, error.getErrorCode());
+    }
+
     @Given("I have a cluster node that cannot reach quorum")
     public void clusterNodeWithoutQuorum() {
         HostPort hostPort = HostPort.parse(TestEnvironment.noQuorumAddress());

diff --git a/bdd/php/tests/RawCommandFeatureTest.php b/bdd/php/tests/RawCommandFeatureTest.php
index 181ca9b..0aed415 100644
--- a/bdd/php/tests/RawCommandFeatureTest.php
+++ b/bdd/php/tests/RawCommandFeatureTest.php

@@ -75,6 +75,20 @@
             }
             return;
         }
+        if (preg_match('/^I send a raw command with code (\d+) and an empty payload to replica (\d+)$/', $step, $matches) === 1) {
+            try {
+                $this->lastResponse = $this->requireClient()->sendBinaryRequestToReplica(
+                    (int) $matches[1],
+                    '',
+                    (int) $matches[2],
+                );
+                $this->lastError = null;
+            } catch (Throwable $error) {
+                $this->lastResponse = null;
+                $this->lastError = $error;
+            }
+            return;
+        }
         if ($step === 'the raw command should succeed with an empty response') {
             assert_same(null, $this->lastError);
             assert_same('', $this->lastResponse);
@@ -91,6 +105,12 @@
             assert_true(str_contains(strtolower((string) $this->lastError?->getMessage()), 'invalid command'));
             return;
         }
+        if ($step === 'the raw command should fail with an invalid identifier error') {
+            assert_same(null, $this->lastResponse);
+            assert_instance_of(IggyException::class, $this->lastError);
+            assert_true(str_contains(strtolower((string) $this->lastError?->getMessage()), 'invalid identifier'));
+            return;
+        }
 
         self::fail("Unsupported BDD step: {$step}");
     }

diff --git a/bdd/python/tests/test_raw_command.py b/bdd/python/tests/test_raw_command.py
index e2b040d..94ff16f 100644
--- a/bdd/python/tests/test_raw_command.py
+++ b/bdd/python/tests/test_raw_command.py

@@ -63,6 +63,26 @@
     asyncio.run(send())
 
 
+@when(
+    parsers.parse(
+        "I send a raw command with code {code:d} "
+        "and an empty payload to replica {replica:d}"
+    )
+)
+def send_raw_command_to_replica(context, code, replica):
+    async def send():
+        try:
+            context.last_raw_response = (
+                await context.client.send_binary_request_to_replica(code, b"", replica)
+            )
+            context.last_raw_error = None
+        except RuntimeError as error:
+            context.last_raw_response = None
+            context.last_raw_error = error
+
+    asyncio.run(send())
+
+
 @then("the raw command should succeed with an empty response")
 def raw_command_succeeds_with_empty_response(context):
     assert context.last_raw_error is None
@@ -80,3 +100,10 @@
     assert context.last_raw_response is None
     assert context.last_raw_error is not None
     assert "invalid command" in str(context.last_raw_error).lower()
+
+
+@then("the raw command should fail with an invalid identifier error")
+def raw_command_fails_with_invalid_identifier(context):
+    assert context.last_raw_response is None
+    assert context.last_raw_error is not None
+    assert "invalid identifier" in str(context.last_raw_error).lower()

diff --git a/bdd/python/uv.lock b/bdd/python/uv.lock
index 3f1a79c..10daa23 100644
--- a/bdd/python/uv.lock
+++ b/bdd/python/uv.lock

@@ -8,7 +8,7 @@
 
 [[package]]
 name = "apache-iggy"
-version = "0.9.1.dev3"
+version = "0.9.1.dev4"
 source = { directory = "../../foreign/python" }
 
 [package.metadata]

diff --git a/bdd/rust/tests/metadata_changes.rs b/bdd/rust/tests/metadata_changes.rs
index e3a979a..390b7df 100644
--- a/bdd/rust/tests/metadata_changes.rs
+++ b/bdd/rust/tests/metadata_changes.rs

@@ -24,7 +24,10 @@
 
 #[tokio::main]
 async fn main() {
+    // Every scenario follows the one server-wide change log, and a stream id one scenario
+    // frees can be reused by another, so the scenarios run one at a time.
     GlobalContext::cucumber()
+        .max_concurrent_scenarios(1)
         .fail_on_skipped()
         .run_and_exit("../../bdd/scenarios/metadata_changes.feature")
         .await;

diff --git a/bdd/rust/tests/steps/metadata_changes.rs b/bdd/rust/tests/steps/metadata_changes.rs
index 9b1604f..9b971e2 100644
--- a/bdd/rust/tests/steps/metadata_changes.rs
+++ b/bdd/rust/tests/steps/metadata_changes.rs

@@ -102,9 +102,7 @@
         if page.has_more {
             continue;
         }
-        let names_stream =
-            |record: &&MetadataChangeRecord| change_stream_id(record) == Some(stream_id);
-        if records.iter().filter(names_stream).count() >= expected.len() {
+        if records_of_stream(&records, stream_id).len() >= expected.len() {
             break;
         }
         sleep(FOLLOW_INTERVAL).await;
@@ -113,10 +111,7 @@
         let group = record.change.group().expect("known group");
         assert!(world.last_change_groups.contains(&group), "{record:?}");
     }
-    let mine: Vec<MetadataChangeRecord> = records
-        .into_iter()
-        .filter(|record| change_stream_id(record) == Some(stream_id))
-        .collect();
+    let mine = records_of_stream(&records, stream_id);
     let kinds: Vec<MetadataChangeKind> = mine
         .iter()
         .filter_map(|record| record.change.kind())
@@ -323,6 +318,31 @@
     options.clone()
 }
 
+/// The followed records of this scenario's stream. A freed stream id can name
+/// a later stream, so when the records show a creation of `stream_id`, only the
+/// records of its last creation, matched by the op that created the stream,
+/// count.
+fn records_of_stream(
+    records: &[MetadataChangeRecord],
+    stream_id: u32,
+) -> Vec<MetadataChangeRecord> {
+    let created_op = records
+        .iter()
+        .rev()
+        .find_map(|record| match &record.change {
+            MetadataChange::StreamCreated(stream) if stream.stream_id == stream_id => {
+                Some(stream.stream_created_op)
+            }
+            _ => None,
+        });
+    records
+        .iter()
+        .filter(|record| change_stream_id(record) == Some(stream_id))
+        .filter(|record| created_op.is_none_or(|op| change_stream_created_op(record) == Some(op)))
+        .cloned()
+        .collect()
+}
+
 const fn change_stream_id(record: &MetadataChangeRecord) -> Option<u32> {
     match &record.change {
         MetadataChange::StreamCreated(stream) | MetadataChange::StreamUpdated(stream) => {
@@ -337,3 +357,18 @@
         _ => None,
     }
 }
+
+const fn change_stream_created_op(record: &MetadataChangeRecord) -> Option<u64> {
+    match &record.change {
+        MetadataChange::StreamCreated(stream) | MetadataChange::StreamUpdated(stream) => {
+            Some(stream.stream_created_op)
+        }
+        MetadataChange::StreamDeleted(stream) => Some(stream.stream_created_op),
+        MetadataChange::TopicCreated(topic) | MetadataChange::TopicUpdated(topic) => {
+            Some(topic.stream_created_op)
+        }
+        MetadataChange::TopicDeleted(topic) => Some(topic.stream_created_op),
+        MetadataChange::PartitionsChanged(partitions) => Some(partitions.stream_created_op),
+        _ => None,
+    }
+}

diff --git a/bdd/rust/tests/steps/raw_command.rs b/bdd/rust/tests/steps/raw_command.rs
index 72676cd..5e97fe8 100644
--- a/bdd/rust/tests/steps/raw_command.rs
+++ b/bdd/rust/tests/steps/raw_command.rs

@@ -26,6 +26,16 @@
     world.last_raw_result = Some(client.send_binary_request(code, Bytes::new()).await);
 }
 
+#[when(regex = r"^I send a raw command with code (\d+) and an empty payload to replica (\d+)$")]
+pub async fn when_send_raw_command_to_replica(world: &mut GlobalContext, code: u32, replica: u8) {
+    let client = world.client.as_ref().expect("Client should be available");
+    world.last_raw_result = Some(
+        client
+            .send_binary_request_to_replica(code, Bytes::new(), replica)
+            .await,
+    );
+}
+
 #[then("the raw command should succeed with an empty response")]
 pub async fn then_raw_command_succeeds_with_empty_response(world: &mut GlobalContext) {
     let response = world
@@ -58,3 +68,14 @@
         .expect_err("Raw command should have failed");
     assert_eq!(*error, IggyError::InvalidCommand);
 }
+
+#[then("the raw command should fail with an invalid identifier error")]
+pub async fn then_raw_command_fails_with_invalid_identifier_error(world: &mut GlobalContext) {
+    let error = world
+        .last_raw_result
+        .as_ref()
+        .expect("Should have sent a raw command")
+        .as_ref()
+        .expect_err("Raw command should have failed");
+    assert_eq!(*error, IggyError::InvalidIdentifier);
+}

diff --git a/bdd/scenarios/raw_command.feature b/bdd/scenarios/raw_command.feature
index 158c35d..aacdc0a 100644
--- a/bdd/scenarios/raw_command.feature
+++ b/bdd/scenarios/raw_command.feature

@@ -39,3 +39,13 @@
   Scenario: Session control command code is rejected
     When I send a raw command with code 38 and an empty payload
     Then the raw command should fail with an invalid command error
+
+  Scenario: Raw command to a chosen replica
+    When I send a raw command with code 10 and an empty payload to replica 0
+    Then the raw command should succeed with a non-empty response
+
+    When I send a raw command with code 10 and an empty payload to replica 9
+    Then the raw command should fail with an invalid identifier error
+
+    When I send a raw command with code 202 and an empty payload to replica 0
+    Then the raw command should fail with an invalid command error

diff --git a/core/common/Cargo.toml b/core/common/Cargo.toml
index 09c5008..b7f8c81 100644
--- a/core/common/Cargo.toml
+++ b/core/common/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy_common"
-version = "0.11.1-edge.3"
+version = "0.11.1-edge.4"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2024"
 rust-version.workspace = true

diff --git a/core/common/src/traits/binary_transport.rs b/core/common/src/traits/binary_transport.rs
index e7d9dcf..2a95c3a 100644
--- a/core/common/src/traits/binary_transport.rs
+++ b/core/common/src/traits/binary_transport.rs

@@ -149,6 +149,28 @@
         payload: Bytes,
         context: PartitionContext,
     ) -> Result<Bytes, IggyError>;
+    /// Send a non-replicated command `code` with an encoded `payload` to the
+    /// replica `replica_id` and return its raw reply body.
+    ///
+    /// The request carries the session's metadata watermark, so the replica
+    /// answers once it applied the metadata the session saw. The reply is the
+    /// chosen replica's own: a refusal such as [`IggyError::NotPrimary`] is
+    /// returned, never retried on another replica. A replicated command, a poll
+    /// or a consumer-offset read returns [`IggyError::InvalidCommand`] before
+    /// anything is sent. A replica the roster does not name, or names without
+    /// an endpoint for this transport, returns [`IggyError::InvalidIdentifier`].
+    /// A link lost after the request was written makes the outcome unknown, so
+    /// a call that no later attempt answers returns
+    /// [`IggyError::TransientNotCommitted`]. Transports without replica links
+    /// return [`IggyError::FeatureUnavailable`].
+    async fn send_raw_to_replica(
+        &self,
+        _code: u32,
+        _payload: Bytes,
+        _replica_id: u8,
+    ) -> Result<Bytes, IggyError> {
+        Err(IggyError::FeatureUnavailable)
+    }
     /// Route a store or delete offset request while retaining the membership connection.
     /// Every retry keeps a captured `context`. Without one, the request takes the context
     /// its route reports.

diff --git a/core/integration/tests/cluster/mod.rs b/core/integration/tests/cluster/mod.rs
index 58ca9b9..2782866 100644
--- a/core/integration/tests/cluster/mod.rs
+++ b/core/integration/tests/cluster/mod.rs

@@ -33,6 +33,7 @@
 mod partition_primary_routing;
 mod partition_state_transfer;
 mod register_forwarding;
+mod replica_raw_request;
 mod replica_read_batching;
 mod rolling_sigkill_repro;
 mod root_credentials;

diff --git a/core/integration/tests/cluster/replica_raw_request.rs b/core/integration/tests/cluster/replica_raw_request.rs
new file mode 100644
index 0000000..5ba9614
--- /dev/null
+++ b/core/integration/tests/cluster/replica_raw_request.rs

@@ -0,0 +1,122 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use bytes::Bytes;
+use iggy::prelude::*;
+use iggy_binary_protocol::WireDecode;
+use iggy_binary_protocol::codes::{CREATE_STREAM_CODE, GET_STATS_CODE, GET_STREAMS_CODE};
+use iggy_binary_protocol::responses::system::get_stats::StatsResponse;
+use integration::iggy_harness;
+use std::collections::BTreeSet;
+
+const STREAM: &str = "replica-raw-stream";
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_cluster_when_reading_stats_from_each_replica_should_answer_from_that_node(
+    harness: &TestHarness,
+) {
+    let client = harness.new_client().await.expect("client");
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .expect("login");
+    let replicas: Vec<u8> = client
+        .get_cluster_metadata()
+        .await
+        .expect("cluster metadata")
+        .nodes
+        .iter()
+        .map(|node| node.replica_id)
+        .collect();
+    assert_eq!(replicas.len(), 3);
+
+    let mut processes = BTreeSet::new();
+    for replica in &replicas {
+        let reply = client
+            .send_binary_request_to_replica(GET_STATS_CODE, Bytes::new(), *replica)
+            .await
+            .unwrap_or_else(|error| panic!("stats of replica {replica}: {error}"));
+        let stats = StatsResponse::decode_from(&reply).expect("stats decode");
+        processes.insert(stats.process_id);
+    }
+
+    assert_eq!(
+        processes.len(),
+        replicas.len(),
+        "each replica answers from its own process"
+    );
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_a_stream_created_by_the_client_when_listing_on_each_replica_should_include_it(
+    harness: &TestHarness,
+) {
+    let client = harness.new_client().await.expect("client");
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .expect("login");
+    client
+        .create_stream(STREAM, &StreamCreateOptions::default())
+        .await
+        .expect("create stream");
+    let replicas: Vec<u8> = client
+        .get_cluster_metadata()
+        .await
+        .expect("cluster metadata")
+        .nodes
+        .iter()
+        .map(|node| node.replica_id)
+        .collect();
+
+    for replica in replicas {
+        let reply = client
+            .send_binary_request_to_replica(GET_STREAMS_CODE, Bytes::new(), replica)
+            .await
+            .unwrap_or_else(|error| panic!("streams of replica {replica}: {error}"));
+        assert!(
+            reply
+                .windows(STREAM.len())
+                .any(|window| window == STREAM.as_bytes()),
+            "replica {replica} answers after applying the stream the session created"
+        );
+    }
+}
+
+#[iggy_harness(cluster_nodes = 3)]
+async fn given_an_unknown_replica_or_a_replicated_command_when_sent_should_refuse_it(
+    harness: &TestHarness,
+) {
+    let client = harness.new_client().await.expect("client");
+    client
+        .login_user(DEFAULT_ROOT_USERNAME, DEFAULT_ROOT_PASSWORD)
+        .await
+        .expect("login");
+
+    assert_eq!(
+        client
+            .send_binary_request_to_replica(GET_STATS_CODE, Bytes::new(), 9)
+            .await,
+        Err(IggyError::InvalidIdentifier)
+    );
+    assert_eq!(
+        client
+            .send_binary_request_to_replica(CREATE_STREAM_CODE, Bytes::new(), 1)
+            .await,
+        Err(IggyError::InvalidCommand)
+    );
+}

diff --git a/core/sdk/Cargo.toml b/core/sdk/Cargo.toml
index aed6019..5dbf22d 100644
--- a/core/sdk/Cargo.toml
+++ b/core/sdk/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy"
-version = "0.11.1-edge.3"
+version = "0.11.1-edge.4"
 description = "Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 edition = "2024"
 rust-version.workspace = true

diff --git a/core/sdk/README.md b/core/sdk/README.md
index ebb5239..2aa9163 100644
--- a/core/sdk/README.md
+++ b/core/sdk/README.md

@@ -249,6 +249,14 @@
 
 For lower-level control over individual commands (login, stream/topic management, raw send, polling by offset or timestamp), use the transport-specific clients directly. See the [examples](https://github.com/apache/iggy/tree/master/examples/rust) and the [Rust SDK docs](https://iggy.apache.org/docs/sdk/rust/intro/).
 
+`IggyClient::send_binary_request_to_replica(code, payload, replica_id)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. The request carries the session's metadata watermark. Replicated commands, polls, consumer-offset reads and session-control codes fail with `InvalidCommand` before anything is sent.
+
+An id that cluster metadata does not name fails with `InvalidIdentifier`. An unreachable node is retried until the request deadline, and a `NotPrimary` reply is returned rather than sent to another node.
+
+HTTP clients return `FeatureUnavailable` for this call, as for `send_binary_request`.
+
 For `IggyConsumerConfig`, `partitions_count` controls topic creation only. An ordinary consumer uses partition `0` unless the builder's `partition_id` or the config's `with_partition_id` selects another partition. Code that previously used `partitions_count` to select an existing partition must set `partition_id` explicitly. Consumer-group assignment ignores `partition_id`.
 
 ## Metadata changes

diff --git a/core/sdk/src/binary/mod.rs b/core/sdk/src/binary/mod.rs
index 00172b2..f71eb63 100644
--- a/core/sdk/src/binary/mod.rs
+++ b/core/sdk/src/binary/mod.rs

@@ -112,6 +112,15 @@
                     .await
             }
 
+            async fn send_raw_to_replica(
+                &self,
+                code: u32,
+                payload: bytes::Bytes,
+                replica_id: u8,
+            ) -> Result<bytes::Bytes, iggy_common::IggyError> {
+                self.driver.send_to_replica(code, payload, replica_id).await
+            }
+
             async fn send_offset_write_with_response(
                 &self,
                 code: u32,

diff --git a/core/sdk/src/clients/client.rs b/core/sdk/src/clients/client.rs
index f1b5033..5fc5196 100644
--- a/core/sdk/src/clients/client.rs
+++ b/core/sdk/src/clients/client.rs

@@ -919,6 +919,74 @@
         })
     }
 
+    /// Sends a raw request to the replica `replica_id` and returns its raw
+    /// reply body.
+    ///
+    /// [`send_binary_request`](IggyClient::send_binary_request) lets the client
+    /// pick the replica, the metadata primary for any command it does not
+    /// route by partition. This function sends to the replica the caller names,
+    /// for example to read the stats each node keeps for itself. Replica ids
+    /// are the ones [`get_cluster_metadata`](crate::prelude::ClusterClient::get_cluster_metadata)
+    /// returns. The request carries the session's metadata watermark, so the
+    /// replica answers once it applied the metadata the session saw. The reply
+    /// is that replica's own, a refusal such as [`IggyError::NotPrimary`]
+    /// included: nothing is retried on another replica. A replica that is
+    /// reconnecting is waited for within the operation deadline.
+    ///
+    /// # Examples
+    ///
+    /// Read the stats of replica 1.
+    ///
+    /// ```no_run
+    /// use iggy::prelude::*;
+    /// use iggy_binary_protocol::codes::GET_STATS_CODE;
+    /// use bytes::Bytes;
+    ///
+    /// # async fn run() -> Result<(), IggyError> {
+    /// let client = IggyClient::from_connection_string(
+    ///     "iggy+tcp://user:secret@localhost:8090",
+    /// )?;
+    /// client.connect().await?;
+    ///
+    /// let stats = client
+    ///     .send_binary_request_to_replica(GET_STATS_CODE, Bytes::new(), 1)
+    ///     .await?;
+    /// # let _ = stats;
+    /// # Ok(())
+    /// # }
+    /// ```
+    ///
+    /// # Errors
+    /// [`IggyError::InvalidCommand`] before anything is sent if `code` is a
+    /// session-control code, a replicated command, a poll, or a consumer-offset
+    /// read. Writes and partition reads reach their primary through the typed
+    /// API and [`send_binary_request`](IggyClient::send_binary_request).
+    /// [`IggyError::InvalidIdentifier`] if the roster does not name
+    /// `replica_id`, or names it without an endpoint for this transport.
+    /// [`IggyError::TransientNotAccepted`] if the replica stays unreachable
+    /// until the deadline. [`IggyError::TransientNotCommitted`] if a link was
+    /// lost after the request was written and no later attempt got a reply,
+    /// because the replica may have run it. [`IggyError::FeatureUnavailable`]
+    /// on the HTTP transport, which has no binary path.
+    pub async fn send_binary_request_to_replica(
+        &self,
+        code: u32,
+        payload: Bytes,
+        replica_id: u8,
+    ) -> Result<Bytes, IggyError> {
+        if SESSION_CONTROL_CODES.contains(&code) {
+            return Err(IggyError::InvalidCommand);
+        }
+        super::with_client!(self, client => match &*client {
+            ClientWrapper::Tcp(client) => client.send_raw_to_replica(code, payload, replica_id).await,
+            ClientWrapper::Quic(client) => client.send_raw_to_replica(code, payload, replica_id).await,
+            ClientWrapper::WebSocket(client) => {
+                client.send_raw_to_replica(code, payload, replica_id).await
+            }
+            ClientWrapper::Http(_) | ClientWrapper::Iggy(_) => Err(IggyError::FeatureUnavailable),
+        })
+    }
+
     /// Invokes a HTTP endpoint and returns the raw response body.
     ///
     /// This is the HTTP counterpart to

diff --git a/core/sdk/src/driver.rs b/core/sdk/src/driver.rs
index 6c630ef..3a61faa 100644
--- a/core/sdk/src/driver.rs
+++ b/core/sdk/src/driver.rs

@@ -66,6 +66,8 @@
 pub(crate) enum RequestTarget<'a> {
     Metadata,
     Partition(PartitionTarget<'a>),
+    /// The replica the caller chose.
+    Replica(u8),
 }
 
 /// A partition request as its caller spelled the partition.
@@ -252,6 +254,26 @@
         .await
     }
 
+    /// Sends a non-replicated request to replica `replica_id` and returns its
+    /// reply. Writes and partition reads go through the routing of their own
+    /// kind: a poll or a consumer-offset read needs the partition's primary and
+    /// its incarnation, which a request to a chosen replica does not carry.
+    pub(crate) async fn send_to_replica(
+        &self,
+        code: u32,
+        payload: Bytes,
+        replica_id: u8,
+    ) -> Result<Bytes, IggyError> {
+        if operation_for_code(code) != Operation::NonReplicated
+            || code == POLL_MESSAGES_CODE
+            || code == GET_CONSUMER_OFFSET_CODE
+        {
+            return Err(IggyError::InvalidCommand);
+        }
+        self.send(code, payload, RequestTarget::Replica(replica_id))
+            .await
+    }
+
     /// Sends one request and returns the body of its reply. A poll goes
     /// through [`Self::poll`], which knows whether the poll moves its consumer.
     pub(crate) async fn send(
@@ -581,13 +603,16 @@
         }
     }
 
-    /// The link of the next attempt: to the primary of the route, else to
-    /// the metadata primary.
+    /// The link of the next attempt: to the replica the caller chose, to the
+    /// primary of the route, else to the metadata primary.
     async fn hop(
         &self,
         call: &mut Call,
         route: Option<PartitionRouteResponse>,
     ) -> Result<Hop, Next> {
+        if let Some(replica) = call.replica {
+            return self.replica_hop(call, replica).await;
+        }
         let target = route.map_or(Target::MetadataPrimary, |route| {
             Target::Replica(route.primary_replica)
         });
@@ -605,6 +630,31 @@
         Err(Next::Pause)
     }
 
+    /// The link of the replica the caller chose. A replica without a ready
+    /// link is waited for. One the roster does not name is looked up in a
+    /// fresh roster once, and then refused.
+    async fn replica_hop(&self, call: &mut Call, replica: u8) -> Result<Hop, Next> {
+        if let Ok(link) = self.links.link(Target::Replica(replica)) {
+            return Ok(Hop { link, route: None });
+        }
+        call.refusal = IggyError::TransientNotAccepted;
+        if self.links.has_replica(replica) {
+            return Err(Next::Pause);
+        }
+        if call.roster_read {
+            return Err(Next::Fail(IggyError::InvalidIdentifier));
+        }
+        call.roster_read = match self.links.link(Target::MetadataPrimary) {
+            Ok(link) => self.read_roster(&link, call.deadline).await,
+            Err(_) => false,
+        };
+        Err(if call.roster_read {
+            Next::Again
+        } else {
+            Next::Pause
+        })
+    }
+
     /// Reads the outcome of one attempt. `sent` tells that this or an
     /// earlier attempt can have taken effect.
     async fn settle(
@@ -639,6 +689,8 @@
                 self.refresh(call, hop, Failure::Refused).await;
                 Err(Next::Pause)
             }
+            // The caller chose the replica, so its refusal is the answer.
+            IggyError::NotPrimary if call.replica.is_some() => Err(Next::Fail(error)),
             IggyError::NotPrimary => {
                 // Neither was admitted, so the deadline reports both as 58.
                 call.refusal = IggyError::TransientNotAccepted;
@@ -714,8 +766,12 @@
         Ok(body)
     }
 
-    /// Learns where the next attempt goes after `hop` failed one.
+    /// Learns where the next attempt goes after `hop` failed one. A call to a
+    /// chosen replica goes there again.
     async fn refresh(&self, call: &Call, hop: &Hop, failure: Failure) {
+        if call.replica.is_some() {
+            return;
+        }
         if let (Some(partition), Some(route)) = (&call.partition, &hop.route) {
             let partition_id = partition.key.partition_id;
             let replica = route.primary_replica;
@@ -884,6 +940,10 @@
     code: u32,
     payload: Bytes,
     partition: Option<PartitionCall>,
+    /// The replica the caller chose, `None` when the driver routes the call.
+    replica: Option<u8>,
+    /// A call to a replica the roster did not name read the roster once.
+    roster_read: bool,
     deadline: Instant,
     ticket: Option<LaneTicket>,
     frame: Option<Frame>,
@@ -960,17 +1020,23 @@
 
 impl Call {
     fn new(code: u32, payload: Bytes, target: RequestTarget<'_>, deadline: Instant) -> Self {
-        let partition = match target {
-            RequestTarget::Metadata => None,
-            RequestTarget::Partition(target) => Some(PartitionCall {
-                key: PartitionKey::from(&target),
-                context: target.context,
-            }),
+        let (partition, replica) = match target {
+            RequestTarget::Metadata => (None, None),
+            RequestTarget::Partition(target) => (
+                Some(PartitionCall {
+                    key: PartitionKey::from(&target),
+                    context: target.context,
+                }),
+                None,
+            ),
+            RequestTarget::Replica(replica) => (None, Some(replica)),
         };
         Self {
             code,
             payload,
             partition,
+            replica,
+            roster_read: false,
             deadline,
             ticket: None,
             frame: None,
@@ -1875,6 +1941,133 @@
     }
 
     #[tokio::test(start_paused = true)]
+    async fn given_a_chosen_replica_when_a_read_is_sent_should_reach_only_that_replica_with_the_metadata_watermark()
+     {
+        let f = Fixture::new().await;
+
+        f.driver
+            .send_to_replica(GET_STREAMS_CODE, Bytes::from_static(PAYLOAD), 2)
+            .await
+            .unwrap();
+
+        let reads = f.cluster.received(GET_STREAMS_CODE);
+        assert_eq!(endpoints(&reads), [endpoint(2)]);
+        assert_eq!(reads[0].body, PAYLOAD);
+        assert_eq!(
+            stamped_context(&reads[0]).metadata_op,
+            f.driver.session.metadata_watermark()
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn given_89_from_a_chosen_replica_when_reading_should_return_it_without_another_replica()
+    {
+        let f = Fixture::new().await;
+        f.cluster.script(
+            &endpoint(2),
+            GET_STREAMS_CODE,
+            [Answer::Refuse(IggyError::NotPrimary)],
+        );
+
+        let result = f
+            .driver
+            .send_to_replica(GET_STREAMS_CODE, Bytes::from_static(PAYLOAD), 2)
+            .await;
+
+        assert_eq!(result, Err(IggyError::NotPrimary));
+        assert_eq!(
+            endpoints(&f.cluster.received(GET_STREAMS_CODE)),
+            [endpoint(2)]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn given_58_and_a_dropped_link_from_a_chosen_replica_when_reading_should_retry_that_replica()
+     {
+        let f = Fixture::new().await;
+        f.cluster.script(
+            &endpoint(2),
+            GET_STREAMS_CODE,
+            [
+                Answer::Refuse(IggyError::TransientNotAccepted),
+                Answer::HangUp,
+            ],
+        );
+
+        f.driver
+            .send_to_replica(GET_STREAMS_CODE, Bytes::from_static(PAYLOAD), 2)
+            .await
+            .unwrap();
+
+        assert_eq!(
+            endpoints(&f.cluster.received(GET_STREAMS_CODE)),
+            [endpoint(2), endpoint(2), endpoint(2)]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn given_a_link_lost_after_the_write_when_the_chosen_replica_then_refuses_should_return_not_committed()
+     {
+        let f = Fixture::new().await;
+        f.cluster.script(
+            &endpoint(2),
+            GET_STREAMS_CODE,
+            [Answer::HangUp, Answer::Refuse(IggyError::NotPrimary)],
+        );
+
+        let result = f
+            .driver
+            .send_to_replica(GET_STREAMS_CODE, Bytes::from_static(PAYLOAD), 2)
+            .await;
+
+        assert_eq!(result, Err(IggyError::TransientNotCommitted));
+        assert_eq!(
+            endpoints(&f.cluster.received(GET_STREAMS_CODE)),
+            [endpoint(2), endpoint(2)]
+        );
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn given_a_replica_the_roster_does_not_name_when_reading_should_fail_after_one_roster_read()
+     {
+        let f = Fixture::new().await;
+        let rosters = f.cluster.count(GET_CLUSTER_METADATA_CODE);
+
+        let result = f
+            .driver
+            .send_to_replica(GET_STREAMS_CODE, Bytes::from_static(PAYLOAD), REPLICAS)
+            .await;
+
+        assert_eq!(result, Err(IggyError::InvalidIdentifier));
+        assert_eq!(f.cluster.count(GET_CLUSTER_METADATA_CODE), rosters + 1);
+        assert!(f.cluster.received(GET_STREAMS_CODE).is_empty());
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn given_a_replicated_command_or_a_partition_read_when_sent_to_a_replica_should_refuse_before_the_wire()
+     {
+        let f = Fixture::new().await;
+        let rosters = f.cluster.count(GET_CLUSTER_METADATA_CODE);
+
+        for code in [
+            CREATE_STREAM_CODE,
+            SEND_MESSAGES_CODE,
+            STORE_CONSUMER_OFFSET_CODE,
+            POLL_MESSAGES_CODE,
+            GET_CONSUMER_OFFSET_CODE,
+        ] {
+            assert_eq!(
+                f.driver
+                    .send_to_replica(code, Bytes::from_static(PAYLOAD), 1)
+                    .await,
+                Err(IggyError::InvalidCommand)
+            );
+            assert!(f.cluster.received(code).is_empty());
+        }
+        assert_eq!(f.cluster.count(GET_CLUSTER_METADATA_CODE), rosters);
+    }
+
+    #[tokio::test(start_paused = true)]
     async fn given_88_before_uncertainty_when_retrying_should_start_a_new_operation() {
         let f = Fixture::new().await;
         f.script_sends([Answer::Refuse(IggyError::LifecycleBusy)]);

diff --git a/core/sdk/src/links/replicas.rs b/core/sdk/src/links/replicas.rs
index 84f96ab..aebe460 100644
--- a/core/sdk/src/links/replicas.rs
+++ b/core/sdk/src/links/replicas.rs

@@ -404,6 +404,12 @@
             .ok_or(IggyError::NotConnected)
     }
 
+    /// Whether the last roster names replica `id` with a port for this
+    /// transport. A node without one has no link slot, so it reads as unnamed.
+    pub(crate) fn has_replica(&self, id: u8) -> bool {
+        self.shared.table().slots.contains_key(&id)
+    }
+
     /// The replicas with a link that serves the session, in id order.
     pub(crate) fn ready_replicas(&self) -> Vec<u8> {
         let epoch = self.shared.session.epoch();

diff --git a/examples/node/package-lock.json b/examples/node/package-lock.json
index ae09d22..19cc4a2 100644
--- a/examples/node/package-lock.json
+++ b/examples/node/package-lock.json

@@ -23,7 +23,7 @@
     },
     "../../foreign/node": {
       "name": "apache-iggy",
-      "version": "0.10.1-edge.3",
+      "version": "0.10.1-edge.4",
       "license": "Apache-2.0",
       "dependencies": {
         "@node-rs/xxhash": "1.7.8",

diff --git a/examples/python/uv.lock b/examples/python/uv.lock
index a986d97..7c474a0 100644
--- a/examples/python/uv.lock
+++ b/examples/python/uv.lock

@@ -8,7 +8,7 @@
 
 [[package]]
 name = "apache-iggy"
-version = "0.9.1.dev3"
+version = "0.9.1.dev4"
 source = { directory = "../../foreign/python" }
 
 [package.metadata]

diff --git a/foreign/cpp/Cargo.toml b/foreign/cpp/Cargo.toml
index 71dc6b3..9f85d8c 100644
--- a/foreign/cpp/Cargo.toml
+++ b/foreign/cpp/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-cpp"
-version = "0.1.5"
+version = "0.1.6"
 edition = "2024"
 license = "Apache-2.0"
 

diff --git a/foreign/cpp/MODULE.bazel b/foreign/cpp/MODULE.bazel
index 8830498..80c7a4e 100644
--- a/foreign/cpp/MODULE.bazel
+++ b/foreign/cpp/MODULE.bazel

@@ -17,7 +17,7 @@
 
 module(
     name = "iggy_cpp",
-    version = "0.1.5",
+    version = "0.1.6",
 )
 
 bazel_dep(name = "rules_cc", version = "0.2.26")

diff --git a/foreign/cpp/README.md b/foreign/cpp/README.md
index 88cb30a..46e1a64 100644
--- a/foreign/cpp/README.md
+++ b/foreign/cpp/README.md

@@ -112,6 +112,10 @@
 
 Direct generated-bridge callers pass a local `iggy::ffi::ErrorInfo` as the last argument to login/logout, mutations, polls, raw commands, recovery calls, and the metadata change functions. It is reset at entry and populated only when the Rust SDK returns an `IggyError`. Bridge-local argument errors leave it empty, except in the metadata change functions, which record invalid input as `invalid_command`.
 
+`send_binary_request_to_replica(code, payload, replica_id, error_info)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with `invalid_command` before anything is sent. An id that cluster metadata does not name fails with `invalid_identifier`. An unreachable node is retried until the request timeout, and `not_primary` is returned rather than sent to another node.
+
 Inspect `kind`, `code`, and, when `has_operation_id` is true, `operation_id`. Use separate values for concurrent calls. Error state is never stored globally or on the client.
 
 The `IggyException(const std::string &, const ffi::ErrorInfo &)` constructor is private. Only the SDK builds an exception with a code. Other code builds an `IggyException` from its message.

diff --git a/foreign/cpp/src/client.rs b/foreign/cpp/src/client.rs
index 0b96925..5f3cbf0 100644
--- a/foreign/cpp/src/client.rs
+++ b/foreign/cpp/src/client.rs

@@ -1495,6 +1495,32 @@
             Ok(Vec::from(response))
         })
     }
+
+    /// Sends a command code and payload to the replica with `replica_id`, as
+    /// cluster metadata names it, and returns its raw response bytes.
+    /// Replicated commands, polls, consumer-offset reads and session-control
+    /// codes return an invalid-command error, and a replica that cluster
+    /// metadata does not name returns an invalid-identifier error.
+    pub fn send_binary_request_to_replica(
+        &self,
+        code: u32,
+        payload: Vec<u8>,
+        replica_id: u8,
+        error_info: &mut ffi::ErrorInfo,
+    ) -> Result<Vec<u8>, String> {
+        *error_info = ffi::ErrorInfo::default();
+        RUNTIME.block_on(async {
+            let response = self
+                .inner
+                .send_binary_request_to_replica(code, Bytes::from(payload), replica_id)
+                .await
+                .inspect_err(|error| error_info.record(error))
+                .map_err(|error| {
+                    format!("Could not send raw command '{code}' to replica {replica_id}: {error}")
+                })?;
+            Ok(Vec::from(response))
+        })
+    }
 }
 
 /// Fingerprint of a 32-byte stream key. Any other length is refused with

diff --git a/foreign/cpp/src/lib.rs b/foreign/cpp/src/lib.rs
index 185c059..2b75025 100644
--- a/foreign/cpp/src/lib.rs
+++ b/foreign/cpp/src/lib.rs

@@ -1005,6 +1005,13 @@
             payload: Vec<u8>,
             error_info: &mut ErrorInfo,
         ) -> Result<Vec<u8>>;
+        fn send_binary_request_to_replica(
+            self: &Client,
+            code: u32,
+            payload: Vec<u8>,
+            replica_id: u8,
+            error_info: &mut ErrorInfo,
+        ) -> Result<Vec<u8>>;
 
         fn disconnect(self: &Client) -> Result<()>;
         fn shutdown(self: &Client) -> Result<()>;

diff --git a/foreign/cpp/tests/e2e/client.cpp b/foreign/cpp/tests/e2e/client.cpp
index e108794..09462c0 100644
--- a/foreign/cpp/tests/e2e/client.cpp
+++ b/foreign/cpp/tests/e2e/client.cpp

@@ -3052,3 +3052,52 @@
     EXPECT_FALSE(error_info.has_operation_id);
     EXPECT_EQ(error_info.operation_id, 0u);
 }
+
+TEST_F(LowLevelE2E_Client, SendBinaryRequestToReplicaGetStatsReturnsNonEmptyBytes) {
+    RecordProperty("description",
+                   "Returns a non-empty response body for a raw get-stats command sent to the server's own replica.");
+    constexpr std::uint32_t get_stats_command_code = 10;
+    constexpr std::uint8_t server_replica_id       = 0;
+    iggy::ffi::Client *client                      = GetLoggedInClient();
+
+    rust::Vec<std::uint8_t> empty_payload;
+    rust::Vec<std::uint8_t> response;
+    ASSERT_NO_THROW({
+        iggy::ffi::ErrorInfo error_info{};
+        response = client->send_binary_request_to_replica(get_stats_command_code, empty_payload, server_replica_id,
+                                                          error_info);
+    });
+    EXPECT_FALSE(response.empty());
+}
+
+TEST_F(LowLevelE2E_Client, SendBinaryRequestToUnknownReplicaThrows) {
+    iggy::ffi::ErrorInfo error_info{};
+    RecordProperty("description",
+                   "Rejects a replica id that cluster metadata does not name with an invalid-identifier error.");
+    constexpr std::uint32_t get_stats_command_code  = 10;
+    constexpr std::uint8_t unknown_replica_id       = 9;
+    constexpr std::uint32_t invalid_identifier_code = 6;
+    iggy::ffi::Client *client                       = GetLoggedInClient();
+
+    rust::Vec<std::uint8_t> empty_payload;
+    ASSERT_THROW(
+        client->send_binary_request_to_replica(get_stats_command_code, empty_payload, unknown_replica_id, error_info),
+        std::exception);
+    EXPECT_EQ(error_info.code, invalid_identifier_code);
+}
+
+TEST_F(LowLevelE2E_Client, SendBinaryRequestToReplicaReplicatedCommandThrows) {
+    iggy::ffi::ErrorInfo error_info{};
+    RecordProperty("description",
+                   "Rejects a replicated command sent to a replica client-side with an invalid-command error.");
+    constexpr std::uint32_t create_stream_command_code = 202;
+    constexpr std::uint8_t server_replica_id           = 0;
+    constexpr std::uint32_t invalid_command_code       = 3;
+    iggy::ffi::Client *client                          = GetLoggedInClient();
+
+    rust::Vec<std::uint8_t> empty_payload;
+    ASSERT_THROW(client->send_binary_request_to_replica(create_stream_command_code, empty_payload, server_replica_id,
+                                                        error_info),
+                 std::exception);
+    EXPECT_EQ(error_info.code, invalid_command_code);
+}

diff --git a/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs
index cf276d8..f4326d1 100644
--- a/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.BDD/StepDefinitions/RawCommandSteps.cs

@@ -47,6 +47,21 @@
         }
     }
 
+    [When(@"I send a raw command with code (\d+) and an empty payload to replica (\d+)")]
+    public async Task WhenISendARawCommandToAReplica(uint code, byte replica)
+    {
+        try
+        {
+            _context.LastRawResponse = await _context.IggyClient.SendBinaryRequestToReplicaAsync(code, [], replica);
+            _context.LastRawError = null;
+        }
+        catch (Exception error)
+        {
+            _context.LastRawResponse = null;
+            _context.LastRawError = error;
+        }
+    }
+
     [Then(@"the raw command should succeed with an empty response")]
     public void ThenTheRawCommandShouldSucceedWithAnEmptyResponse()
     {
@@ -68,4 +83,12 @@
         var error = _context.LastRawError.ShouldBeOfType<IggyInvalidStatusCodeException>();
         error.StatusCode.ShouldBe(3);
     }
+
+    [Then(@"the raw command should fail with an invalid identifier error")]
+    public void ThenTheRawCommandShouldFailWithAnInvalidIdentifierError()
+    {
+        _context.LastRawResponse.ShouldBeNull();
+        var error = _context.LastRawError.ShouldBeOfType<IggyInvalidStatusCodeException>();
+        error.StatusCode.ShouldBe(6);
+    }
 }

diff --git a/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs b/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs
index 1cbfcd0..df3801a 100644
--- a/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs
+++ b/foreign/csharp/Iggy_SDK.Tests.Integration/RawCommandTests.cs

@@ -80,4 +80,62 @@
 
         await Should.ThrowAsync<FeatureUnavailableException>(() => client.SendBinaryRequestAsync(1, []));
     }
+
+    [Test]
+    [SkipHttp]
+    [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
+    public async Task SendBinaryRequestToReplica_Tcp_ShouldReturnRawPayloadsOfEveryReplica(Protocol protocol)
+    {
+        var client = await Fixture.CreateAuthenticatedClient(protocol);
+        var metadata = await client.GetClusterMetadataAsync();
+
+        metadata.ShouldNotBeNull();
+        metadata.Nodes.ShouldNotBeEmpty();
+        foreach (var node in metadata.Nodes)
+        {
+            var statsResponse = await client.SendBinaryRequestToReplicaAsync(10, [], node.ReplicaId);
+            statsResponse.ShouldNotBeEmpty();
+        }
+    }
+
+    [Test]
+    [SkipHttp]
+    [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
+    public async Task SendBinaryRequestToReplica_Tcp_ShouldRejectUnknownReplica(Protocol protocol)
+    {
+        var client = await Fixture.CreateAuthenticatedClient(protocol);
+        var metadata = await client.GetClusterMetadataAsync();
+        metadata.ShouldNotBeNull();
+        var unknown = (byte)(metadata.Nodes.Max(node => node.ReplicaId) + 1);
+
+        var exception = await Should.ThrowAsync<IggyInvalidStatusCodeException>(() =>
+            client.SendBinaryRequestToReplicaAsync(10, [], unknown));
+
+        exception.StatusCode.ShouldBe(6);
+    }
+
+    [Test]
+    [SkipHttp]
+    [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
+    public async Task SendBinaryRequestToReplica_Tcp_ShouldRejectReplicatedCommands(Protocol protocol)
+    {
+        var client = await Fixture.CreateAuthenticatedClient(protocol);
+
+        foreach (var code in new uint[] { 100, 202 })
+        {
+            var exception = await Should.ThrowAsync<IggyInvalidStatusCodeException>(() =>
+                client.SendBinaryRequestToReplicaAsync(code, [], 0));
+            exception.StatusCode.ShouldBe(3);
+        }
+    }
+
+    [Test]
+    [SkipTcp]
+    [MethodDataSource<IggyServerFixture>(nameof(IggyServerFixture.ProtocolData))]
+    public async Task SendBinaryRequestToReplica_Http_ShouldThrowFeatureUnavailable(Protocol protocol)
+    {
+        var client = await Fixture.CreateAuthenticatedClient(protocol);
+
+        await Should.ThrowAsync<FeatureUnavailableException>(() => client.SendBinaryRequestToReplicaAsync(10, [], 0));
+    }
 }

diff --git a/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs b/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs
index ff93f88..81d14cf 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/IIggyClient.cs

@@ -74,6 +74,26 @@
     Task<byte[]> SendBinaryRequestAsync(uint code, byte[] payload, CancellationToken token = default);
 
     /// <summary>
+    ///     Sends a command code with a payload to the replica with the given id and returns its raw response bytes.
+    /// </summary>
+    /// <remarks>
+    ///     The replica id is the one that cluster metadata names for the node. Every node answers non-replicated
+    ///     commands such as GetStats from its own state, so this reads the state of the chosen node. The request carries
+    ///     the metadata watermark of the session. Replicated commands, PollMessages, GetConsumerOffset and session-control
+    ///     codes are rejected with an invalid-command error before anything is sent. An id that cluster metadata does not name
+    ///     after one read of it fails with an invalid-identifier error. A replica that cannot be reached, or that answers
+    ///     with TransientNotAccepted, is asked again until the request timeout. Any other refusal, NotPrimary included, is
+    ///     returned and never sent to another replica. HTTP clients report that this operation is unavailable.
+    /// </remarks>
+    /// <param name="code">The numeric command code to send.</param>
+    /// <param name="payload">The request payload.</param>
+    /// <param name="replicaId">The replica id that cluster metadata names.</param>
+    /// <param name="token">The cancellation token to cancel the operation.</param>
+    /// <returns>A task that represents the asynchronous operation and returns the raw response payload bytes.</returns>
+    Task<byte[]> SendBinaryRequestToReplicaAsync(uint code, byte[] payload, byte replicaId,
+        CancellationToken token = default);
+
+    /// <summary>
     ///     Replays a retained operation within one timeout. Success proves commit; a proved business rejection
     ///     returns its original error and frees the lane. An unknown outcome keeps the operation retained.
     /// </summary>

diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
index c027675..3bc8a50 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs

@@ -620,6 +620,21 @@
         throw new FeatureUnavailableException();
     }
 
+    /// <summary>
+    ///     This method is only supported in TCP protocol
+    /// </summary>
+    /// <param name="code">The numeric command code to send.</param>
+    /// <param name="payload">The opaque request payload.</param>
+    /// <param name="replicaId">The replica id that cluster metadata names.</param>
+    /// <param name="token">The cancellation token to cancel the operation.</param>
+    /// <returns>A task representing the asynchronous operation.</returns>
+    /// <exception cref="FeatureUnavailableException"></exception>
+    public Task<byte[]> SendBinaryRequestToReplicaAsync(uint code, byte[] payload, byte replicaId,
+        CancellationToken token = default)
+    {
+        throw new FeatureUnavailableException();
+    }
+
     /// <inheritdoc />
     public Task ConnectAsync(CancellationToken token = default)
     {

diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs
index daba7e9..18e859a 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.PollRouting.cs

@@ -177,7 +177,7 @@
 
     private async Task<IMemoryOwner<byte>> RunClientOperationAsync(int code, ReadOnlyMemory<byte> payload,
         PartitionContext? context, SendTarget? target, CancellationToken token, ClientOperation? retained = null,
-        TimeSpan? timeout = null, ulong? startedUnder = null)
+        TimeSpan? timeout = null, ulong? startedUnder = null, byte? replica = null)
     {
         using var cancellation = CancellationTokenSource.CreateLinkedTokenSource(token, _clientLifetime.Token);
         var budget = timeout ?? TimeSpan.FromMilliseconds(VsrRequestTimeoutMs);
@@ -200,6 +200,7 @@
         var routingRetries = 0;
         var lastWasNotPrimary = false;
         var askMetadataPrimaryFirst = false;
+        var rosterRead = false;
         var terminal = false;
         var frameAcquired = false;
         var startOver = false;
@@ -303,6 +304,31 @@
                         }
                         connection = primary;
                     }
+                    else if (replica is { } chosen)
+                    {
+                        if (await FindReadyReplicaAsync(chosen, cancellation.Token) is not { } ready)
+                        {
+                            bool known;
+                            lock (_pollRoutingGate)
+                            {
+                                known = _replicaLinks.ContainsKey(chosen);
+                            }
+                            if (!known && rosterRead)
+                            {
+                                throw VsrError.Exception(VsrError.INVALID_IDENTIFIER,
+                                    $"Cluster metadata names no replica {chosen}.");
+                            }
+                            if (!known)
+                            {
+                                rosterRead = true;
+                                await RefreshPartitionRosterAsync(deadline, cancellation.Token);
+                                continue;
+                            }
+                            await Task.Delay(PollRoutingRetryDelay(routingRetries++, Random.Shared.NextDouble()), cancellation.Token);
+                            continue;
+                        }
+                        connection = ready;
+                    }
                     else if (_metadataReplica is { } metadataPrimary)
                     {
                         if (await FindReadyReplicaAsync(metadataPrimary, cancellation.Token) is not { } primary)
@@ -424,7 +450,8 @@
                     await Task.Delay(PollRoutingRetryDelay(routingRetries++, Random.Shared.NextDouble()), cancellation.Token);
                 }
                 catch (IggyInvalidStatusCodeException error) when (error.StatusCode is
-                    VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.NOT_PRIMARY or VsrError.TRANSIENT_NOT_COMMITTED)
+                    VsrError.TRANSIENT_NOT_ACCEPTED or VsrError.TRANSIENT_NOT_COMMITTED
+                    || error.StatusCode == VsrError.NOT_PRIMARY && replica is null)
                 {
                     if (generation != _consensusSession.Generation && retained is null
                         && operation is not { Uncertain: true } && ResumesUnderNewSession(generation))
@@ -464,7 +491,7 @@
                         RemovePartitionRoute(failed.Key, error.StatusCode == VsrError.NOT_PRIMARY ? route : null);
                     }
                     await Task.Delay(pause, cancellation.Token);
-                    if (target is null)
+                    if (target is null && replica is null)
                     {
                         try
                         {
@@ -550,7 +577,8 @@
             {
                 ExceptionDispatchInfo.Throw(refusal);
             }
-            throw _consensusSession.Ended(generation) ?? PollNotAccepted();
+            throw _consensusSession.Ended(generation)
+                ?? (replica is { } chosen ? ReplicaNotAccepted(chosen) : PollNotAccepted());
         }
         catch (Exception error) when (error is IOException or SocketException
             && refusal is not null
@@ -993,4 +1021,9 @@
     {
         return VsrError.Exception(VsrError.TRANSIENT_NOT_ACCEPTED, "The primary poll was not admitted.");
     }
+
+    private static IggyInvalidStatusCodeException ReplicaNotAccepted(byte replica)
+    {
+        return VsrError.Exception(VsrError.TRANSIENT_NOT_ACCEPTED, $"Replica {replica} did not answer before the deadline.");
+    }
 }

diff --git a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
index fce9166..592517f 100644
--- a/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs
+++ b/foreign/csharp/Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs

@@ -621,6 +621,24 @@
     }
 
     /// <inheritdoc />
+    public async Task<byte[]> SendBinaryRequestToReplicaAsync(uint code, byte[] payload, byte replicaId,
+        CancellationToken token = default)
+    {
+        if (SessionControlCodes.Contains(code) || code == CommandCodes.POLL_MESSAGES_CODE
+            || code == CommandCodes.GET_CONSUMER_OFFSET_CODE
+            || VsrOperations.ForCode((int)code) != VsrOperation.NonReplicated)
+        {
+            throw VsrError.Exception(VsrError.INVALID_COMMAND,
+                $"Command {code} cannot be sent to a chosen replica.");
+        }
+
+        using IMemoryOwner<byte> result = await RunClientOperationAsync((int)code, payload, null, null, token,
+            replica: replicaId);
+
+        return result.Memory.Length == 0 ? [] : result.Memory.Span.ToArray();
+    }
+
+    /// <inheritdoc />
     public Task ConnectAsync(CancellationToken token = default)
     {
         if (_configuration.ReconnectionSettings.Enabled && !_configuration.AutoLoginSettings.Enabled)

diff --git a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
index 6d01e0f..ddb5cf2 100644
--- a/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj
+++ b/foreign/csharp/Iggy_SDK/Iggy_SDK.csproj

@@ -26,7 +26,7 @@
         <TargetFrameworks>net8.0;net10.0</TargetFrameworks>
         <AssemblyName>Apache.Iggy</AssemblyName>
         <RootNamespace>Apache.Iggy</RootNamespace>
-        <Version>0.9.1-edge.3</Version>
+        <Version>0.9.1-edge.4</Version>
         <GenerateDocumentationFile>true</GenerateDocumentationFile>
     </PropertyGroup>
 

diff --git a/foreign/csharp/Iggy_SDK_Tests/VsrTests/ReplicaRequestTests.cs b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ReplicaRequestTests.cs
new file mode 100644
index 0000000..b5a408b
--- /dev/null
+++ b/foreign/csharp/Iggy_SDK_Tests/VsrTests/ReplicaRequestTests.cs

@@ -0,0 +1,229 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+using System.Buffers.Binary;
+using System.Collections.Concurrent;
+using Apache.Iggy.Configuration;
+using Apache.Iggy.Contracts;
+using Apache.Iggy.Enums;
+using Apache.Iggy.Exceptions;
+using Apache.Iggy.IggyClient.Implementations;
+using Apache.Iggy.Utils;
+using Apache.Iggy.Vsr;
+using Microsoft.Extensions.Logging.Abstractions;
+using static Apache.Iggy.Tests.VsrTests.MockFrames;
+
+namespace Apache.Iggy.Tests.VsrTests;
+
+/// <summary>
+///     Raw requests that name the replica to answer them. Mirrors the replica request tests of
+///     <c>core/sdk/src/driver.rs</c>.
+/// </summary>
+public sealed class ReplicaRequestTests
+{
+    private const uint GetStatsCode = 10;
+    private const uint NotPrimary = 89;
+    private const ulong MetadataCommit = 7;
+    private static readonly byte[] LeaderReply = "leader"u8.ToArray();
+    private static readonly byte[] BackupReply = "backup"u8.ToArray();
+
+    [Fact]
+    public async Task given_two_replicas_when_a_raw_request_names_each_should_reach_it_with_the_session_watermark()
+    {
+        var token = TestContext.Current.CancellationToken;
+        using var cluster = new TwoReplicas(_ => Reply(OPERATION_NON_REPLICATED, BackupReply));
+        using var client = await cluster.SignedInAsync();
+        await client.SendBinaryRequestAsync((uint)CommandCodes.CREATE_STREAM_CODE, [], token);
+
+        var backup = await client.SendBinaryRequestToReplicaAsync(GetStatsCode, [1], 1, token);
+        var leader = await client.SendBinaryRequestToReplicaAsync(GetStatsCode, [0], 0, token);
+
+        Assert.Equal(BackupReply, backup);
+        Assert.Equal(LeaderReply, leader);
+        var backupRequest = Assert.Single(cluster.BackupStats);
+        Assert.Equal([1], backupRequest.Body);
+        Assert.Equal(MetadataCommit, backupRequest.Context.MetadataOp);
+        var leaderRequest = Assert.Single(cluster.LeaderStats);
+        Assert.Equal([0], leaderRequest.Body);
+        Assert.Equal(MetadataCommit, leaderRequest.Context.MetadataOp);
+    }
+
+    [Fact]
+    public async Task given_a_backup_that_answers_not_primary_when_a_raw_request_names_it_should_return_the_refusal()
+    {
+        var token = TestContext.Current.CancellationToken;
+        using var cluster = new TwoReplicas(_ => Reply(OPERATION_NON_REPLICATED, [], NotPrimary));
+        using var client = await cluster.SignedInAsync();
+
+        var error = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() =>
+            client.SendBinaryRequestToReplicaAsync(GetStatsCode, [], 1, token));
+
+        Assert.Equal((int)NotPrimary, error.StatusCode);
+        Assert.Single(cluster.BackupStats);
+        Assert.Empty(cluster.LeaderStats);
+    }
+
+    [Fact]
+    public async Task given_a_backup_that_refuses_and_drops_its_link_when_a_raw_request_names_it_should_ask_it_again()
+    {
+        var token = TestContext.Current.CancellationToken;
+        using var cluster = new TwoReplicas(attempt => attempt switch
+        {
+            1 => Reply(OPERATION_NON_REPLICATED, [], TRANSIENT_NOT_ACCEPTED),
+            2 => throw new IOException("lost link"),
+            _ => Reply(OPERATION_NON_REPLICATED, BackupReply)
+        });
+        using var client = await cluster.SignedInAsync();
+
+        var backup = await client.SendBinaryRequestToReplicaAsync(GetStatsCode, [], 1, token);
+
+        Assert.Equal(BackupReply, backup);
+        Assert.Equal(3, cluster.BackupStats.Count);
+        Assert.Empty(cluster.LeaderStats);
+    }
+
+    [Fact]
+    public async Task given_an_id_the_roster_does_not_name_when_a_raw_request_names_it_should_fail_after_one_roster_read()
+    {
+        var token = TestContext.Current.CancellationToken;
+        using var cluster = new TwoReplicas(_ => Reply(OPERATION_NON_REPLICATED, BackupReply));
+        using var client = await cluster.SignedInAsync();
+        var rosterReads = cluster.Leader.Requests(GET_CLUSTER_METADATA_CODE)
+            + cluster.Backup.Requests(GET_CLUSTER_METADATA_CODE);
+
+        var error = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() =>
+            client.SendBinaryRequestToReplicaAsync(GetStatsCode, [], 2, token));
+
+        Assert.Equal(VsrError.INVALID_IDENTIFIER, error.StatusCode);
+        Assert.Equal(rosterReads + 1, cluster.Leader.Requests(GET_CLUSTER_METADATA_CODE)
+            + cluster.Backup.Requests(GET_CLUSTER_METADATA_CODE));
+        Assert.Empty(cluster.BackupStats);
+        Assert.Empty(cluster.LeaderStats);
+    }
+
+    [Fact]
+    public async Task given_a_replicated_partition_read_or_session_command_when_sent_to_a_replica_should_refuse_it_before_sending()
+    {
+        var token = TestContext.Current.CancellationToken;
+        using var cluster = new TwoReplicas(_ => Reply(OPERATION_NON_REPLICATED, BackupReply));
+        using var client = await cluster.SignedInAsync();
+        var requests = cluster.Recorded;
+
+        foreach (var code in new[]
+                 {
+                     (uint)CommandCodes.CREATE_STREAM_CODE, (uint)CommandCodes.POLL_MESSAGES_CODE,
+                     (uint)CommandCodes.GET_CONSUMER_OFFSET_CODE, (uint)CommandCodes.LOGIN_USER_CODE,
+                     (uint)CommandCodes.LOGOUT_USER_CODE
+                 })
+        {
+            var error = await Assert.ThrowsAsync<IggyInvalidStatusCodeException>(() =>
+                client.SendBinaryRequestToReplicaAsync(code, [], 1, token));
+            Assert.Equal(VsrError.INVALID_COMMAND, error.StatusCode);
+        }
+
+        Assert.Equal(requests, cluster.Recorded);
+        Assert.Empty(cluster.BackupStats);
+        Assert.Equal(BackupReply, await client.SendBinaryRequestToReplicaAsync(GetStatsCode, [], 1, token));
+    }
+
+    /// <summary>
+    ///     Replica 0 leads metadata and is the node the client signs in on. Replica 1 is a backup that answers the
+    ///     stats requests with what the test scripts for each attempt, numbered from 1.
+    /// </summary>
+    private sealed class TwoReplicas : IDisposable
+    {
+        internal readonly MockNode Leader = new();
+        internal readonly MockNode Backup = new();
+        internal readonly ConcurrentQueue<MockRequest> LeaderStats = new();
+        internal readonly ConcurrentQueue<MockRequest> BackupStats = new();
+        private int _recorded;
+
+        internal TwoReplicas(Func<int, byte[]> backupAnswer)
+        {
+            Leader.Serve(request =>
+            {
+                Record(request);
+                if (request.Code == (int)GetStatsCode)
+                {
+                    LeaderStats.Enqueue(request);
+                    return Reply(OPERATION_NON_REPLICATED, LeaderReply);
+                }
+                return Handle(request);
+            });
+            Backup.Serve(request =>
+            {
+                Record(request);
+                if (request.Code == (int)GetStatsCode)
+                {
+                    BackupStats.Enqueue(request);
+                    return backupAnswer(BackupStats.Count);
+                }
+                return Handle(request);
+            });
+        }
+
+        /// <summary>How many requests reached a node, leaving out the sign-in, link binds, pings and roster reads.</summary>
+        internal int Recorded => Volatile.Read(ref _recorded);
+
+        internal async Task<TcpMessageStream> SignedInAsync()
+        {
+            var client = new TcpMessageStream(new IggyClientConfigurator
+            {
+                BaseAddress = $"127.0.0.1:{Leader.Port}",
+                Protocol = Protocol.Tcp,
+                HeartbeatInterval = TimeSpan.FromMinutes(1)
+            }, NullLoggerFactory.Instance);
+            await client.ConnectAsync(TestContext.Current.CancellationToken);
+            await client.LoginUserAsync("iggy", "iggy", TestContext.Current.CancellationToken);
+            return client;
+        }
+
+        public void Dispose()
+        {
+            Leader.Dispose();
+            Backup.Dispose();
+        }
+
+        private void Record(MockRequest request)
+        {
+            if (request.Operation != OPERATION_REGISTER
+                && request.Code is not (BIND_SESSION_CODE or PING_CODE or GET_CLUSTER_METADATA_CODE))
+            {
+                Interlocked.Increment(ref _recorded);
+            }
+        }
+
+        private byte[] Handle(MockRequest request)
+        {
+            if (request.Code == GET_CLUSTER_METADATA_CODE)
+            {
+                var body = new List<byte>();
+                WriteString(body, "replicas");
+                body.AddRange(BitConverter.GetBytes(2u));
+                WriteNode(body, Leader.Port, true, 0);
+                WriteNode(body, Backup.Port, false, 1);
+                return Reply(OPERATION_NON_REPLICATED, body.ToArray());
+            }
+            var reply = Answer(request);
+            if (request.Operation == (byte)VsrOperation.CreateStream)
+            {
+                BinaryPrimitives.WriteUInt64LittleEndian(reply.AsSpan(VsrHeader.REPLY_COMMIT_OFFSET), MetadataCommit);
+            }
+            return reply;
+        }
+    }
+}

diff --git a/foreign/csharp/README.md b/foreign/csharp/README.md
index 9d2f14f..501826f 100644
--- a/foreign/csharp/README.md
+++ b/foreign/csharp/README.md

@@ -348,10 +348,14 @@
 the cluster roster from that replica. When it has no ready link to the metadata primary, it does the same. If that
 roster names this replica as the metadata primary, the reply is final. Otherwise the call retries.
 
+`SendBinaryRequestToReplicaAsync(code, payload, replicaId)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with status 3 before anything is sent. An id that cluster metadata does not name fails with status 6. An unreachable node is retried until the request timeout, and a `NotPrimary` (89) reply is returned rather than sent to another node.
+
 ### Limitations
 
 - TCP uses VSR framing; HTTP uses the separate REST API. Join/leave, `GetMeAsync`, segment deletion,
-  and raw binary requests are TCP-only and throw `FeatureUnavailableException` on HTTP.
+  and raw binary requests, also to a chosen replica, are TCP-only and throw `FeatureUnavailableException` on HTTP.
 - `StoreOffsetAsync` / `DeleteOffsetAsync` need an explicit partition id under VSR: the broker does not
   resolve a `null` partition for a consumer-offset request, so passing one throws client-side.
 - Polling a missing topic throws `IggyInvalidStatusCodeException`. An existing topic with no available

diff --git a/foreign/go/README.md b/foreign/go/README.md
index aa4a884..561769e 100644
--- a/foreign/go/README.md
+++ b/foreign/go/README.md

@@ -232,6 +232,10 @@
 `SendBinaryRequest` and `ResolveOperation`. A rejected change leaves the
 credentials unchanged.
 
+`SendBinaryRequestToReplica(ctx, code, payload, replicaID)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with `ErrInvalidCommand` before anything is sent. An id that cluster metadata does not name fails with `ErrInvalidIdentifier`. An unreachable node is retried until the request deadline, and `ErrNotPrimary` is returned rather than sent to another node.
+
 If a mutation remains uncertain when its call ends, `errors.As` exposes
 `errors.OperationUncertain` with an exact `uint64` `OperationID`. The SDK retains
 that request and blocks later mutations on its lane with `errors.LaneBlocked`

diff --git a/foreign/go/client/tcp/tcp_core.go b/foreign/go/client/tcp/tcp_core.go
index 8db23df..6b4070a 100644
--- a/foreign/go/client/tcp/tcp_core.go
+++ b/foreign/go/client/tcp/tcp_core.go

@@ -324,12 +324,13 @@
 // do sends the command and returns the response body. Commands implementing
 // the appender interface encode directly into a pooled buffer.
 func (c *IggyTcpClient) do(ctx context.Context, cmd command.Command) ([]byte, error) {
-	return c.doCaptured(ctx, cmd, nil)
+	return c.doCaptured(ctx, cmd, nil, nil)
 }
 
 // doCaptured is do for a partition command that carries captured, a context
-// the caller holds, instead of the context its route names.
-func (c *IggyTcpClient) doCaptured(ctx context.Context, cmd command.Command, captured *iggcon.PartitionContext) ([]byte, error) {
+// the caller holds, instead of the context its route names, and for a command
+// sent to the replica the caller chose.
+func (c *IggyTcpClient) doCaptured(ctx context.Context, cmd command.Command, captured *iggcon.PartitionContext, replica *uint8) ([]byte, error) {
 	if ctx == nil {
 		return nil, ierror.ErrNilContext
 	}
@@ -352,9 +353,9 @@
 		if poll, ok := cmd.(*command.PollMessages); ok && poll.Strategy.Context != nil {
 			captured = poll.Strategy.Context
 		}
-		return c.drive(ctx, cmd.Code(), query, frame, captured)
+		return c.drive(ctx, cmd.Code(), query, frame, captured, nil)
 	}
-	return c.drive(ctx, cmd.Code(), nil, frame, nil)
+	return c.drive(ctx, cmd.Code(), nil, frame, nil, replica)
 }
 
 // rawRequest carries a SendBinaryRequest through do, so a raw partition
@@ -383,6 +384,19 @@
 	return c.do(ctx, rawRequest{code: command.Code(code), payload: payload})
 }
 
+// SendBinaryRequestToReplica sends a command code and payload to the replica with id
+// replicaID and returns that replica's raw response body. Session-control codes,
+// replicated commands, polls and consumer offset reads return ierror.ErrInvalidCommand
+// without writing to the connection. A replica the roster does not name, after one
+// roster read, returns ierror.ErrInvalidIdentifier. The replica's own refusal, such as
+// ierror.ErrNotPrimary, is returned as it is.
+func (c *IggyTcpClient) SendBinaryRequestToReplica(ctx context.Context, code uint32, payload []byte, replicaID uint8) ([]byte, error) {
+	if _, replicated := vsr.ReplicatedOperation(code); replicated || isSessionControlCode(code) || isPartitionRead(command.Code(code)) {
+		return nil, ierror.ErrInvalidCommand
+	}
+	return c.doCaptured(ctx, rawRequest{code: command.Code(code), payload: payload}, nil, &replicaID)
+}
+
 // reserveHeader returns buf truncated to exactly the header prologue, growing
 // it as needed. The prologue bytes are not zeroed here: EncodeRequestHeader
 // clears the full header when the frame is stamped, and zeroing twice would
@@ -391,6 +405,12 @@
 	return slices.Grow(buf[:0], vsr.HeaderSize)[:vsr.HeaderSize]
 }
 
+// isPartitionRead reports whether code reads a partition through its primary and
+// incarnation, which a request to a chosen replica does not carry.
+func isPartitionRead(code command.Code) bool {
+	return code == command.PollMessagesCode || code == command.GetOffsetCode
+}
+
 func isSessionControlCode(code uint32) bool {
 	switch code {
 	case uint32(command.LoginUserCode),

diff --git a/foreign/go/client/tcp/tcp_core_test.go b/foreign/go/client/tcp/tcp_core_test.go
index afb384a..9c3192f 100644
--- a/foreign/go/client/tcp/tcp_core_test.go
+++ b/foreign/go/client/tcp/tcp_core_test.go

@@ -252,7 +252,7 @@
 	}
 	frame, err := appendCommandFrame(nil, request)
 	require.NoError(t, err)
-	_, err = client.drive(context.Background(), request.Code(), nil, frame, nil)
+	_, err = client.drive(context.Background(), request.Code(), nil, frame, nil, nil)
 	require.NoError(t, err, "a later refusal must not escape as permission to register a fresh identity")
 	allRecorded := server.recorded()
 	recorded := allRecorded[:0]
@@ -290,7 +290,7 @@
 	join := lifecycleBusyJoin(t)
 	frame, err := appendCommandFrame(nil, join)
 	require.NoError(t, err)
-	response, err := client.drive(context.Background(), join.Code(), nil, frame, &captured)
+	response, err := client.drive(context.Background(), join.Code(), nil, frame, &captured, nil)
 	require.NoError(t, err)
 	assert.Equal(t, []byte("joined"), response)
 

diff --git a/foreign/go/client/tcp/tcp_driver.go b/foreign/go/client/tcp/tcp_driver.go
index 7c0d4a0..83cd545 100644
--- a/foreign/go/client/tcp/tcp_driver.go
+++ b/foreign/go/client/tcp/tcp_driver.go

@@ -34,7 +34,7 @@
 func (e *completedResultError) Error() string { return e.err.Error() }
 func (e *completedResultError) Unwrap() error { return e.err }
 
-func (c *IggyTcpClient) drive(ctx context.Context, code command.Code, query, frame []byte, exact *iggcon.PartitionContext) ([]byte, error) {
+func (c *IggyTcpClient) drive(ctx context.Context, code command.Code, query, frame []byte, exact *iggcon.PartitionContext, replica *uint8) ([]byte, error) {
 	ctx, cancel := c.operationContext(ctx)
 	defer cancel()
 	epoch := c.callEpoch(ctx)
@@ -144,7 +144,7 @@
 		if ticket != nil {
 			operation = ticket.freeze(code, query, frozen)
 		} else {
-			operation = &pendingOperation{id: vsr.StampedRequestID(frozen), code: code, query: append([]byte(nil), query...), fence: fence, frame: newOperationFrame(frozen, nil)}
+			operation = &pendingOperation{id: vsr.StampedRequestID(frozen), code: code, query: append([]byte(nil), query...), fence: fence, frame: newOperationFrame(frozen, nil), replica: replica}
 		}
 		response, err := c.driveOperation(ctx, operation, ticket, false, mutation)
 		var completed *completedResultError
@@ -262,6 +262,17 @@
 					link = c.links.replica(route.primary)
 				}
 			}
+		} else if operation.replica != nil {
+			link = c.links.replica(*operation.replica)
+			if link == nil && !c.links.contains(*operation.replica) {
+				if !rosterRefreshed {
+					rosterRefreshed = c.refreshRoster(ctx, nil) == nil
+				}
+				if rosterRefreshed && !c.links.contains(*operation.replica) {
+					return nil, ierror.ErrInvalidIdentifier
+				}
+				link = c.links.replica(*operation.replica)
+			}
 		} else {
 			link = c.links.metadata()
 			if link == nil && operation.code != command.GetClusterMetadataCode {
@@ -351,7 +362,7 @@
 			}
 			if operation.query != nil {
 				c.polls.invalidate(route, false)
-			} else if sent && linkReady(link) {
+			} else if sent && linkReady(link) && operation.replica == nil {
 				c.links.invalidateMetadataPrimary(link)
 			}
 			if ctx.Err() != nil {
@@ -424,9 +435,12 @@
 		case errors.Is(err, ierror.ErrTransientNotAccepted):
 			if operation.query != nil {
 				c.polls.invalidate(route, false)
-			} else if operation.code != command.GetClusterMetadataCode {
+			} else if operation.code != command.GetClusterMetadataCode && operation.replica == nil {
 				_ = c.refreshRoster(ctx, link)
 			}
+		case errors.Is(err, ierror.ErrNotPrimary) && operation.replica != nil:
+			// The caller chose the replica, so its refusal is the answer.
+			return nil, err
 		case errors.Is(err, ierror.ErrNotPrimary):
 			firstRefusal := !notPrimarySeen
 			notPrimarySeen = true

diff --git a/foreign/go/client/tcp/tcp_offset_management.go b/foreign/go/client/tcp/tcp_offset_management.go
index 2af026b..741a685 100644
--- a/foreign/go/client/tcp/tcp_offset_management.go
+++ b/foreign/go/client/tcp/tcp_offset_management.go

@@ -69,6 +69,6 @@
 }
 
 func (c *IggyTcpClient) writeOffset(ctx context.Context, cmd command.Command, captured *iggcon.PartitionContext) error {
-	_, err := c.doCaptured(ctx, cmd, captured)
+	_, err := c.doCaptured(ctx, cmd, captured, nil)
 	return err
 }

diff --git a/foreign/go/client/tcp/tcp_pending.go b/foreign/go/client/tcp/tcp_pending.go
index 5a09840..b1efffe 100644
--- a/foreign/go/client/tcp/tcp_pending.go
+++ b/foreign/go/client/tcp/tcp_pending.go

@@ -134,6 +134,8 @@
 	key   laneKey
 	fence uint64
 	frame *operationFrame
+	// replica is the replica the caller chose, nil when the client routes the request.
+	replica *uint8
 }
 
 type laneTicket struct {

diff --git a/foreign/go/client/tcp/tcp_replica_request_test.go b/foreign/go/client/tcp/tcp_replica_request_test.go
new file mode 100644
index 0000000..913cac6
--- /dev/null
+++ b/foreign/go/client/tcp/tcp_replica_request_test.go

@@ -0,0 +1,173 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package tcp
+
+import (
+	"context"
+	"encoding/binary"
+	"net"
+	"testing"
+	"time"
+
+	iggcon "github.com/apache/iggy/foreign/go/contracts"
+	ierror "github.com/apache/iggy/foreign/go/errors"
+	"github.com/apache/iggy/foreign/go/internal/command"
+	"github.com/apache/iggy/foreign/go/internal/vsr"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	firstReplicaAddress  = "127.0.0.1:19101"
+	secondReplicaAddress = "127.0.0.1:19102"
+)
+
+func TestSendBinaryRequestToReplica_SendsOnlyToTheChosenReplicaWithTheMetadataWatermark(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	client.metadataWatermark.Store(42)
+	firstServer := serve(first, func(_ int, _ request) []byte { return replyFrame(vsr.OperationNonReplicated, nil) })
+	secondServer := serve(second, func(_ int, _ request) []byte {
+		return replyFrame(vsr.OperationNonReplicated, []byte("replica-1"))
+	})
+
+	response, err := client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetStatsCode), nil, 1)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte("replica-1"), response)
+	assert.Equal(t, 0, requestCount(firstServer.recorded(), command.GetStatsCode))
+	reads := secondServer.recorded()
+	require.Equal(t, 1, requestCount(reads, command.GetStatsCode))
+	assert.Equal(t, uint64(42), binary.LittleEndian.Uint64(reads[0].header[frameOffsetContext+16:]))
+}
+
+func TestSendBinaryRequestToReplica_ReturnsNotPrimaryWithoutTryingAnotherReplica(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	firstServer := serve(first, func(_ int, _ request) []byte { return replyFrame(vsr.OperationNonReplicated, nil) })
+	secondServer := serve(second, func(_ int, _ request) []byte {
+		return statusReplyFrame(vsr.OperationNonReplicated, uint32(ierror.NotPrimaryCode), nil)
+	})
+
+	_, err := client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetStatsCode), nil, 1)
+
+	require.ErrorIs(t, err, ierror.ErrNotPrimary)
+	assert.Equal(t, 1, requestCount(secondServer.recorded(), command.GetStatsCode))
+	assert.Equal(t, 0, requestCount(firstServer.recorded(), command.GetStatsCode))
+}
+
+func TestSendBinaryRequestToReplica_RetriesATransientRefusalOnTheSameReplica(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	firstServer := serve(first, func(_ int, _ request) []byte { return replyFrame(vsr.OperationNonReplicated, nil) })
+	secondServer := serve(second, func(index int, _ request) []byte {
+		if index == 0 {
+			return statusReplyFrame(vsr.OperationNonReplicated, uint32(ierror.TransientNotAcceptedCode), nil)
+		}
+		return replyFrame(vsr.OperationNonReplicated, []byte("served"))
+	})
+
+	response, err := client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetStatsCode), nil, 1)
+
+	require.NoError(t, err)
+	assert.Equal(t, []byte("served"), response)
+	assert.Equal(t, 2, requestCount(secondServer.recorded(), command.GetStatsCode))
+	assert.Equal(t, 0, requestCount(firstServer.recorded(), command.GetStatsCode))
+}
+
+func TestSendBinaryRequestToReplica_FailsAReplicaTheRosterDoesNotNameAfterOneRosterRead(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	roster := func(_ int, read request) []byte {
+		if read.code() == uint32(command.GetClusterMetadataCode) {
+			return clusterMetadataFrame(t, 0, firstReplicaAddress, secondReplicaAddress)
+		}
+		return replyFrame(vsr.OperationNonReplicated, nil)
+	}
+	firstServer := serve(first, roster)
+	secondServer := serve(second, roster)
+
+	_, err := client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetStatsCode), nil, 9)
+
+	require.ErrorIs(t, err, ierror.ErrInvalidIdentifier)
+	reads := append(firstServer.recorded(), secondServer.recorded()...)
+	assert.Equal(t, 1, requestCount(reads, command.GetClusterMetadataCode))
+	assert.Equal(t, 0, requestCount(reads, command.GetStatsCode))
+}
+
+func TestSendBinaryRequestToReplica_FailsAnUnknownReplicaRightAfterTheRosterRead(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	roster := func(_ int, read request) []byte {
+		if read.code() == uint32(command.GetClusterMetadataCode) {
+			return clusterMetadataFrame(t, 0, firstReplicaAddress, secondReplicaAddress)
+		}
+		return replyFrame(vsr.OperationNonReplicated, nil)
+	}
+	serve(first, roster)
+	serve(second, roster)
+	const calls = 40
+
+	started := time.Now()
+	for range calls {
+		_, err := client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetStatsCode), nil, 9)
+		require.ErrorIs(t, err, ierror.ErrInvalidIdentifier)
+	}
+
+	// A retry pause before the answer averages half of replayInterval per call.
+	assert.Less(t, time.Since(started), calls*replayInterval/5)
+}
+
+func TestSendBinaryRequestToReplica_RefusesReplicatedCommandsPartitionReadsAndSessionControlWithoutWriting(t *testing.T) {
+	client, first, second := newTwoReplicaClient(t)
+	firstServer := serve(first, func(_ int, _ request) []byte { return nil })
+	secondServer := serve(second, func(_ int, _ request) []byte { return nil })
+
+	for _, code := range []command.Code{
+		command.CreateStreamCode,
+		command.SendMessagesCode,
+		command.StoreOffsetCode,
+		command.PollMessagesCode,
+		command.LoginUserCode,
+		command.LoginRegisterCode,
+	} {
+		_, err := client.SendBinaryRequestToReplica(context.Background(), uint32(code), nil, 1)
+		require.ErrorIs(t, err, ierror.ErrInvalidCommand, "code %d", code)
+	}
+	stream, topic, group := groupConsumer(t)
+	partition := uint32(1)
+	offsetRead, err := (&command.GetConsumerOffset{StreamId: stream, TopicId: topic,
+		Consumer: iggcon.NewSingleConsumer(group.Id), PartitionId: &partition}).MarshalBinary()
+	require.NoError(t, err)
+	_, err = client.SendBinaryRequestToReplica(context.Background(), uint32(command.GetOffsetCode), offsetRead, 1)
+	require.ErrorIs(t, err, ierror.ErrInvalidCommand)
+	assert.Empty(t, firstServer.recorded())
+	assert.Empty(t, secondServer.recorded())
+}
+
+// newTwoReplicaClient is a pipe client bound on two replicas, 0 the metadata
+// primary and 1 a backup, at the addresses a roster names them by.
+func newTwoReplicaClient(t *testing.T) (*IggyTcpClient, net.Conn, net.Conn) {
+	t.Helper()
+	client, first := newPipeClient(t)
+	client.links.metadata().endpoint = firstReplicaAddress
+	client.links.slots[0].endpoint = firstReplicaAddress
+	client.links.bootstrap = nil
+	second, connection := net.Pipe()
+	t.Cleanup(func() { _ = second.Close() })
+	link := newTCPLink(connection, secondReplicaAddress)
+	link.markBound(client.sessionFence)
+	client.links.slots[1] = &replicaSlot{endpoint: secondReplicaAddress, link: link, cancel: func() {}}
+	client.links.rosterOrder = []uint8{0, 1}
+	return client, first, second
+}

diff --git a/foreign/go/contracts/client.go b/foreign/go/contracts/client.go
index d7eca07..e142366 100644
--- a/foreign/go/contracts/client.go
+++ b/foreign/go/contracts/client.go

@@ -419,4 +419,15 @@
 	// explicit partition id: a balanced or key-partitioned send returns
 	// ierror.ErrFeatureUnavailable without writing, so use SendMessages for those.
 	SendBinaryRequest(ctx context.Context, code uint32, payload []byte) ([]byte, error)
+
+	// SendBinaryRequestToReplica sends a command code and payload to the replica with id
+	// replicaID, for example to read the stats that replica keeps for itself, and returns
+	// that replica's raw response body. Replica ids are the ones GetClusterMetadata returns.
+	// The request carries the session's metadata watermark, so the replica answers once it
+	// applied the metadata the session saw. Session-control codes, replicated commands,
+	// polls and consumer offset reads return ierror.ErrInvalidCommand without writing to
+	// the connection. An id the roster does not name returns ierror.ErrInvalidIdentifier.
+	// The replica's own refusal, such as ierror.ErrNotPrimary, is returned and never
+	// retried on another replica.
+	SendBinaryRequestToReplica(ctx context.Context, code uint32, payload []byte, replicaID uint8) ([]byte, error)
 }

diff --git a/foreign/go/contracts/version.go b/foreign/go/contracts/version.go
index 703b69f..8022e33 100644
--- a/foreign/go/contracts/version.go
+++ b/foreign/go/contracts/version.go

@@ -17,4 +17,4 @@
 
 package iggcon
 
-const Version = "0.9.1-edge.3"
+const Version = "0.9.1-edge.4"

diff --git a/foreign/java/README.md b/foreign/java/README.md
index 32993dc..0a96937 100644
--- a/foreign/java/README.md
+++ b/foreign/java/README.md

@@ -38,7 +38,7 @@
 
 ### Snapshot Versions
 
-Development builds of the next release carry the version `0.9.3-SNAPSHOT`. Get them from the ASF snapshot repository:
+Development builds of the next release carry the version `0.9.4-SNAPSHOT`. Get them from the ASF snapshot repository:
 
 **Gradle:**
 
@@ -51,7 +51,7 @@
 }
 
 dependencies {
-    implementation 'org.apache.iggy:iggy:0.9.3-SNAPSHOT'
+    implementation 'org.apache.iggy:iggy:0.9.4-SNAPSHOT'
 }
 ```
 
@@ -72,7 +72,7 @@
     <dependency>
         <groupId>org.apache.iggy</groupId>
         <artifactId>iggy</artifactId>
-        <version>0.9.3-SNAPSHOT</version>
+        <version>0.9.4-SNAPSHOT</version>
     </dependency>
 </dependencies>
 ```
@@ -123,6 +123,12 @@
 `sendBinaryRequest` and `resolveOperation`. A rejected change leaves the
 credentials unchanged.
 
+`sendBinaryRequestToReplica(code, payload, replicaId)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with `INVALID_COMMAND` before anything is sent.
+
+An id that cluster metadata does not name fails with `INVALID_IDENTIFIER`. An unreachable node is retried until the request timeout, and a `NOT_PRIMARY` reply is returned rather than sent to another node. The HTTP client throws `IggyOperationNotSupportedException`.
+
 ## Quick Start
 
 `Iggy.tcpClientBuilder()` handles the routing and the session for you. Over TCP and TCP/TLS, the client keeps the consumer group membership on the coordinator. It maintains one shared-session connection per advertised replica and multiplexes partition requests on those connections. With a one-node roster, the client dials the configured endpoint instead of the advertised one. This keeps a single node behind a port map or a port-forward reachable.

diff --git a/foreign/java/gradle.properties b/foreign/java/gradle.properties
index 855b42c..7eefd2a 100644
--- a/foreign/java/gradle.properties
+++ b/foreign/java/gradle.properties

@@ -15,5 +15,5 @@
 # specific language governing permissions and limitations
 # under the License.
 
-version=0.9.3-SNAPSHOT
+version=0.9.4-SNAPSHOT
 group=org.apache.iggy

diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
index 2fb0585..caee709 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClient.java

@@ -36,6 +36,7 @@
 import org.apache.iggy.client.async.UsersClient;
 import org.apache.iggy.client.async.tcp.AsyncTcpConnection.TcpConnectionPoolConfig;
 import org.apache.iggy.client.async.tcp.vsr.VsrFrameDecoder;
+import org.apache.iggy.client.async.tcp.vsr.VsrOperation;
 import org.apache.iggy.cluster.ClusterMetadata;
 import org.apache.iggy.config.RetryPolicy;
 import org.apache.iggy.exception.IggyConnectionException;
@@ -448,18 +449,45 @@
             throw new IggyNotConnectedException();
         }
 
-        return pollRouter.onClientLoop(AsyncTcpConnection.mapResponse(raw.send(code, payload), response -> {
-            try {
-                if (response.readableBytes() <= 1) {
-                    return new byte[0];
-                }
-                byte[] responsePayload = new byte[response.readableBytes()];
-                response.readBytes(responsePayload);
-                return responsePayload;
-            } finally {
-                response.release();
-            }
-        }));
+        return pollRouter.onClientLoop(
+                AsyncTcpConnection.mapResponse(raw.send(code, payload), AsyncIggyTcpClient::responsePayload));
+    }
+
+    /**
+     * Sends a command code and payload to the replica with the given id and returns its raw response payload. The
+     * replica id is the one that cluster metadata names for the node. Every node answers non-replicated commands such
+     * as GetStats from its own state, so this reads the state of the chosen node.
+     *
+     * <p>The request carries the metadata watermark of the session, so the replica answers after it applied the
+     * metadata that this client saw. Replicated commands, PollMessages, GetConsumerOffset and session-control codes
+     * complete the returned future with an invalid-command error before anything is sent. An id that cluster metadata does not
+     * name after one read of it completes the future with an invalid-identifier error. A replica that cannot be
+     * reached, or that answers with code 58, is asked again until the request timeout, which ends the call with code
+     * 58. Any other refusal, NotPrimary (89) included, completes the future unchanged and is never sent to another
+     * replica.
+     *
+     * <p>The future completes on the client event loop, as the callbacks section of this class says.
+     *
+     * @param code the command code
+     * @param payload the command payload
+     * @param replicaId the replica id that cluster metadata names
+     * @return a future containing the raw response payload
+     * @throws IggyNotConnectedException if {@link #connect()} has not been called
+     */
+    public CompletableFuture<byte[]> sendBinaryRequestToReplica(int code, byte[] payload, int replicaId) {
+        Objects.requireNonNull(payload, "payload cannot be null");
+        if (isSessionControlCode(code)
+                || code == CommandCode.Messages.POLL.getValue()
+                || code == CommandCode.ConsumerOffset.GET.getValue()
+                || VsrOperation.operationForCode(code) != VsrOperation.NON_REPLICATED) {
+            return CompletableFuture.failedFuture(
+                    IggyServerException.fromTcpResponse(INVALID_COMMAND_ERROR_CODE, new byte[0]));
+        }
+        if (connection.get() == null || rawRequests == null) {
+            throw new IggyNotConnectedException();
+        }
+        return pollRouter.onClientLoop(AsyncTcpConnection.mapResponse(
+                pollRouter.replicaRequest(replicaId, code, payload), AsyncIggyTcpClient::responsePayload));
     }
 
     /**
@@ -1688,6 +1716,19 @@
         return CompletableFuture.completedFuture(null);
     }
 
+    private static byte[] responsePayload(ByteBuf response) {
+        try {
+            if (response.readableBytes() <= 1) {
+                return new byte[0];
+            }
+            byte[] payload = new byte[response.readableBytes()];
+            response.readBytes(payload);
+            return payload;
+        } finally {
+            response.release();
+        }
+    }
+
     private static boolean isSessionControlCode(int code) {
         return code == CommandCode.User.LOGIN.getValue()
                 || code == CommandCode.User.LOGOUT.getValue()

diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java
index 91ea9fe..a56c06e 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/async/tcp/PollRouter.java

@@ -319,6 +319,18 @@
         return send(poll);
     }
 
+    /**
+     * Sends a non-replicated request to the replica with {@code replicaId} on its link, stamped with the session's
+     * metadata watermark. An id that the roster does not name after a roster read fails with code 6. A lost link or a
+     * refusal with code 58 goes to the same replica again until the deadline, which ends the call with code 58. Any
+     * other reply, code 89 included, is the caller's, because the caller chose the node.
+     */
+    CompletableFuture<ByteBuf> replicaRequest(int replicaId, int code, byte[] payload) {
+        CompletableFuture<ByteBuf> result = new CompletableFuture<>();
+        replicaAttempt(replicaId, code, payload, operationDeadline(), result);
+        return result;
+    }
+
     CompletableFuture<Long> partitionCount(
             StreamId stream, TopicId topic, long deadline, AsyncTcpConnection.TransientFailoverState state) {
         if (state.signedOut()) {
@@ -625,6 +637,71 @@
         return poll.result;
     }
 
+    private void replicaAttempt(
+            int replicaId, int code, byte[] payload, long deadline, CompletableFuture<ByteBuf> result) {
+        replicaLink(replicaId, deadline)
+                .thenCompose(link -> link.sendRouted(
+                        code,
+                        Unpooled.wrappedBuffer(payload),
+                        0,
+                        deadline,
+                        new AsyncTcpConnection.TransientFailoverState()))
+                .whenComplete((response, error) -> {
+                    if (error == null) {
+                        if (!result.complete(response)) {
+                            response.release();
+                        }
+                        return;
+                    }
+                    if (!connectionFailed(error) && !isNotAccepted(error)) {
+                        result.completeExceptionally(unwrap(error));
+                        return;
+                    }
+                    if (result.isDone()
+                            || deadline - System.nanoTime() <= TimeUnit.MILLISECONDS.toNanos(RETRY_INTERVAL_MILLIS)) {
+                        result.completeExceptionally(notAccepted());
+                        return;
+                    }
+                    try {
+                        coordinator
+                                .get()
+                                .eventLoop()
+                                .schedule(
+                                        () -> replicaAttempt(replicaId, code, payload, deadline, result),
+                                        RETRY_INTERVAL_MILLIS,
+                                        TimeUnit.MILLISECONDS);
+                    } catch (RejectedExecutionException loopGone) {
+                        result.completeExceptionally(notAccepted());
+                    }
+                });
+    }
+
+    private CompletableFuture<AsyncTcpConnection> replicaLink(int replicaId, long deadline) {
+        AsyncTcpConnection parent = coordinator.get();
+        if (!parent.isAuthenticated()) {
+            return parent.awaitAuthentication(deadline, new AsyncTcpConnection.TransientFailoverState())
+                    .thenComposeAsync(ignored -> replicaLink(replicaId, deadline), parent.eventLoop());
+        }
+        ConnectionInfo known = replicaEndpoint(replicaId);
+        CompletableFuture<ConnectionInfo> endpoint = known != null
+                ? CompletableFuture.completedFuture(known)
+                : refreshRoster(parent, deadline).thenApply(ignored -> replicaEndpoint(replicaId));
+        return endpoint.thenCompose(target -> {
+            if (target == null) {
+                return CompletableFuture.failedFuture(
+                        IggyServerException.fromTcpResponse(IggyErrorCode.INVALID_IDENTIFIER.getCode(), new byte[0]));
+            }
+            if (coordinatorEndpoint != null && LeaderAwareness.isSameSpelling(target, coordinatorEndpoint.get())) {
+                return CompletableFuture.completedFuture(parent);
+            }
+            return connectionFor(target, deadline);
+        });
+    }
+
+    private synchronized ConnectionInfo replicaEndpoint(int replicaId) {
+        return replicas.get(replicaId);
+    }
+
     CompletableFuture<Void> clearSession() {
         return clearSession(null);
     }

diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java
index dbf5f8f..325a404 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/IggyBaseClient.java

@@ -34,6 +34,22 @@
      */
     byte[] sendBinaryRequest(int code, byte[] payload);
 
+    /**
+     * Sends a command code and payload to the replica with the given id and returns its raw response payload.
+     *
+     * <p>The replica id is the one that cluster metadata names for the node. Every node answers non-replicated
+     * commands such as GetStats from its own state, so this reads the state of the chosen node. Replicated commands,
+     * PollMessages, GetConsumerOffset and session-control codes are rejected with an invalid-command error before
+     * anything is sent. An id that cluster metadata does not name fails with an invalid-identifier error. A NotPrimary refusal is returned and
+     * never sent to another replica. HTTP clients report that this operation is unsupported.
+     *
+     * @param code the command code
+     * @param payload the command payload
+     * @param replicaId the replica id that cluster metadata names
+     * @return the raw response payload
+     */
+    byte[] sendBinaryRequestToReplica(int code, byte[] payload, int replicaId);
+
     SystemClient system();
 
     StreamsClient streams();

diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java
index 7d6226e..2dc0d20 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/http/IggyHttpClient.java

@@ -118,6 +118,12 @@
         throw new IggyOperationNotSupportedException("sendBinaryRequest", "HTTP");
     }
 
+    /** {@inheritDoc} */
+    @Override
+    public byte[] sendBinaryRequestToReplica(int code, byte[] payload, int replicaId) {
+        throw new IggyOperationNotSupportedException("sendBinaryRequestToReplica", "HTTP");
+    }
+
     @Override
     public SystemClient system() {
         return systemClient;

diff --git a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java
index c4a5fbc..b559198 100644
--- a/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java
+++ b/foreign/java/java-sdk/src/main/java/org/apache/iggy/client/blocking/tcp/IggyTcpClient.java

@@ -112,6 +112,12 @@
         return FutureUtil.resolve(asyncClient.sendBinaryRequest(code, payload));
     }
 
+    /** {@inheritDoc} */
+    @Override
+    public byte[] sendBinaryRequestToReplica(int code, byte[] payload, int replicaId) {
+        return FutureUtil.resolve(asyncClient.sendBinaryRequestToReplica(code, payload, replicaId));
+    }
+
     /**
      * Resolves the retained operation under its original identity and returns its response body. A later resolve
      * returns the same outcome without a request, see {@link AsyncIggyTcpClient#resolveOperation(long, Duration)}.

diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientReplicaRequestTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientReplicaRequestTest.java
new file mode 100644
index 0000000..4dc9aeb
--- /dev/null
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/async/tcp/AsyncIggyTcpClientReplicaRequestTest.java

@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.iggy.client.async.tcp;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.Request;
+import org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.Response;
+import org.apache.iggy.exception.IggyErrorCode;
+import org.junit.jupiter.api.Test;
+
+import java.net.InetAddress;
+import java.net.ServerSocket;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.BIND_SESSION_CODE;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.CREATE_STREAM_CODE;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.GET_CLUSTER_METADATA_CODE;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.OPERATION_CREATE_STREAM;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.OPERATION_NON_REPLICATED;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.OPERATION_REGISTER;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.POLL_CODE;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.TRANSIENT_NOT_ACCEPTED;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.assertServerError;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.bindBody;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.client;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.clusterMetadata;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.registerBody;
+import static org.apache.iggy.client.async.tcp.AsyncIggyTcpClientTransientFailoverTest.serve;
+import static org.assertj.core.api.Assertions.assertThat;
+
+class AsyncIggyTcpClientReplicaRequestTest {
+    private static final int GET_STATS_CODE = 10;
+    private static final int LOGIN_CODE = 38;
+    private static final int LOGOUT_CODE = 39;
+    private static final int GET_OFFSET_CODE = 120;
+    private static final int NOT_PRIMARY = 89;
+    private static final long METADATA_COMMIT = 11;
+    private static final int TEST_TIMEOUT_SECONDS = 5;
+    private static final byte[] LEADER_STATS = "leader".getBytes(StandardCharsets.UTF_8);
+    private static final byte[] BACKUP_STATS = "backup".getBytes(StandardCharsets.UTF_8);
+
+    @Test
+    void shouldSendToTheChosenReplicaWithTheSessionWatermark() throws Exception {
+        try (Cluster cluster = new Cluster(1, stats -> Response.success(OPERATION_NON_REPLICATED, backupStats()))) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                client.sendBinaryRequest(CREATE_STREAM_CODE, new byte[0]).get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+                byte[] backup = client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[] {1}, 1)
+                        .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+                byte[] leader = client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[] {0}, 0)
+                        .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+                assertThat(backup).isEqualTo(BACKUP_STATS);
+                assertThat(leader).isEqualTo(LEADER_STATS);
+                assertThat(cluster.backupStats).singleElement().satisfies(request -> {
+                    assertThat(request.body()).containsExactly(1);
+                    assertThat(request.metadataOp()).isEqualTo(METADATA_COMMIT);
+                });
+                assertThat(cluster.leaderStats).singleElement().satisfies(request -> {
+                    assertThat(request.body()).containsExactly(0);
+                    assertThat(request.metadataOp()).isEqualTo(METADATA_COMMIT);
+                });
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    @Test
+    void shouldReturnNotPrimaryOfTheChosenReplicaWithoutSendingElsewhere() throws Exception {
+        try (Cluster cluster = new Cluster(1, stats -> Response.error(OPERATION_NON_REPLICATED, NOT_PRIMARY))) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                assertServerError(client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[0], 1), NOT_PRIMARY);
+
+                assertThat(cluster.backupStats).hasSize(1);
+                assertThat(cluster.leaderStats).isEmpty();
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    @Test
+    void shouldAskTheSameReplicaAgainAfterNotAcceptedAndALostLink() throws Exception {
+        try (Cluster cluster = new Cluster(2, stats -> switch (stats) {
+            case 1 -> Response.error(OPERATION_NON_REPLICATED, TRANSIENT_NOT_ACCEPTED);
+            case 2 -> Response.disconnect();
+            default -> Response.success(OPERATION_NON_REPLICATED, backupStats());
+        })) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                byte[] backup = client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[0], 1)
+                        .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+
+                assertThat(backup).isEqualTo(BACKUP_STATS);
+                assertThat(cluster.backupStats).hasSize(3);
+                assertThat(cluster.leaderStats).isEmpty();
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    @Test
+    void shouldEndWithNotAcceptedWhenTheReplicaRefusesUntilTheDeadline() throws Exception {
+        try (Cluster cluster =
+                new Cluster(1, stats -> Response.error(OPERATION_NON_REPLICATED, TRANSIENT_NOT_ACCEPTED))) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                assertServerError(
+                        client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[0], 1), TRANSIENT_NOT_ACCEPTED);
+
+                assertThat(cluster.backupStats.size()).isGreaterThan(1);
+                assertThat(cluster.leaderStats).isEmpty();
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    @Test
+    void shouldRejectAReplicaThatTheRosterDoesNotNameAfterOneRosterRead() throws Exception {
+        try (Cluster cluster = new Cluster(1, stats -> Response.success(OPERATION_NON_REPLICATED, backupStats()))) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                int rosterReads = cluster.rosterReads.get();
+
+                assertServerError(
+                        client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[0], 2),
+                        IggyErrorCode.INVALID_IDENTIFIER.getCode());
+
+                assertThat(cluster.rosterReads).hasValue(rosterReads + 1);
+                assertThat(cluster.backupStats).isEmpty();
+                assertThat(cluster.leaderStats).isEmpty();
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    @Test
+    void shouldRejectReplicatedPartitionReadAndSessionControlCodesBeforeSending() throws Exception {
+        try (Cluster cluster = new Cluster(1, stats -> Response.success(OPERATION_NON_REPLICATED, backupStats()))) {
+            AsyncIggyTcpClient client = cluster.signIn();
+            try {
+                int requests = cluster.leaderRequests.get();
+
+                for (int code : List.of(CREATE_STREAM_CODE, POLL_CODE, GET_OFFSET_CODE, LOGIN_CODE, LOGOUT_CODE)) {
+                    assertServerError(
+                            client.sendBinaryRequestToReplica(code, new byte[0], 1),
+                            IggyErrorCode.INVALID_COMMAND.getCode());
+                }
+
+                assertThat(cluster.leaderRequests).hasValue(requests);
+                assertThat(cluster.backupStats).isEmpty();
+                assertThat(client.sendBinaryRequestToReplica(GET_STATS_CODE, new byte[0], 1)
+                                .get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS))
+                        .isEqualTo(BACKUP_STATS);
+            } finally {
+                client.close().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            }
+            cluster.awaitServed();
+        }
+    }
+
+    private static ByteBuf backupStats() {
+        return Unpooled.wrappedBuffer(BACKUP_STATS);
+    }
+
+    /**
+     * Replica 0 leads metadata and is the node the client signs in on. Replica 1 is a backup that answers the stats
+     * requests with what the test scripts for each of them, numbered from 1.
+     */
+    private static final class Cluster implements AutoCloseable {
+        final List<Request> leaderStats = new CopyOnWriteArrayList<>();
+        final List<Request> backupStats = new CopyOnWriteArrayList<>();
+        final AtomicInteger rosterReads = new AtomicInteger();
+        final AtomicInteger leaderRequests = new AtomicInteger();
+        private final ServerSocket leader = new ServerSocket(0, 4, InetAddress.getLoopbackAddress());
+        private final ServerSocket backup = new ServerSocket(0, 4, InetAddress.getLoopbackAddress());
+        private final Function<Integer, Response> backupAnswer;
+        private final CompletableFuture<Void> leaderServed;
+        private final CompletableFuture<Void> backupServed;
+
+        Cluster(int backupLinks, Function<Integer, Response> backupAnswer) throws Exception {
+            this.backupAnswer = backupAnswer;
+            leaderServed = serve(leader, this::answerOnLeader);
+            backupServed = serve(backup, backupLinks, this::answerOnBackup);
+        }
+
+        AsyncIggyTcpClient signIn() throws Exception {
+            AsyncIggyTcpClient client = client(leader);
+            client.connect().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            client.login().get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            return client;
+        }
+
+        /** Raises what a node failed on, such as a request it did not expect. */
+        void awaitServed() throws Exception {
+            leaderServed.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+            backupServed.get(TEST_TIMEOUT_SECONDS, TimeUnit.SECONDS);
+        }
+
+        private Response answerOnLeader(Request request) {
+            leaderRequests.incrementAndGet();
+            if (request.operation() == OPERATION_REGISTER) {
+                return Response.success(OPERATION_REGISTER, registerBody(1));
+            }
+            if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) {
+                rosterReads.incrementAndGet();
+                return Response.success(OPERATION_NON_REPLICATED, roster());
+            }
+            if (request.operation() == OPERATION_CREATE_STREAM) {
+                return Response.committed(
+                        OPERATION_CREATE_STREAM,
+                        METADATA_COMMIT,
+                        Unpooled.buffer().writeIntLE(0));
+            }
+            if (request.is(GET_STATS_CODE, OPERATION_NON_REPLICATED)) {
+                leaderStats.add(request);
+                return Response.success(OPERATION_NON_REPLICATED, Unpooled.wrappedBuffer(LEADER_STATS));
+            }
+            throw new IllegalStateException("Unexpected leader request: " + request);
+        }
+
+        private Response answerOnBackup(Request request) {
+            if (request.operation() == OPERATION_REGISTER) {
+                return Response.success(OPERATION_REGISTER, registerBody(2));
+            }
+            if (request.is(BIND_SESSION_CODE, OPERATION_NON_REPLICATED)) {
+                return Response.success(OPERATION_NON_REPLICATED, bindBody(request));
+            }
+            if (request.is(GET_CLUSTER_METADATA_CODE, OPERATION_NON_REPLICATED)) {
+                return Response.success(OPERATION_NON_REPLICATED, roster());
+            }
+            if (request.is(GET_STATS_CODE, OPERATION_NON_REPLICATED)) {
+                backupStats.add(request);
+                return backupAnswer.apply(backupStats.size());
+            }
+            throw new IllegalStateException("Unexpected backup request: " + request);
+        }
+
+        private ByteBuf roster() {
+            return clusterMetadata(leader.getLocalPort(), backup.getLocalPort(), leader.getLocalPort());
+        }
+
+        @Override
+        public void close() throws Exception {
+            leader.close();
+            backup.close();
+        }
+    }
+}

diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java
index 4eb9838..dba2bda 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/http/RawCommandHttpClientTest.java

@@ -33,4 +33,12 @@
         assertThatThrownBy(() -> client.sendBinaryRequest(1, new byte[0]))
                 .isInstanceOf(IggyOperationNotSupportedException.class);
     }
+
+    @Test
+    void shouldRejectRawCommandsToAReplica() {
+        var client = new IggyHttpClient("http://127.0.0.1:3000");
+
+        assertThatThrownBy(() -> client.sendBinaryRequestToReplica(10, new byte[0], 0))
+                .isInstanceOf(IggyOperationNotSupportedException.class);
+    }
 }

diff --git a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java
index af3f921..ffb25ec 100644
--- a/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java
+++ b/foreign/java/java-sdk/src/test/java/org/apache/iggy/client/blocking/tcp/RawCommandTcpClientTest.java

@@ -65,6 +65,29 @@
     }
 
     @Test
+    void shouldReturnRawResponsePayloadsOfTheServerReplica() {
+        assertThat(client.sendBinaryRequestToReplica(10, new byte[0], 0)).isNotEmpty();
+    }
+
+    @Test
+    void shouldRejectAReplicaThatClusterMetadataDoesNotName() {
+        assertThatThrownBy(() -> client.sendBinaryRequestToReplica(10, new byte[0], 9))
+                .isInstanceOf(IggyServerException.class)
+                .extracting(exception -> ((IggyServerException) exception).getErrorCode())
+                .isEqualTo(IggyErrorCode.INVALID_IDENTIFIER);
+    }
+
+    @Test
+    void shouldRejectReplicatedCommandsToAReplica() {
+        for (int code : new int[] {100, 202}) {
+            assertThatThrownBy(() -> client.sendBinaryRequestToReplica(code, new byte[0], 0))
+                    .isInstanceOf(IggyServerException.class)
+                    .extracting(exception -> ((IggyServerException) exception).getErrorCode())
+                    .isEqualTo(IggyErrorCode.INVALID_COMMAND);
+        }
+    }
+
+    @Test
     void shouldPropagateUnknownCommandError() {
         assertThatThrownBy(() -> client.sendBinaryRequest(60_000, new byte[0]))
                 .isInstanceOf(IggyServerException.class)

diff --git a/foreign/node/README.md b/foreign/node/README.md
index e7aa6d4..0516415 100644
--- a/foreign/node/README.md
+++ b/foreign/node/README.md

@@ -59,6 +59,10 @@
 partitions. A raw `SendMessages` command without an explicit partition id
 fails with error code 5 before the client sends it.
 
+`sendBinaryRequestToReplica(code, payload, replica)` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns, and `sendCommand` takes the same id as the `replica` option.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with error code 3 before anything is sent. An id that cluster metadata does not name fails with error code 6. An unreachable node is retried until the request deadline, and a `NotPrimary` (89) reply is returned rather than sent to another node.
+
 Partition commands carry a context containing the partition incarnation, owner
 generation and metadata operation. Pass the context of an earlier poll reply to
 keep an offset write or continuation tied to that history. Otherwise the client

diff --git a/foreign/node/package-lock.json b/foreign/node/package-lock.json
index 032e38c..cde42ac 100644
--- a/foreign/node/package-lock.json
+++ b/foreign/node/package-lock.json

@@ -1,12 +1,12 @@
 {
   "name": "apache-iggy",
-  "version": "0.10.1-edge.3",
+  "version": "0.10.1-edge.4",
   "lockfileVersion": 3,
   "requires": true,
   "packages": {
     "": {
       "name": "apache-iggy",
-      "version": "0.10.1-edge.3",
+      "version": "0.10.1-edge.4",
       "license": "Apache-2.0",
       "dependencies": {
         "@node-rs/xxhash": "1.7.8",

diff --git a/foreign/node/package.json b/foreign/node/package.json
index 97de345..97e12d3 100644
--- a/foreign/node/package.json
+++ b/foreign/node/package.json

@@ -1,7 +1,7 @@
 {
   "name": "apache-iggy",
   "type": "module",
-  "version": "0.10.1-edge.3",
+  "version": "0.10.1-edge.4",
   "description": "Official Apache Iggy NodeJS SDK",
   "keywords": [
     "iggy",

diff --git a/foreign/node/src/bdd/raw.ts b/foreign/node/src/bdd/raw.ts
index f4daceb..816c0db 100644
--- a/foreign/node/src/bdd/raw.ts
+++ b/foreign/node/src/bdd/raw.ts

@@ -35,6 +35,23 @@
   }
 );
 
+When(
+  'I send a raw command with code {int} and an empty payload to replica {int}',
+  async function (this: TestWorld, code: number, replica: number) {
+    try {
+      this.rawResponse = await this.client.sendBinaryRequestToReplica(
+        code,
+        Buffer.alloc(0),
+        replica
+      );
+      this.rawError = undefined;
+    } catch (error) {
+      this.rawResponse = undefined;
+      this.rawError = error instanceof Error ? error : new Error(String(error));
+    }
+  }
+);
+
 Then('the raw command should succeed with an empty response', function (this: TestWorld) {
   assert.equal(this.rawError, undefined);
   assert.deepEqual(this.rawResponse, Buffer.alloc(0));
@@ -49,3 +66,8 @@
   assert.equal(this.rawResponse, undefined);
   assert.match(this.rawError?.message ?? '', /code: 3, message: Invalid command/);
 });
+
+Then('the raw command should fail with an invalid identifier error', function (this: TestWorld) {
+  assert.equal(this.rawResponse, undefined);
+  assert.match(this.rawError?.message ?? '', /code: 6, message: Invalid identifier/);
+});

diff --git a/foreign/node/src/client/client.socket.test.ts b/foreign/node/src/client/client.socket.test.ts
index 37049a0..963a451 100644
--- a/foreign/node/src/client/client.socket.test.ts
+++ b/foreign/node/src/client/client.socket.test.ts

@@ -423,6 +423,109 @@
   return server;
 };
 
+describe('raw commands to a chosen replica', () => {
+  const statsCode = 10;
+  const isStats = (frame: Buffer) =>
+    frame[REQUEST_OFFSET.operation] === Operation.NonReplicated &&
+    frame.readUInt32LE(REQUEST_OFFSET.reserved) === statsCode;
+
+  it('sends the command to the chosen replica only, with the metadata watermark', async () => {
+    const cluster = await startPollCluster((frame, socket) => {
+      if (!isStats(frame))
+        return false;
+      socket.write(replyFrame(frame, Operation.NonReplicated, Buffer.from('replica-7')));
+      return true;
+    });
+    try {
+      await cluster.client.sendCommand(statsCode, Buffer.alloc(0), { deadline: Date.now() + 1_000 });
+      const response = await cluster.client.sendCommand(statsCode, Buffer.alloc(0),
+        { replica: 7, deadline: Date.now() + 1_000 });
+      assert.deepEqual(response.data, Buffer.from('replica-7'));
+      const sent = cluster.primary.frames.filter(isStats);
+      const routed = cluster.coordinator.frames.filter(isStats);
+      assert.equal(sent.length, 1);
+      assert.equal(routed.length, 1, 'only the request without a replica reached the metadata primary');
+      assert.equal(stampedContext(sent[0]).incarnation, 0n);
+      assert.equal(stampedContext(sent[0]).metadataOp, stampedContext(routed[0]).metadataOp);
+    } finally {
+      await cluster.close();
+    }
+  });
+
+  it('returns 89 from the chosen replica without trying another replica', async () => {
+    const cluster = await startPollCluster((frame, socket) => {
+      if (!isStats(frame))
+        return false;
+      socket.write(replyFrame(frame, Operation.NonReplicated, Buffer.alloc(0), 89));
+      return true;
+    });
+    try {
+      await assert.rejects(cluster.client.sendCommand(statsCode, Buffer.alloc(0),
+        { replica: 7, deadline: Date.now() + 1_000 }),
+      (error: unknown) => error instanceof ResponseError && error.errorCode === 89);
+      assert.equal(cluster.primary.frames.filter(isStats).length, 1);
+      assert.equal(cluster.coordinator.frames.filter(isStats).length, 0);
+    } finally {
+      await cluster.close();
+    }
+  });
+
+  it('retries a transient refusal on the same replica', async () => {
+    let refused = false;
+    const cluster = await startPollCluster((frame, socket) => {
+      if (!isStats(frame))
+        return false;
+      socket.write(replyFrame(frame, Operation.NonReplicated, Buffer.from('served'), refused ? 0 : 58));
+      refused = true;
+      return true;
+    });
+    try {
+      const response = await cluster.client.sendCommand(statsCode, Buffer.alloc(0),
+        { replica: 7, deadline: Date.now() + 1_000 });
+      assert.deepEqual(response.data, Buffer.from('served'));
+      assert.equal(cluster.primary.frames.filter(isStats).length, 2);
+      assert.equal(cluster.coordinator.frames.filter(isStats).length, 0);
+    } finally {
+      await cluster.close();
+    }
+  });
+
+  it('fails a replica the roster does not name with status 6 after one roster read', async () => {
+    const cluster = await startPollCluster();
+    try {
+      await cluster.client.sendCommand(COMMAND_CODE.GetStreams, Buffer.alloc(0), { deadline: Date.now() + 1_000 });
+      const rosters = countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata) +
+        countCommand(cluster.primary, COMMAND_CODE.GetClusterMetadata);
+      await assert.rejects(cluster.client.sendCommand(statsCode, Buffer.alloc(0),
+        { replica: 200, deadline: Date.now() + 1_000 }),
+      (error: unknown) => error instanceof ResponseError && error.errorCode === 6);
+      assert.equal(countCommand(cluster.coordinator, COMMAND_CODE.GetClusterMetadata) +
+        countCommand(cluster.primary, COMMAND_CODE.GetClusterMetadata), rosters + 1);
+      assert.equal(cluster.primary.frames.filter(isStats).length + cluster.coordinator.frames.filter(isStats).length, 0);
+    } finally {
+      await cluster.close();
+    }
+  });
+
+  it('refuses replicated commands, partition reads, session control and invalid ids before the wire', async () => {
+    const cluster = await startPollCluster();
+    try {
+      for (const command of [COMMAND_CODE.SendMessages, COMMAND_CODE.StoreOffset, COMMAND_CODE.CreateStream,
+        COMMAND_CODE.PollMessages, COMMAND_CODE.GetOffset, LOGIN.code, LOGOUT.code, COMMAND_CODE.LoginRegister,
+        COMMAND_CODE.LoginRegisterWithAccessToken])
+        await assert.rejects(cluster.client.sendCommand(command, Buffer.alloc(0), { replica: 7 }),
+          (error: unknown) => error instanceof ResponseError && error.errorCode === 3);
+      for (const replica of [-1, 1.5, 256])
+        await assert.rejects(cluster.client.sendCommand(statsCode, Buffer.alloc(0), { replica }),
+          (error: unknown) => error instanceof ResponseError && error.errorCode === 6);
+      assert.equal(cluster.primary.frames.length, 0);
+      assert.equal(cluster.coordinator.frames.length, 0);
+    } finally {
+      await cluster.close();
+    }
+  });
+});
+
 describe('unified partition routing', () => {
   const offset = { streamId: 1, topicId: 2, partitionId: 3,
     consumer: { kind: ConsumerKind.Single, id: 1 } };

diff --git a/foreign/node/src/client/client.socket.ts b/foreign/node/src/client/client.socket.ts
index 7dbf8b1..f9b8947 100644
--- a/foreign/node/src/client/client.socket.ts
+++ b/foreign/node/src/client/client.socket.ts

@@ -71,6 +71,17 @@
 const NOT_PRIMARY = 89;
 const INVALID_COMMAND = 3;
 const FEATURE_UNAVAILABLE = 5;
+const INVALID_IDENTIFIER = 6;
+// Session control, and the partition reads that need the primary and its incarnation.
+const REPLICA_REFUSED_CODES = new Set<number>([
+  COMMAND_CODE.LoginUser,
+  COMMAND_CODE.LogoutUser,
+  COMMAND_CODE.LoginRegister,
+  COMMAND_CODE.LoginWithAccessToken,
+  COMMAND_CODE.LoginRegisterWithAccessToken,
+  COMMAND_CODE.PollMessages,
+  COMMAND_CODE.GetOffset,
+]);
 const INVALID_SESSION_BINDING = 14004;
 const ROUTE_NAMESPACE_MISSING = new Set([1009, 1010, 2010, 2011, 3007]);
 const POLL_OPTIONS_SIZE = 9 + 4 + 1;
@@ -187,6 +198,10 @@
   get busy(): boolean { return this.pendingSubmissions > 0; }
 
   async sendCommand(command: number, payload: Buffer, options: SendCommandOptions = {}): Promise<CommandResponse> {
+    if (options.replica !== undefined && !isReplicaCommand(command))
+      throw responseError(command, INVALID_COMMAND);
+    if (options.replica !== undefined && !(Number.isInteger(options.replica) && options.replica >= 0 && options.replica < MAX_REPLICAS))
+      throw responseError(command, INVALID_IDENTIFIER);
     if (isLoginCommand(command)) return this.loginByCaller(command, () => Buffer.from(payload), options);
     const lineage = options.lineage ?? this.lineage;
     payload = Buffer.from(payload);
@@ -371,12 +386,14 @@
         try {
           if (refresh && target)
             selected = await this.routeFor(target, deadline, attemptSignal, generation, true);
-          else if (refresh)
+          else if (refresh && options.replica === undefined)
             await this.refreshRoster(deadline, attemptSignal);
           refresh = false;
           const link = target
             ? await this.primaryLink(selected!, deadline, attemptSignal)
-            : await this.readyLink(deadline, attemptSignal, safeRead ? previousLink : undefined);
+            : options.replica !== undefined
+              ? await this.replicaLink(command, options.replica, deadline, attemptSignal)
+              : await this.readyLink(deadline, attemptSignal, safeRead ? previousLink : undefined);
           previousLink = link;
           this.ensureGeneration(generation, command);
           attempt = await this.exchange(link, command, () => {
@@ -444,6 +461,9 @@
         } else if (code === OPERATION_NOT_RETAINED) {
           if (operation?.retainable) operation.uncertain = true;
           throw uncertainOutcome(command, operation, error);
+        } else if (code === NOT_PRIMARY && options.replica !== undefined) {
+          // The caller chose the replica, so its refusal is the answer.
+          throw error;
         } else if (code === NOT_PRIMARY || code === TRANSIENT_NOT_ACCEPTED) {
           if (selected) this.invalidateRoute(selected, code === NOT_PRIMARY);
           refresh = true;
@@ -738,6 +758,30 @@
     }
   }
 
+  /** The link of the replica the caller chose, waited for while it reconnects. An id the roster does not name is looked up in one fresh roster, then refused. */
+  private async replicaLink(command: number, replica: number, deadline: number, signal: AbortSignal): Promise<IggyConnection> {
+    let rosterRead = false;
+    for (;;) {
+      const slot = this.replicas.get(replica);
+      if (slot?.attached && slot.connection?.connected) return slot.connection;
+      if (slot?.failure instanceof ResponseError) throw slot.failure;
+      if (slot?.failure instanceof TypeError) throw slot.failure;
+      if (!slot) {
+        if (rosterRead) throw responseError(command, INVALID_IDENTIFIER);
+        rosterRead = true;
+        await this.refreshRoster(deadline, signal);
+        continue;
+      }
+      try { await waitWithin(this.changed.promise, deadline, signal); }
+      catch (error) {
+        if (!(error instanceof VsrResponseTimeoutError)) throw error;
+        const refusal = responseError(command, TRANSIENT_NOT_ACCEPTED);
+        refusal.fromServer = true;
+        throw refusal;
+      }
+    }
+  }
+
   private async readyLink(deadline: number, signal: AbortSignal, avoid?: IggyConnection): Promise<IggyConnection> {
     for (;;) {
       if (avoid) {
@@ -1330,6 +1374,10 @@
   return { query, key: query.toString('hex') };
 };
 
+/** A command the caller may send to a replica it chooses: non-replicated, not a partition read, not session control. */
+const isReplicaCommand = (command: number): boolean =>
+  !REPLICA_REFUSED_CODES.has(command) && operationForCode(command) === Operation.NonReplicated;
+
 const identifierEnd = (payload: Buffer, position: number, command: number): number => {
   if (position + IDENTIFIER_PREFIX_SIZE > payload.length)
     throw responseError(command, INVALID_COMMAND);

diff --git a/foreign/node/src/client/client.type.ts b/foreign/node/src/client/client.type.ts
index 50e76ba..b57d499 100644
--- a/foreign/node/src/client/client.type.ts
+++ b/foreign/node/src/client/client.type.ts

@@ -69,6 +69,16 @@
    */
   context?: PartitionContext,
   /**
+   * Sends the command to the replica with this id instead of the replica the
+   * client picks, for example to read the stats that replica keeps for itself.
+   * Replica ids are the ones the cluster metadata returns. Only non-replicated
+   * commands other than a poll, a consumer-offset read or session control are
+   * accepted, others fail with status 3 before anything is sent. An id the roster does not name fails with status 6. The
+   * replica's reply is returned as it is, a `NotPrimary` (89) refusal
+   * included, and nothing is retried on another replica.
+   */
+  replica?: number,
+  /**
    * The `RawClient.lineage` that a call of several commands started with. A
    * command of an older lineage fails with status 40 and sends nothing, so no
    * part of the call runs under a session the caller opened after it started.

diff --git a/foreign/node/src/e2e/tcp.raw.e2e.ts b/foreign/node/src/e2e/tcp.raw.e2e.ts
index 9ed5bc2..701eb96 100644
--- a/foreign/node/src/e2e/tcp.raw.e2e.ts
+++ b/foreign/node/src/e2e/tcp.raw.e2e.ts

@@ -42,6 +42,29 @@
     assert.ok(response.length > 0);
   });
 
+  it('e2e -> raw::getStatsFromTheServerReplica', async () => {
+    const response = await c.sendBinaryRequestToReplica(
+      COMMAND_CODE.GetStats,
+      Buffer.alloc(0),
+      0
+    );
+    assert.ok(response.length > 0);
+  });
+
+  it('e2e -> raw::unknownReplicaRejected', async () => {
+    await assert.rejects(
+      () => c.sendBinaryRequestToReplica(COMMAND_CODE.GetStats, Buffer.alloc(0), 9),
+      /code: 6, message: Invalid identifier/
+    );
+  });
+
+  it('e2e -> raw::replicatedCommandToAReplicaRejectedClientSide', async () => {
+    await assert.rejects(
+      () => c.sendBinaryRequestToReplica(COMMAND_CODE.CreateStream, Buffer.alloc(0), 0),
+      /code: 3/
+    );
+  });
+
   it('e2e -> raw::sessionControlCodeRejectedClientSide', async () => {
     await assert.rejects(
       () => c.sendBinaryRequest(COMMAND_CODE.LoginUser, Buffer.alloc(0))

diff --git a/foreign/node/src/wire/command-set.test.ts b/foreign/node/src/wire/command-set.test.ts
index 56069ef..786d4a3 100644
--- a/foreign/node/src/wire/command-set.test.ts
+++ b/foreign/node/src/wire/command-set.test.ts

@@ -107,3 +107,32 @@
     assert.deepEqual(response, Buffer.alloc(0));
   });
 });
+
+describe('CommandAPI.sendBinaryRequestToReplica', () => {
+  it('rejects a session-control code before reaching the client provider', async () => {
+    const client = new SimpleClient(mockRawClient());
+    await assert.rejects(
+      () => client.sendBinaryRequestToReplica(COMMAND_CODE.LoginUser, Buffer.alloc(0), 1),
+      /code: 3, message: Invalid command/
+    );
+  });
+
+  it('forwards the code, payload and replica to sendCommand', async () => {
+    const payload = Buffer.from([0xAA, 0xBB]);
+    const expectedResponse = Buffer.from('replica stats');
+    const raw = mockRawClient();
+    raw.sendCommand = async (code, sentPayload, options) => {
+      assert.equal(code, COMMAND_CODE.GetStats);
+      assert.deepEqual(sentPayload, payload);
+      assert.deepEqual(options, { replica: 2 });
+      return {
+        status: 0,
+        length: expectedResponse.length,
+        data: expectedResponse,
+      };
+    };
+    const client = new SimpleClient(raw);
+    const response = await client.sendBinaryRequestToReplica(COMMAND_CODE.GetStats, payload, 2);
+    assert.deepEqual(response, expectedResponse);
+  });
+});

diff --git a/foreign/node/src/wire/command-set.ts b/foreign/node/src/wire/command-set.ts
index ca6ddad..e77b20a 100644
--- a/foreign/node/src/wire/command-set.ts
+++ b/foreign/node/src/wire/command-set.ts

@@ -264,13 +264,44 @@
     code: number,
     payload: Buffer
   ): Promise<Buffer> {
+    return this.sendRawCommand(code, payload);
+  }
+
+  /**
+   * Sends a command code with a payload to the replica with the given id, as
+   * the cluster metadata names it, and returns the raw response payload. Every
+   * node answers non-replicated commands such as GetStats from its own state.
+   * Replicated commands, polls, consumer-offset reads and session-control codes
+   * are rejected with an invalid-command error, and a replica id the cluster
+   * metadata does not name with an invalid-identifier error. A `NotPrimary`
+   * refusal is returned and not retried on another replica.
+   *
+   * @param code - Command code to send
+   * @param payload - Raw command payload
+   * @param replica - Replica id, as the cluster metadata names it
+   * @returns Raw response payload
+   */
+  async sendBinaryRequestToReplica(
+    code: number,
+    payload: Buffer,
+    replica: number
+  ): Promise<Buffer> {
+    return this.sendRawCommand(code, payload, { replica });
+  }
+
+  private async sendRawCommand(
+    code: number,
+    payload: Buffer,
+    options?: SendCommandOptions
+  ): Promise<Buffer> {
     if (SESSION_CONTROL_CODES.has(code))
       throw responseError(code, INVALID_COMMAND_ERROR_CODE);
 
     const snapshot = Buffer.from(payload);
     const response = await (await this.clientProvider()).sendCommand(
       code,
-      snapshot
+      snapshot,
+      options
     );
     return response.length <= 1 ? Buffer.alloc(0) : response.data;
   }

diff --git a/foreign/php/Cargo.toml b/foreign/php/Cargo.toml
index 9c7933c..0355998 100644
--- a/foreign/php/Cargo.toml
+++ b/foreign/php/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "iggy-php"
-version = "0.1.5"
+version = "0.1.6"
 edition = "2024"
 authors = ["Iggy Committers <dev@iggy.apache.org>"]
 license = "Apache-2.0"

diff --git a/foreign/php/README.md b/foreign/php/README.md
index 51cc55b..9b2cb53 100644
--- a/foreign/php/README.md
+++ b/foreign/php/README.md

@@ -389,6 +389,7 @@
   getters return PHP integers and are limited to `PHP_INT_MAX`.
 - `Iggy\Client::sendBinaryRequest(int $code, string $payload): string` sends a
   command code and payload and returns the raw response body.
+- `Iggy\Client::sendBinaryRequestToReplica(int $code, string $payload, int $replica_id): string` sends a raw command to one node of a cluster, named by the replica id that cluster metadata returns, for example to read that node's `GetStats`. Replicated commands, polls, consumer-offset reads and session-control codes throw an invalid-command exception before anything is sent. An unknown id throws an invalid-identifier exception, and a `NotPrimary` reply is thrown, never sent to another node.
 - `Iggy\Client` is synchronous and blocks the current PHP thread.
 - The extension owns a lazy global Tokio runtime. Do not call `pcntl_fork()` after
   the first Iggy SDK call; the child process inherits file descriptors but not

diff --git a/foreign/php/iggy-php.stubs.php b/foreign/php/iggy-php.stubs.php
index b7454f8..3a3f551 100644
--- a/foreign/php/iggy-php.stubs.php
+++ b/foreign/php/iggy-php.stubs.php

@@ -314,6 +314,20 @@
         public function sendBinaryRequest(int $code, string $payload): string {}
 
         /**
+         * Sends a command code with a payload to the replica with the given id, as
+         * cluster metadata names it, and returns the raw response bytes.
+         * Replicated commands, polls, consumer-offset reads and session-control
+         * codes return an invalid-command exception, and a replica that cluster
+         * metadata does not name returns an invalid-identifier exception.
+         *
+         * @param int $code
+         * @param string $payload
+         * @param int $replica_id
+         * @return string
+         */
+        public function sendBinaryRequestToReplica(int $code, string $payload, int $replica_id): string {}
+
+        /**
          * Sends messages to a topic and returns the commit confirmations, one per
          * partition the batch landed in.
          *

diff --git a/foreign/php/src/client.rs b/foreign/php/src/client.rs
index b5b70b1..3dbf5ee 100644
--- a/foreign/php/src/client.rs
+++ b/foreign/php/src/client.rs

@@ -712,6 +712,32 @@
                 .map_err(to_php_exception)
         })
     }
+
+    /// Sends a command code with a payload to the replica with the given id, as
+    /// cluster metadata names it, and returns the raw response bytes.
+    /// Replicated commands, polls, consumer-offset reads and session-control
+    /// codes return an invalid-command exception, and a replica that cluster
+    /// metadata does not name returns an invalid-identifier exception.
+    pub fn send_binary_request_to_replica(
+        &self,
+        code: u32,
+        payload: Binary<u8>,
+        replica_id: u8,
+    ) -> PhpResult<Binary<u8>> {
+        let inner = self.inner.clone();
+
+        runtime().block_on(async move {
+            inner
+                .send_binary_request_to_replica(
+                    code,
+                    Bytes::from(Vec::<u8>::from(payload)),
+                    replica_id,
+                )
+                .await
+                .map(|response| Binary::new(Vec::from(response)))
+                .map_err(to_php_exception)
+        })
+    }
 }
 
 pub(crate) fn non_zero_duration_micros(field: &str, micros: u64) -> PhpResult<NonZeroIggyDuration> {

diff --git a/foreign/php/src/sensitive_parameters.rs b/foreign/php/src/sensitive_parameters.rs
index 12735e9..73aeea1 100644
--- a/foreign/php/src/sensitive_parameters.rs
+++ b/foreign/php/src/sensitive_parameters.rs

@@ -53,6 +53,7 @@
     ("Iggy\\Client", "fromConnectionString", 0),
     ("Iggy\\Client", "loginUser", 1),
     ("Iggy\\Client", "sendBinaryRequest", 1),
+    ("Iggy\\Client", "sendBinaryRequestToReplica", 1),
     ("Iggy\\Client", "createStream", 1),
     ("Iggy\\Client", "createStream", 2),
     ("Iggy\\Client", "updateStream", 3),

diff --git a/foreign/php/tests/RawCommandTest.php b/foreign/php/tests/RawCommandTest.php
index e30945c..b0626f0 100644
--- a/foreign/php/tests/RawCommandTest.php
+++ b/foreign/php/tests/RawCommandTest.php

@@ -31,6 +31,9 @@
     private const PING_CODE = 1;
     private const GET_STATS_CODE = 10;
     private const UNKNOWN_CODE = 60_000;
+    private const CREATE_STREAM_CODE = 202;
+    private const SERVER_REPLICA_ID = 0;
+    private const UNKNOWN_REPLICA_ID = 9;
 
     #[TestDox('A raw ping command returns an empty successful response')]
     public function testRawPingReturnsEmptyResponse(): void
@@ -73,6 +76,42 @@
         assert_instance_of(IggyException::class, $throwable);
     }
 
+    #[TestDox('A raw get-stats command sent to the server replica returns a non-empty response')]
+    public function testRawGetStatsToTheServerReplicaReturnsNonEmptyResponse(): void
+    {
+        $client = new_client();
+
+        $response = $client->sendBinaryRequestToReplica(self::GET_STATS_CODE, '', self::SERVER_REPLICA_ID);
+
+        assert_true($response !== '', 'expected a non-empty stats response');
+    }
+
+    #[TestDox('A replica that cluster metadata does not name is rejected')]
+    public function testRawCommandToAnUnknownReplicaIsRejected(): void
+    {
+        $client = new_client();
+
+        $throwable = assert_throws(
+            static fn () => $client->sendBinaryRequestToReplica(self::GET_STATS_CODE, '', self::UNKNOWN_REPLICA_ID)
+        );
+
+        assert_instance_of(IggyException::class, $throwable);
+        assert_same(6, $throwable->getCode());
+    }
+
+    #[TestDox('A replicated command sent to a replica is rejected before reaching the server')]
+    public function testRawReplicatedCommandToAReplicaIsRejected(): void
+    {
+        $client = new_client();
+
+        $throwable = assert_throws(
+            static fn () => $client->sendBinaryRequestToReplica(self::CREATE_STREAM_CODE, '', self::SERVER_REPLICA_ID)
+        );
+
+        assert_instance_of(IggyException::class, $throwable);
+        assert_same(3, $throwable->getCode());
+    }
+
     public static function sessionControlCodes(): array
     {
         return [[38], [39], [40], [44], [45]];

diff --git a/foreign/php/tests/SensitiveParameterTest.php b/foreign/php/tests/SensitiveParameterTest.php
index 42142b2..96073e7 100644
--- a/foreign/php/tests/SensitiveParameterTest.php
+++ b/foreign/php/tests/SensitiveParameterTest.php

@@ -34,6 +34,7 @@
             'connection string' => ['fromConnectionString', 'connection_string'],
             'password' => ['loginUser', 'password'],
             'raw binary payload' => ['sendBinaryRequest', 'payload'],
+            'raw binary payload to a replica' => ['sendBinaryRequestToReplica', 'payload'],
             'stream key' => ['createStream', 'encryption_key'],
             'fingerprinted key' => ['encryptionKeyFingerprint', 'key'],
             'raw stream options' => ['createStream', 'raw'],

diff --git a/foreign/python/Cargo.toml b/foreign/python/Cargo.toml
index 93a3cd1..2ff1f96 100644
--- a/foreign/python/Cargo.toml
+++ b/foreign/python/Cargo.toml

@@ -17,7 +17,7 @@
 
 [package]
 name = "apache-iggy"
-version = "0.9.1-dev3"
+version = "0.9.1-dev4"
 edition = "2024"
 authors = ["Iggy Committers <dev@iggy.apache.org>"]
 license = "Apache-2.0"
@@ -37,7 +37,7 @@
 [dependencies]
 bytes = "1.12.1"
 futures = "0.3.34"
-iggy = { path = "../../core/sdk", version = "0.11.1-edge.3" }
+iggy = { path = "../../core/sdk", version = "0.11.1-edge.4" }
 paste = "1"
 pyo3 = "0.29.3"
 pyo3-async-runtimes = { version = "0.29.0", features = [

diff --git a/foreign/python/README.md b/foreign/python/README.md
index abeb9a4..a7ac8f6 100644
--- a/foreign/python/README.md
+++ b/foreign/python/README.md

@@ -343,6 +343,12 @@
 
 `str()` of a cursor gives its `mc1_` text and of a page token its `cs1_` text, and `MetadataChangeCursor.parse` and `MetadataCurrentPageToken.parse` read them back, so a mirror can store its position as a string. Log ids, checksums and client ids are Python ints.
 
+## Raw Commands
+
+`send_binary_request(code, payload)` sends a command code with a payload and returns the raw response bytes. `send_binary_request_to_replica(code, payload, replica_id)` sends it to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes raise an invalid-command error before anything is sent. An id that cluster metadata does not name raises an invalid-identifier error. An unreachable node is retried until the request timeout, and a `NotPrimary` reply is raised rather than sent to another node. HTTP does not support raw commands.
+
 ## High-Level Producer
 
 The Python high-level producer API is a port of the Rust high-level producer

diff --git a/foreign/python/apache_iggy.pyi b/foreign/python/apache_iggy.pyi
index ef519c9..cfaef53 100644
--- a/foreign/python/apache_iggy.pyi
+++ b/foreign/python/apache_iggy.pyi

@@ -2141,6 +2141,31 @@
         Raises:
             RuntimeError: If the command cannot be sent or the server returns an error.
         """
+    def send_binary_request_to_replica(
+        self, code: builtins.int, payload: builtins.bytes, replica_id: builtins.int
+    ) -> collections.abc.Awaitable[bytes]:
+        r"""
+        Send a command code with a payload to the replica with the given id and
+        return the raw response bytes.
+
+        The replica id is the one cluster metadata names for the node. Every
+        node answers non-replicated commands such as GetStats from its own
+        state. Replicated commands, polls, consumer-offset reads and
+        session-control codes are rejected client-side, and a replica id that
+        cluster metadata does not name fails. HTTP transport does not support
+        raw binary commands.
+
+        Args:
+            code: Command code as `int`.
+            payload: Request payload as `bytes`.
+            replica_id: Replica id as `int`, as cluster metadata names it.
+
+        Returns:
+            An awaitable that resolves to the raw response `bytes`.
+
+        Raises:
+            RuntimeError: If the command cannot be sent or the replica returns an error.
+        """
 
 @typing.final
 class IggyConsumer:

diff --git a/foreign/python/pyproject.toml b/foreign/python/pyproject.toml
index 5914cdc..3b87649 100644
--- a/foreign/python/pyproject.toml
+++ b/foreign/python/pyproject.toml

@@ -22,7 +22,7 @@
 [project]
 name = "apache-iggy"
 requires-python = ">=3.11"
-version = "0.9.1.dev3"
+version = "0.9.1.dev4"
 description = "Apache Iggy is the persistent message streaming platform written in Rust, supporting QUIC, TCP and HTTP transport protocols, capable of processing millions of messages per second."
 readme = "README.md"
 license = { file = "LICENSE" }

diff --git a/foreign/python/src/client.rs b/foreign/python/src/client.rs
index 8faa778..34a0dfb 100644
--- a/foreign/python/src/client.rs
+++ b/foreign/python/src/client.rs

@@ -1937,6 +1937,44 @@
             Ok(Python::attach(|py| PyBytes::new(py, &response).unbind()))
         })
     }
+
+    /// Send a command code with a payload to the replica with the given id and
+    /// return the raw response bytes.
+    ///
+    /// The replica id is the one cluster metadata names for the node. Every
+    /// node answers non-replicated commands such as GetStats from its own
+    /// state. Replicated commands, polls, consumer-offset reads and
+    /// session-control codes are rejected client-side, and a replica id that
+    /// cluster metadata does not name fails. HTTP transport does not support
+    /// raw binary commands.
+    ///
+    /// Args:
+    ///     code: Command code as `int`.
+    ///     payload: Request payload as `bytes`.
+    ///     replica_id: Replica id as `int`, as cluster metadata names it.
+    ///
+    /// Returns:
+    ///     An awaitable that resolves to the raw response `bytes`.
+    ///
+    /// Raises:
+    ///     RuntimeError: If the command cannot be sent or the replica returns an error.
+    #[gen_stub(override_return_type(type_repr="collections.abc.Awaitable[bytes]", imports=("collections.abc")))]
+    fn send_binary_request_to_replica<'a>(
+        &self,
+        py: Python<'a>,
+        code: u32,
+        #[gen_stub(override_type(type_repr = "builtins.bytes"))] payload: Vec<u8>,
+        replica_id: u8,
+    ) -> PyResult<Bound<'a, PyAny>> {
+        let inner = self.inner.clone();
+        future_into_py(py, async move {
+            let response = inner
+                .send_binary_request_to_replica(code, Bytes::from(payload), replica_id)
+                .await
+                .map_err(to_runtime_error)?;
+            Ok(Python::attach(|py| PyBytes::new(py, &response).unbind()))
+        })
+    }
 }
 
 define_stub_info_gatherer!(stub_info);

diff --git a/foreign/python/tests/test_raw_command.py b/foreign/python/tests/test_raw_command.py
index b2841df..f1bce67 100644
--- a/foreign/python/tests/test_raw_command.py
+++ b/foreign/python/tests/test_raw_command.py

@@ -49,6 +49,30 @@
         await iggy_client.send_binary_request(60_000, b"")
 
 
+@pytest.mark.asyncio
+async def test_raw_get_stats_to_the_server_replica_returns_non_empty_response(
+    iggy_client: IggyClient,
+):
+    response = await iggy_client.send_binary_request_to_replica(10, b"", 0)
+
+    assert response
+
+
+@pytest.mark.asyncio
+async def test_raw_command_to_an_unknown_replica_is_rejected(iggy_client: IggyClient):
+    with pytest.raises(RuntimeError, match="(?i)invalid identifier"):
+        await iggy_client.send_binary_request_to_replica(10, b"", 9)
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("code", [100, 202])
+async def test_raw_replicated_command_or_poll_to_a_replica_is_rejected(
+    iggy_client: IggyClient, code: int
+):
+    with pytest.raises(RuntimeError, match="(?i)invalid command"):
+        await iggy_client.send_binary_request_to_replica(code, b"", 0)
+
+
 @pytest.mark.unit
 @pytest.mark.asyncio
 @pytest.mark.parametrize("operation_id", [2**53 + 17, 2**64 - 1])

diff --git a/foreign/python/uv.lock b/foreign/python/uv.lock
index cd4c439..1f94aae 100644
--- a/foreign/python/uv.lock
+++ b/foreign/python/uv.lock

@@ -11,7 +11,7 @@
 
 [[package]]
 name = "apache-iggy"
-version = "0.9.1.dev3"
+version = "0.9.1.dev4"
 source = { editable = "." }
 
 [package.optional-dependencies]

diff --git a/foreign/swift/README.md b/foreign/swift/README.md
index c8717df..5c94ef6 100644
--- a/foreign/swift/README.md
+++ b/foreign/swift/README.md

@@ -322,6 +322,12 @@
 failed request does not hold back the ones after it, so a resend from `errorHandler`
 lands behind them.
 
+### Raw requests
+
+`sendRawRequest(code:payload:)` sends a command code with an encoded payload and returns the raw reply body. `sendRawRequest(code:payload:replica:)` sends it to one node of a cluster, named by the replica id that cluster metadata returns.
+
+Each node answers non-replicated commands such as `GetStats` from its own state, so this reads the state of that node. Replicated commands, polls, consumer-offset reads and session-control codes fail with `invalidCommand` before anything is sent. An id that cluster metadata does not name fails with `invalidIdentifier`. An unreachable node is retried until the request timeout, and `notPrimary` is returned rather than sent to another node.
+
 ### Errors
 
 Every client operation throws `IggyError`, a value carrying the typed `code`

diff --git a/foreign/swift/Sources/Iggy/Client/IggyClient.swift b/foreign/swift/Sources/Iggy/Client/IggyClient.swift
index 357102a..a5bae7c 100644
--- a/foreign/swift/Sources/Iggy/Client/IggyClient.swift
+++ b/foreign/swift/Sources/Iggy/Client/IggyClient.swift

@@ -211,6 +211,34 @@
         return try await core.exchange(code: code, payload: payload, lineage: lineage)
     }
 
+    /// Sends a command code with an already-encoded payload to the replica
+    /// with id `replica`, as the cluster roster names it, and returns its raw
+    /// reply body. Every node answers non-replicated commands such as
+    /// `GetStats` from its own state, so this reads the state of the chosen
+    /// node. The request carries the session's metadata watermark, so the
+    /// replica answers once it applied the metadata this client saw.
+    ///
+    /// Replicated commands, `PollMessages`, `GetConsumerOffset`, and
+    /// session-control codes fail with ``IggyErrorCode/invalidCommand``
+    /// before anything is sent. A
+    /// replica the roster does not list after one more roster read fails with
+    /// ``IggyErrorCode/invalidIdentifier``. A replica that cannot be reached,
+    /// or that answers ``IggyErrorCode/transientNotAccepted``, is asked again
+    /// until the request timeout. Any other refusal,
+    /// ``IggyErrorCode/notPrimary`` included, is returned and never sent to
+    /// another replica.
+    public func sendRawRequest(code: UInt32, payload: [UInt8], replica: UInt8) async throws -> [UInt8] {
+        guard !CommandCode.isSessionControl(code), code != CommandCode.pollMessages.rawValue,
+            code != CommandCode.getConsumerOffset.rawValue, ConsensusOperation.forCode(code) == .nonReplicated
+        else {
+            throw IggyError(.invalidCommand, context: "command \(code) cannot be sent to a chosen replica")
+        }
+        let lineage = try await requireAuthenticated()
+        return try await core.pollRouter.sendToReplica(
+            replica, code: code, payload: payload, lineage: lineage, coordinator: core,
+            deadline: ContinuousClock.now + core.configuration.requestTimeout)
+    }
+
     // MARK: Internals shared by the domain extensions
 
     /// One request of a call that started under `lineage`, or, for a call

diff --git a/foreign/swift/Sources/Iggy/Client/PollRouter.swift b/foreign/swift/Sources/Iggy/Client/PollRouter.swift
index a1aeaba..600cdf5 100644
--- a/foreign/swift/Sources/Iggy/Client/PollRouter.swift
+++ b/foreign/swift/Sources/Iggy/Client/PollRouter.swift

@@ -220,6 +220,56 @@
         }
     }
 
+    /// Sends a non-replicated request to `replica` and returns its answer.
+    /// The request carries the session's metadata watermark. On a single node
+    /// it goes over the coordinator, in a cluster over a data connection to
+    /// the replica, bound to the client's session. A replica the roster does
+    /// not list after one more roster read fails with `invalidIdentifier`. A
+    /// lost connection or a `transientNotAccepted` refusal sends the request
+    /// to the same replica again until `deadline`, which ends the call with
+    /// the last refusal. Any other answer, `notPrimary` included, is the
+    /// caller's, as the caller chose the replica.
+    func sendToReplica(
+        _ replica: UInt8, code: UInt32, payload: [UInt8], lineage: UInt64, coordinator: ClientCore, deadline: ContinuousClock.Instant
+    ) async throws -> [UInt8] {
+        _ = try await isClustered(coordinator, lineage: lineage, deadline: deadline)
+        if state.withLockedValue({ $0.nodes[replica] }) == nil {
+            try await readRoster(coordinator, lineage: lineage, deadline: deadline)
+        }
+        let (listed, clustered) = state.withLockedValue { ($0.nodes[replica] != nil, $0.nodes.count > 1) }
+        guard listed else {
+            throw IggyError(.invalidIdentifier, context: "the cluster roster does not list replica \(replica)")
+        }
+        guard clustered else {
+            return try await coordinator.exchange(code: code, payload: payload, deadline: deadline, lineage: lineage)
+        }
+        let clock = ContinuousClock()
+        var backoff = Backoff()
+        var refusal = IggyError(.transientNotAccepted, context: "the budget ran out before command \(code) reached replica \(replica)")
+        while clock.now < deadline {
+            let watermark = await coordinator.metadataWatermark
+            guard let identity = await coordinator.sessionIdentity(at: watermark) else {
+                throw IggyError(.unauthenticated, context: "no session is bound to send command \(code) to replica \(replica)")
+            }
+            let fence = state.withLockedValue { $0.fence }
+            do {
+                return try await exchange(
+                    onReplica: replica, code: code, payload: payload, context: ClientCore.readContext(watermark), identity: identity,
+                    lineage: lineage, fence: fence, coordinator: coordinator, deadline: deadline,
+                    dialDeadline: min(deadline, .now + ClientCore.failoverDialTimeout))
+            } catch let unsent as UnsentRequest {
+                guard unsent.refusal.code == .transientNotAccepted else {
+                    throw unsent.refusal
+                }
+                refusal = unsent.refusal
+            } catch let error as IggyError where error.code == .transientNotAccepted || error.code == .transientNotCommitted {
+                refusal = error
+            }
+            try await backoff.pause(after: refusal, before: deadline)
+        }
+        throw refusal
+    }
+
     /// The pauses between the refused attempts of one request: from
     /// `retryInterval`, doubling up to `maxRetryInterval`.
     struct Backoff {

diff --git a/foreign/swift/Sources/Iggy/Iggy.swift b/foreign/swift/Sources/Iggy/Iggy.swift
index efd248a..9e8307f 100644
--- a/foreign/swift/Sources/Iggy/Iggy.swift
+++ b/foreign/swift/Sources/Iggy/Iggy.swift

@@ -21,5 +21,5 @@
     public static let sdkName = "swift-sdk"
     /// Version of this SDK as reported to the server. Provisional until the
     /// release wiring lands; the release tooling reads this constant.
-    public static let sdkVersion = "0.1.2"
+    public static let sdkVersion = "0.1.3"
 }

diff --git a/foreign/swift/Tests/IggyTests/Fixtures/golden.json b/foreign/swift/Tests/IggyTests/Fixtures/golden.json
index 5303079..6ae26e6 100644
--- a/foreign/swift/Tests/IggyTests/Fixtures/golden.json
+++ b/foreign/swift/Tests/IggyTests/Fixtures/golden.json

@@ -927,7 +927,7 @@
     "polling.next": "050000000000000000",
     "polling.offset.100": "016400000000000000",
     "polling.timestamp.1700000000000": "020068e5cf8b010000",
-    "request.bind_session": "003000000973776966742d73646b05302e312e3201000000000000000000000000000000640000000000000078000000000000005a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a",
+    "request.bind_session": "003000000973776966742d73646b05302e312e3301000000000000000000000000000000640000000000000078000000000000005a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a",
     "request.change_password": "010401000000086f6c642d706173730c6e65772d706173732d313233",
     "request.create_consumer_group": "01040100000001040200000003677270",
     "request.create_partitions": "01040100000001040200000005000000",
@@ -967,8 +967,8 @@
     "request.get_users": "",
     "request.join_consumer_group": "010401000000010402000000010403000000",
     "request.leave_consumer_group": "010401000000010402000000020167",
-    "request.login_register": "003000000973776966742d73646b05302e312e325a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a0469676779046967677900000000",
-    "request.login_register_with_pat": "003000000973776966742d73646b05302e312e325a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a107061742d61626331323364656634353600000000",
+    "request.login_register": "003000000973776966742d73646b05302e312e335a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a0469676779046967677900000000",
+    "request.login_register_with_pat": "003000000973776966742d73646b05302e312e335a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a5a107061742d61626331323364656634353600000000",
     "request.logout_user": "",
     "request.ping": "",
     "request.poll_messages": "0101040100000001040a00000001041400000001050000000164000000000000003200000001",
@@ -1019,7 +1019,7 @@
     "response.user_details.no_permissions": "0100000000e090663c130600010561646d696e000000000000000000000000",
     "stream_key.fingerprint": "8441e38538cbfa57",
     "user_headers.sample": "0207000000617474656d70740b0400000003000000020800000074726163652d69640203000000616263",
-    "version_info": "003000000973776966742d73646b05302e312e32",
+    "version_info": "003000000973776966742d73646b05302e312e33",
     "vsr.eviction.incompatible_protocol": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523010030000000300000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000e",
     "vsr.eviction.stale_client": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001000000000000000000000d000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab89674523010000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000d",
     "vsr.reply.committed": "0000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000c0100000000000000000000080000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000000001032547698badcfeefcdab896745230100000000000000000c0000000000000000000000000000000700000000000000800000000000000000000000010000000000000000000000000000000000000000000000000000000000000000000000",

diff --git a/foreign/swift/Tests/IggyTests/ReplicaRequestTests.swift b/foreign/swift/Tests/IggyTests/ReplicaRequestTests.swift
new file mode 100644
index 0000000..f9d1bdb
--- /dev/null
+++ b/foreign/swift/Tests/IggyTests/ReplicaRequestTests.swift

@@ -0,0 +1,179 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import Testing
+
+@testable import Iggy
+
+private let leaderReply: [UInt8] = Array("leader".utf8)
+private let backupReply: [UInt8] = Array("backup".utf8)
+private let metadataCommit: UInt64 = 7
+
+/// Raw requests that name the replica to answer them, on a fake leader
+/// (replica 0, the coordinator) and a fake backup (replica 1). Mirrors the
+/// replica request tests of `core/sdk/src/driver.rs`.
+@Suite("Replica requests", .serialized)
+struct ReplicaRequestTests {
+    @Test func aRawRequestReachesTheChosenReplicaWithTheSessionWatermark() async throws {
+        let cluster = try await TwoReplicas.started { _ in .ok(backupReply) }
+        defer { cluster.stop() }
+        let client = try await cluster.signedInClient()
+        _ = try await client.sendRawRequest(code: CommandCode.createStream.rawValue, payload: [])
+
+        let backup = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [1], replica: 1)
+        let leader = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [0], replica: 0)
+
+        #expect(backup == backupReply)
+        #expect(leader == leaderReply)
+        let backupStats = await cluster.stats(on: cluster.backup)
+        let leaderStats = await cluster.stats(on: cluster.leader)
+        #expect(backupStats.map(\.payload) == [[1]])
+        #expect(backupStats.map(\.context.metadataOp) == [metadataCommit])
+        #expect(leaderStats.map(\.payload) == [[0]])
+        #expect(leaderStats.map(\.context.metadataOp) == [metadataCommit])
+        await client.shutdown()
+    }
+
+    @Test func notPrimaryFromTheChosenReplicaIsReturnedWithoutTryingAnother() async throws {
+        let cluster = try await TwoReplicas.started { _ in .status(IggyErrorCode.notPrimary.rawValue) }
+        defer { cluster.stop() }
+        let client = try await cluster.signedInClient()
+
+        await expectCode(.notPrimary) { _ = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [], replica: 1) }
+
+        #expect(await cluster.stats(on: cluster.backup).count == 1)
+        #expect(await cluster.stats(on: cluster.leader).isEmpty)
+        await client.shutdown()
+    }
+
+    @Test func aRefusalAndALostConnectionAskTheSameReplicaAgain() async throws {
+        let cluster = try await TwoReplicas.started { attempt in
+            switch attempt {
+            case 1: .status(IggyErrorCode.transientNotAccepted.rawValue)
+            case 2: .close
+            default: .ok(backupReply)
+            }
+        }
+        defer { cluster.stop() }
+        let client = try await cluster.signedInClient()
+
+        let backup = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [], replica: 1)
+
+        #expect(backup == backupReply)
+        #expect(await cluster.stats(on: cluster.backup).count == 3)
+        #expect(await cluster.stats(on: cluster.leader).isEmpty)
+        await client.shutdown()
+    }
+
+    @Test func aReplicaTheRosterDoesNotListFailsAfterOneRosterRead() async throws {
+        let cluster = try await TwoReplicas.started { _ in .ok(backupReply) }
+        defer { cluster.stop() }
+        let client = try await cluster.signedInClient()
+        _ = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [], replica: 1)
+        let rosterReads = await cluster.rosterReads()
+
+        await expectCode(.invalidIdentifier) { _ = try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [], replica: 2) }
+
+        #expect(await cluster.rosterReads() == rosterReads + 1)
+        #expect(await cluster.stats(on: cluster.backup).count == 1)
+        await client.shutdown()
+    }
+
+    @Test func replicatedPartitionReadAndSessionCommandsAreRefusedBeforeAnythingIsSent() async throws {
+        let cluster = try await TwoReplicas.started { _ in .ok(backupReply) }
+        defer { cluster.stop() }
+        let client = try await cluster.signedInClient()
+        let requests = await cluster.leader.requests.count + cluster.backup.requests.count
+        let refused: [CommandCode] = [.createStream, .pollMessages, .getConsumerOffset, .loginUser, .logoutUser]
+
+        for code in refused {
+            await expectCode(.invalidCommand) { _ = try await client.sendRawRequest(code: code.rawValue, payload: [], replica: 1) }
+        }
+
+        #expect(await cluster.leader.requests.count + cluster.backup.requests.count == requests)
+        #expect(try await client.sendRawRequest(code: CommandCode.getStats.rawValue, payload: [], replica: 1) == backupReply)
+        await client.shutdown()
+    }
+}
+
+/// A leader that commits metadata writes at `metadataCommit` and a backup
+/// that answers its stats requests as the test scripts each attempt,
+/// numbered from 1.
+private struct TwoReplicas {
+    let leader: FakeVSRServer
+    let backup: FakeVSRServer
+
+    static func started(backupStats: @escaping @Sendable (Int) -> FakeVSRServer.Reply) async throws -> TwoReplicas {
+        let leader = try await FakeVSRServer.started()
+        let backup = try await FakeVSRServer.started()
+        let roster = FakeVSRServer.rosterBody([leader.address, backup.address])
+        let attempts = Locked(0)
+        await leader.respond { request in
+            switch request.code.flatMap(CommandCode.init(rawValue:)) {
+            case .getClusterMetadata:
+                return .ok(roster)
+            case .createStream:
+                return .committed([], commit: metadataCommit)
+            case .getStats:
+                return .ok(leaderReply)
+            default:
+                return FakeVSRServer.defaultReply(request, address: leader.address)
+            }
+        }
+        await backup.respond { request in
+            switch request.code.flatMap(CommandCode.init(rawValue:)) {
+            case .getClusterMetadata:
+                return .ok(roster)
+            case .getStats:
+                return backupStats(
+                    attempts.withLock { attempt in
+                        attempt += 1
+                        return attempt
+                    })
+            default:
+                return FakeVSRServer.defaultReply(request, address: backup.address)
+            }
+        }
+        return TwoReplicas(leader: leader, backup: backup)
+    }
+
+    func signedInClient() async throws -> IggyClient {
+        var configuration = ClientConfiguration(address: leader.address, autoLogin: .usernamePassword(username: "iggy", password: "iggy"))
+        configuration.reconnection = ReconnectionOptions(maxRetries: nil, interval: .milliseconds(20), reestablishAfter: .zero)
+        configuration.requestTimeout = .seconds(10)
+        configuration.heartbeatInterval = .seconds(60)
+        let client = IggyClient(configuration: configuration, connector: NIOConnector(), logger: testLogger)
+        try await client.connect()
+        return client
+    }
+
+    func stats(on node: FakeVSRServer) async -> [FakeVSRServer.Request] {
+        await node.requests.filter { $0.code == CommandCode.getStats.rawValue }
+    }
+
+    func rosterReads() async -> Int {
+        let reads = { (node: FakeVSRServer) async in await node.requests.filter { $0.code == CommandCode.getClusterMetadata.rawValue }.count }
+        return await reads(leader) + reads(backup)
+    }
+
+    func stop() {
+        Task {
+            await leader.stop()
+            await backup.stop()
+        }
+    }
+}

diff --git a/foreign/swift/Tools/golden-vectors/src/main.rs b/foreign/swift/Tools/golden-vectors/src/main.rs
index d850acb..2c009d1 100644
--- a/foreign/swift/Tools/golden-vectors/src/main.rs
+++ b/foreign/swift/Tools/golden-vectors/src/main.rs

@@ -208,7 +208,7 @@
     ClientVersionInfo {
         protocol_version: IGGY_PROTOCOL_VERSION,
         sdk_name: name("swift-sdk"),
-        sdk_version: name("0.1.2"),
+        sdk_version: name("0.1.3"),
     }
 }