diff --git a/mssql-mock-tds/src/protocol.rs b/mssql-mock-tds/src/protocol.rs
index 3ddc6210..8bc608f0 100644
--- a/mssql-mock-tds/src/protocol.rs
+++ b/mssql-mock-tds/src/protocol.rs
@@ -853,20 +853,78 @@ pub fn build_query_result(response: &crate::query_response::QueryResponse) -> By
         result.extend_from_slice(&build_info_token(info));
     }
 
-    // Serialize each row
-    for row in &response.rows {
-        result.put_u8(TokenType::Row as u8);
-        for value in &row.values {
-            value.write_to_buffer(&mut result);
+    match &response.error_after {
+        Some(err) => {
+            // Stream the first `after_rows` rows, then an ERROR token and a
+            // terminal DONE (row count 0) — no trailing rows.
+            for row in response.rows.iter().take(err.after_rows) {
+                result.put_u8(TokenType::Row as u8);
+                for value in &row.values {
+                    value.write_to_buffer(&mut result);
+                }
+            }
+            result.extend_from_slice(&build_error_token(
+                err.number,
+                err.state,
+                err.severity,
+                &err.message,
+            ));
+            // INFO tokens emitted during the drain (after ERROR, before DONE).
+            for info in &err.drain_info {
+                result.extend_from_slice(&build_info_token(info));
+            }
+            result.extend_from_slice(&build_done_token(0));
         }
-    }
+        None => {
+            // Serialize each row
+            for row in &response.rows {
+                result.put_u8(TokenType::Row as u8);
+                for value in &row.values {
+                    value.write_to_buffer(&mut result);
+                }
+            }
 
-    // DONE token
-    result.extend_from_slice(&build_done_token(response.rows.len() as u64));
+            // DONE token
+            result.extend_from_slice(&build_done_token(response.rows.len() as u64));
+        }
+    }
 
     wrap_in_packet(PacketType::TabularResult, result)
 }
 
+/// Build a bare ERROR token (0xAA) with no surrounding DONE or packet framing,
+/// for injecting mid-stream into a result set.
+fn build_error_token(number: u32, state: u8, severity: u8, message: &str) -> BytesMut {
+    let mut token = BytesMut::new();
+    token.put_u8(TokenType::Error as u8);
+
+    let length_pos = token.len();
+    token.put_u16_le(0); // Placeholder for length (little-endian on the wire)
+
+    token.put_u32_le(number);
+    token.put_u8(state);
+    token.put_u8(severity);
+
+    // Message (US_VARCHAR: u16 code-unit count + UTF-16LE)
+    token.put_u16_le(message.chars().count() as u16);
+    for ch in message.encode_utf16() {
+        token.put_u16_le(ch);
+    }
+
+    // Server name / procedure name (empty B_VARCHARs)
+    token.put_u8(0);
+    token.put_u8(0);
+
+    // Line number
+    token.put_u32_le(1);
+
+    let token_length = (token.len() - length_pos - 2) as u16;
+    let mut length_bytes = &mut token[length_pos..length_pos + 2];
+    length_bytes.put_u16_le(token_length);
+
+    token
+}
+
 /// Build an error response
 pub fn build_error_response(message: &str) -> BytesMut {
     let mut response = BytesMut::new();
diff --git a/mssql-mock-tds/src/query_response.rs b/mssql-mock-tds/src/query_response.rs
index d81483e7..922bf477 100644
--- a/mssql-mock-tds/src/query_response.rs
+++ b/mssql-mock-tds/src/query_response.rs
@@ -162,12 +162,31 @@ impl InfoMessage {
     }
 }
 
+/// A server ERROR token injected partway through a result set, after
+/// `after_rows` rows have been streamed, followed by a terminal DONE. Used to
+/// exercise the fetch-time error/drain path.
+#[derive(Debug, Clone)]
+pub struct MidStreamError {
+    pub after_rows: usize,
+    pub number: u32,
+    pub state: u8,
+    pub severity: u8,
+    pub message: String,
+    /// INFO tokens emitted after the ERROR and before the terminal DONE, so the
+    /// fetch-time drain path (async `drain_stream` / sync blocking drain) is
+    /// exercised on Info capture, not just the happy pre-error stream.
+    pub drain_info: Vec<InfoMessage>,
+}
+
 /// A complete query response definition
 #[derive(Debug, Clone)]
 pub struct QueryResponse {
     pub columns: Vec<ColumnDefinition>,
     pub rows: Vec<Row>,
     pub info_tokens: Vec<InfoMessage>,
+    /// When set, only the first `after_rows` rows are streamed, then an ERROR
+    /// token and a terminal DONE are emitted (no trailing rows).
+    pub error_after: Option<MidStreamError>,
 }
 
 impl QueryResponse {
@@ -177,6 +196,7 @@ impl QueryResponse {
             columns,
             rows,
             info_tokens: Vec::new(),
+            error_after: None,
         }
     }
 
@@ -185,12 +205,19 @@ impl QueryResponse {
         self
     }
 
+    /// Inject a mid-stream ERROR token after `after_rows` rows.
+    pub fn with_error_after(mut self, error_after: MidStreamError) -> Self {
+        self.error_after = Some(error_after);
+        self
+    }
+
     /// Helper to create a response for SELECT 1
     pub fn select_one() -> Self {
         Self {
             columns: vec![ColumnDefinition::new("", SqlDataType::Int)],
             rows: vec![Row::new(vec![ColumnValue::Int(1)])],
             info_tokens: Vec::new(),
+            error_after: None,
         }
     }
 
@@ -208,6 +235,7 @@ impl QueryResponse {
                 ColumnValue::Int(3),
             ])],
             info_tokens: Vec::new(),
+            error_after: None,
         }
     }
 }
diff --git a/mssql-tds/src/connection.rs b/mssql-tds/src/connection.rs
index 0eaad446..76bc5d7b 100644
--- a/mssql-tds/src/connection.rs
+++ b/mssql-tds/src/connection.rs
@@ -23,5 +23,7 @@ pub(crate) mod metadata_retriever;
 pub(crate) mod session_recovery;
 /// Primary client type and result set traits.
 pub mod tds_client;
+/// Synchronous, reactor-free row-fetch client over the blocking TDS edge.
+pub mod tds_sync_client;
 /// Transport layer (TCP, Named Pipes, Shared Memory).
 pub mod transport;
diff --git a/mssql-tds/src/connection/tds_client.rs b/mssql-tds/src/connection/tds_client.rs
index 7fd3155b..45636b54 100644
--- a/mssql-tds/src/connection/tds_client.rs
+++ b/mssql-tds/src/connection/tds_client.rs
@@ -78,6 +78,27 @@ enum ActiveRowReadState {
     PlpPaused(Box<PlpPauseState>),
 }
 
+/// Outcome of applying a single non-row token during row iteration.
+///
+/// [`TdsClient::apply_row_read_token`] performs every token side-effect and
+/// returns this so the async ([`TdsClient::handle_row_read_token`]) and sync
+/// ([`TdsSyncClient`](crate::connection::tds_sync_client::TdsSyncClient)) fetch
+/// shells share one authoritative handler — their per-token state mutations
+/// cannot drift. The only flavour-specific step is the ERROR drain, which each
+/// shell performs in its own edge before calling
+/// [`TdsClient::finalize_row_error`].
+#[derive(Debug)]
+pub(crate) enum TokenOutcome {
+    /// A non-terminal token was handled; keep reading the stream.
+    Continue,
+    /// A terminal DONE closed the result set; no row was produced.
+    Terminal,
+    /// An ERROR token was seen. The shell must drain the rest of the batch to
+    /// its terminal DONE in its own flavour, extend these errors with any
+    /// collected during the drain, then call [`TdsClient::finalize_row_error`].
+    DrainThenError(Vec<SqlErrorInfo>),
+}
+
 /// Active TDS connection to a SQL Server instance.
 ///
 /// Created by [`TdsConnectionProvider::create_client()`](crate::connection_provider::tds_connection_provider::TdsConnectionProvider::create_client).
@@ -2961,7 +2982,11 @@ impl TdsClient {
         }
     }
 
-    async fn handle_row_read_token(&mut self, token: Tokens) -> TdsResult<Option<bool>> {
+    /// Applies one non-row token's side-effects to the connection state and
+    /// reports how the fetch shell should proceed. Shared verbatim by the async
+    /// and sync fetch paths (see [`TokenOutcome`]); contains no I/O, so it is a
+    /// plain synchronous method callable from either edge.
+    pub(crate) fn apply_row_read_token(&mut self, token: Tokens) -> TdsResult<TokenOutcome> {
         match token {
             Tokens::DoneInProc(done) | Tokens::DoneProc(done) | Tokens::Done(done) => {
                 info!("done while get_next_row: {:?}", done);
@@ -2981,11 +3006,11 @@ impl TdsClient {
                     info!("No more rows for current command: {:?}", done.cur_cmd);
                     self.execution_context.set_has_open_batch(false);
                 }
-                Ok(Some(false))
+                Ok(TokenOutcome::Terminal)
             }
             Tokens::Order(order_token) => {
                 info!(?order_token);
-                Ok(None)
+                Ok(TokenOutcome::Continue)
             }
             Tokens::EnvChange(env_change) => {
                 info!(?env_change);
@@ -2994,31 +3019,23 @@ impl TdsClient {
                 }
                 self.execution_context
                     .capture_change_property(&env_change, &mut self.negotiated_settings)?;
-                Ok(None)
+                Ok(TokenOutcome::Continue)
             }
             Tokens::SessionState(session_state) => {
                 self.recovery_context
                     .process_session_state(&session_state)?;
-                Ok(None)
+                Ok(TokenOutcome::Continue)
             }
             Tokens::ReturnValue(return_value_token) => {
                 let return_value = self.finalize_return_value(return_value_token)?;
                 self.push_return_value(return_value);
-                Ok(None)
+                Ok(TokenOutcome::Continue)
             }
             Tokens::Error(error_token) => {
                 info!(?error_token);
-                let mut all_errors = vec![SqlErrorInfo::from(&error_token)];
-                let drain_errors = self.drain_stream().await?;
-                all_errors.extend(drain_errors);
-                // The error drained the rest of the batch to its terminal
-                // DONE, so the connection is idle again. Clear the batch
-                // state so a subsequent `close_query` / `advance` does not
-                // block trying to read a stream that is already consumed.
-                self.execution_context.set_has_open_batch(false);
-                self.current_result_set_has_been_read_till_end = true;
-                self.current_metadata = None;
-                Err(crate::error::Error::from_sql_errors(all_errors))
+                Ok(TokenOutcome::DrainThenError(vec![SqlErrorInfo::from(
+                    &error_token,
+                )]))
             }
             Tokens::ColMetadata(_) => Err(crate::error::Error::UsageError(
                 "Unexpected ColMetadata token encountered while reading rows. \
@@ -3029,15 +3046,44 @@ impl TdsClient {
             Tokens::Info(info_token) => {
                 info!(?info_token);
                 self.capture_info_message(&info_token);
-                Ok(None)
+                Ok(TokenOutcome::Continue)
             }
-            Tokens::TabName | Tokens::ColInfo => Ok(None),
+            Tokens::TabName | Tokens::ColInfo => Ok(TokenOutcome::Continue),
             _ => Err(crate::error::Error::ProtocolError(format!(
                 "Unexpected token while finding the next row: {token:?}"
             ))),
         }
     }
 
+    /// Finalizes connection state after a fetch-time ERROR has been drained to
+    /// its terminal DONE, and builds the surfaced error. Shared by the async and
+    /// sync fetch shells so their post-error cleanup cannot drift.
+    pub(crate) fn finalize_row_error(
+        &mut self,
+        all_errors: Vec<SqlErrorInfo>,
+    ) -> crate::error::Error {
+        // The error drained the rest of the batch to its terminal DONE, so the
+        // connection is idle again. Clear the batch state so a subsequent
+        // `close_query` / `advance` does not block trying to read a stream that
+        // is already consumed.
+        self.execution_context.set_has_open_batch(false);
+        self.current_result_set_has_been_read_till_end = true;
+        self.current_metadata = None;
+        crate::error::Error::from_sql_errors(all_errors)
+    }
+
+    async fn handle_row_read_token(&mut self, token: Tokens) -> TdsResult<Option<bool>> {
+        match self.apply_row_read_token(token)? {
+            TokenOutcome::Continue => Ok(None),
+            TokenOutcome::Terminal => Ok(Some(false)),
+            TokenOutcome::DrainThenError(mut all_errors) => {
+                let drain_errors = self.drain_stream().await?;
+                all_errors.extend(drain_errors);
+                Err(self.finalize_row_error(all_errors))
+            }
+        }
+    }
+
     /// Returns a clone of all [`ReturnValue`]s collected during the current
     /// batch — output parameters and UDF return values.
     ///
@@ -3452,6 +3498,56 @@ impl TdsClient {
 
         Ok(())
     }
+
+    /// Consumes this async client and flips its connection to the synchronous,
+    /// reactor-free fetch edge ([`TdsSyncClient`](crate::connection::tds_sync_client::TdsSyncClient)),
+    /// when the transport is a raw TCP socket that can be handed off as an owned
+    /// blocking stream.
+    ///
+    /// This is the owning half of the owning-reversible model: the returned
+    /// [`TdsSyncClient`](crate::connection::tds_sync_client::TdsSyncClient) holds
+    /// the connection by value, so the socket stays in blocking mode across an
+    /// entire result set (no per-row flip). Revert with
+    /// [`into_async`](crate::connection::tds_sync_client::TdsSyncClient::into_async)
+    /// to run control-plane work (execute/advance/close).
+    ///
+    /// - [`SyncConversion::Converted`](crate::connection::tds_sync_client::SyncConversion::Converted):
+    ///   raw TCP; the socket was flipped to std-blocking and any buffered residual
+    ///   bytes were carried over intact.
+    /// - [`SyncConversion::NotEligible`](crate::connection::tds_sync_client::SyncConversion::NotEligible):
+    ///   a TLS (or otherwise non-extractable) transport; the async client is
+    ///   returned **unchanged** so the caller keeps using `block_on` — not an error.
+    /// - [`SyncConversion::Failed`](crate::connection::tds_sync_client::SyncConversion::Failed):
+    ///   the flip was attempted but constructing the blocking edge failed.
+    pub fn into_sync(mut self) -> crate::connection::tds_sync_client::SyncConversion {
+        use crate::connection::tds_sync_client::{SyncConversion, TdsSyncClient};
+
+        let runtime_handle = tokio::runtime::Handle::try_current().ok();
+        let packet_size = self.transport.packet_size();
+        let cancel = self.cancel_handle.as_ref().map(|h| h.cancel_token.clone());
+        let request_timeout = self.remaining_request_timeout;
+
+        let (std_stream, residual) = match self.transport.take_blocking_parts() {
+            Some(parts) => parts,
+            None => return SyncConversion::NotEligible(self),
+        };
+
+        let source = match crate::io::std_byte_source::StdTcpByteSource::new(std_stream, cancel) {
+            Ok(source) => source,
+            Err(err) => return SyncConversion::Failed(err),
+        };
+        let reader = crate::io::blocking_reader::BlockingPacketReader::with_seeded_buffer(
+            source,
+            packet_size as usize,
+            &residual,
+        );
+        SyncConversion::Converted(TdsSyncClient::from_established(
+            self,
+            reader,
+            runtime_handle,
+            request_timeout,
+        ))
+    }
 }
 
 #[async_trait]
@@ -3662,7 +3758,7 @@ mod tests {
         ParserContext, RowPauseState, RowReadResult, TdsTokenStreamReader,
     };
     use crate::token::tokens::{
-        ColMetadataToken, CurrentCommand, DoneStatus, DoneToken, InfoToken, Tokens,
+        ColMetadataToken, CurrentCommand, DoneStatus, DoneToken, ErrorToken, InfoToken, Tokens,
     };
     use async_trait::async_trait;
     use std::collections::VecDeque;
@@ -4463,6 +4559,171 @@ mod tests {
         assert_eq!(rv.value, ColumnValues::Int(7));
     }
 
+    // ── Characterization of the shared fetch-token handler ──
+    //
+    // These pin each side-effect that `apply_row_read_token` / `finalize_row_error`
+    // own, so the async and sync fetch shells (which both delegate to them) cannot
+    // drift. EnvChange / SessionState arms need wire-parsed container tokens to
+    // construct and are covered by the pre-existing async integration suite.
+
+    fn error_token(number: u32, severity: u8, message: &str) -> Tokens {
+        Tokens::Error(ErrorToken {
+            number,
+            state: 1,
+            severity,
+            message: message.to_string(),
+            server_name: "test-server".to_string(),
+            proc_name: String::new(),
+            line_number: 7,
+        })
+    }
+
+    #[test]
+    fn apply_row_read_token_done_terminal_accounts_and_closes_batch() {
+        let mut client = create_test_client();
+        client.execution_context.set_has_open_batch(true);
+
+        let outcome = client
+            .apply_row_read_token(done_count(CurrentCommand::Insert, 3, false))
+            .unwrap();
+
+        assert!(matches!(outcome, TokenOutcome::Terminal));
+        assert_eq!(client.count_map.get(&CurrentCommand::Insert), Some(&3));
+        assert!(client.current_result_set_has_been_read_till_end);
+        assert!(!client.execution_context.has_open_batch());
+    }
+
+    #[test]
+    fn apply_row_read_token_done_more_accumulates_and_keeps_batch_open() {
+        let mut client = create_test_client();
+        client.execution_context.set_has_open_batch(true);
+
+        // Two non-terminal DONE_COUNT tokens for the same command accumulate.
+        client
+            .apply_row_read_token(done_count(CurrentCommand::Insert, 2, true))
+            .unwrap();
+        let outcome = client
+            .apply_row_read_token(done_count(CurrentCommand::Insert, 5, true))
+            .unwrap();
+
+        assert!(matches!(outcome, TokenOutcome::Terminal));
+        assert_eq!(client.count_map.get(&CurrentCommand::Insert), Some(&7));
+        assert!(client.current_result_set_has_been_read_till_end);
+        // DONE_MORE => the batch stays open.
+        assert!(client.execution_context.has_open_batch());
+    }
+
+    #[test]
+    fn apply_row_read_token_done_with_error_flag_is_protocol_error() {
+        let mut client = create_test_client();
+        let done = Tokens::Done(DoneToken {
+            status: DoneStatus::FINAL | DoneStatus::ERROR,
+            cur_cmd: CurrentCommand::Select,
+            row_count: 0,
+        });
+        assert!(matches!(
+            client.apply_row_read_token(done),
+            Err(crate::error::Error::ProtocolError(_))
+        ));
+    }
+
+    #[test]
+    fn apply_row_read_token_order_continues_without_side_effects() {
+        let mut client = create_test_client();
+        let outcome = client
+            .apply_row_read_token(Tokens::Order(crate::token::tokens::OrderToken {
+                _order_columns: vec![1],
+            }))
+            .unwrap();
+        assert!(matches!(outcome, TokenOutcome::Continue));
+        assert!(client.count_map.is_empty());
+        assert!(!client.current_result_set_has_been_read_till_end);
+    }
+
+    #[test]
+    fn apply_row_read_token_info_captures_message_and_continues() {
+        let mut client = create_test_client();
+        let outcome = client
+            .apply_row_read_token(info_token(50_000, 10, "print hello"))
+            .unwrap();
+        assert!(matches!(outcome, TokenOutcome::Continue));
+        assert_eq!(client.info_messages.len(), 1);
+        assert_eq!(client.info_messages[0].message, "print hello");
+    }
+
+    #[test]
+    fn apply_row_read_token_return_value_is_pushed_and_continues() {
+        let mut client = create_test_client();
+        let token = ae_return_value_token("@out", ColumnValues::Int(9), None);
+        let outcome = client
+            .apply_row_read_token(Tokens::ReturnValue(token))
+            .unwrap();
+        assert!(matches!(outcome, TokenOutcome::Continue));
+        let values = client.get_return_values();
+        assert_eq!(values.len(), 1);
+        assert_eq!(values[0].value, ColumnValues::Int(9));
+    }
+
+    #[test]
+    fn apply_row_read_token_error_defers_drain_without_mutation() {
+        let mut client = create_test_client();
+        client.execution_context.set_has_open_batch(true);
+
+        let outcome = client
+            .apply_row_read_token(error_token(208, 16, "invalid object name"))
+            .unwrap();
+
+        match outcome {
+            TokenOutcome::DrainThenError(errors) => {
+                assert_eq!(errors.len(), 1);
+                assert_eq!(errors[0].number, 208);
+            }
+            other => panic!("expected DrainThenError, got {other:?}"),
+        }
+        // The Error arm defers all state mutation to the post-drain finalize.
+        assert!(client.count_map.is_empty());
+        assert!(!client.current_result_set_has_been_read_till_end);
+        assert!(client.execution_context.has_open_batch());
+    }
+
+    #[test]
+    fn apply_row_read_token_colmetadata_is_usage_error() {
+        let mut client = create_test_client();
+        assert!(matches!(
+            client.apply_row_read_token(empty_col_metadata()),
+            Err(crate::error::Error::UsageError(_))
+        ));
+    }
+
+    #[test]
+    fn finalize_row_error_clears_batch_state_and_builds_server_error() {
+        let mut client = create_test_client();
+        client.execution_context.set_has_open_batch(true);
+        client.current_metadata = Some(stale_metadata());
+
+        let errors = vec![SqlErrorInfo::from(&ErrorToken {
+            number: 2601,
+            state: 1,
+            severity: 14,
+            message: "duplicate key".to_string(),
+            server_name: "test-server".to_string(),
+            proc_name: String::new(),
+            line_number: 1,
+        })];
+        let err = client.finalize_row_error(errors);
+
+        assert!(!client.execution_context.has_open_batch());
+        assert!(client.current_result_set_has_been_read_till_end);
+        assert!(client.current_metadata.is_none());
+        match err {
+            crate::error::Error::SqlServerError { diagnostics } => {
+                assert_eq!(diagnostics.errors.len(), 1);
+                assert_eq!(diagnostics.errors[0].number, 2601);
+            }
+            other => panic!("expected SqlServerError, got {other:?}"),
+        }
+    }
+
     #[test]
     fn finalize_return_value_passes_through_ciphertext_when_disabled() {
         // Encrypted value but the command disabled AE => ciphertext is passed
diff --git a/mssql-tds/src/connection/tds_sync_client.rs b/mssql-tds/src/connection/tds_sync_client.rs
new file mode 100644
index 00000000..3ca8683c
--- /dev/null
+++ b/mssql-tds/src/connection/tds_sync_client.rs
@@ -0,0 +1,405 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+//! Synchronous, reactor-free row-fetch client over the blocking TDS edge.
+//!
+//! [`TdsSyncClient`] is the first surfaced consumer of the L3 blocking row
+//! driver. It mirrors the per-row *fetch* surface of the async
+//! [`TdsClient`](crate::connection::tds_client::TdsClient) — the methods a
+//! `block_on(next_row())` consumer calls in its hot loop — but drives them with
+//! no tokio reactor: it reads TDS packets straight off an owned blocking socket
+//! and re-uses [`TdsCore::step_row`](crate::io::tds_core::TdsCore) verbatim via
+//! [`drive_row_over_buffer_blocking`]. The parse body is identical to the async
+//! path; only the byte-pull edge blocks instead of awaiting.
+//!
+//! # OWNING (into_sync/into_async), NOT borrowed
+//!
+//! > OWNING (into_sync/into_async), NOT borrowed: a SyncRowFetcher<'a> borrowing
+//! > dbc.client cannot persist across SQLFetch FFI calls (storing it beside the
+//! > client it borrows = self-referential, illegal in safe Rust) ⇒ re-created
+//! > every SQLFetch ⇒ fd into_std/revert flip PER FETCH. On our base
+//! > fetch.rs:133 = block_on(next_row()) PER ROW ⇒ borrowed flips PER ROW
+//! > (millions/1M-row set). Owning stores TdsSyncClient by value ⇒ fd stays
+//! > blocking across the whole result set (~2 flips). Prototype ii revert PROVEN
+//! > sound @13.2µs. Borrowed is DEAD; do not reintroduce.
+//!
+//! # Reversible, terminal drop
+//!
+//! The flip is reversible: [`TdsSyncClient::into_async`] hands the socket back to
+//! a tokio stream and returns the original [`TdsClient`], so control-plane work
+//! (execute/advance/close) stays async. Abandoning a `TdsSyncClient` without
+//! reverting performs a **terminal clean-close** — the owned `std::net::TcpStream`
+//! closes on drop (its natural RAII), which is why there is deliberately no
+//! explicit `Drop` impl. There is **never** a Drop-driven fd-revert-to-async;
+//! that is the dead borrowed RAII per-fetch anti-pattern.
+
+use std::sync::Arc;
+use std::time::{Duration, Instant};
+
+use tokio::runtime::Handle;
+
+use crate::connection::tds_client::{ReturnStatus, TdsClient, TokenOutcome};
+use crate::core::TdsResult;
+use crate::datatypes::column_values::ColumnValues;
+use crate::datatypes::row_writer::{DefaultRowWriter, RowWriter};
+use crate::error::{Error, SqlErrorInfo, SqlInfoMessage};
+use crate::io::blocking_reader::BlockingPacketReader;
+use crate::io::std_byte_source::StdTcpByteSource;
+use crate::io::token_stream::{
+    ParserContext, PlpPauseState, RowPauseState, RowReadResult, drive_row_over_buffer_blocking,
+};
+use crate::query::metadata::ColumnMetadata;
+use crate::token::tokens::{SqlCollation, Tokens};
+
+/// Outcome of [`TdsClient::into_sync`]. The flip is opportunistic: only a raw TCP
+/// transport can be handed to the blocking edge.
+///
+/// The variant payloads are the established clients themselves, so this enum is
+/// intentionally large; boxing would alter the frozen public surface
+/// (`Converted(TdsSyncClient)` / `NotEligible(TdsClient)`), so the size lint is
+/// suppressed rather than changing the shape.
+#[allow(clippy::large_enum_variant)]
+pub enum SyncConversion {
+    /// The connection was a raw TCP socket and is now driven synchronously.
+    Converted(TdsSyncClient),
+    /// The transport could not be flipped (e.g. TLS). The async client is
+    /// returned **unchanged** so the caller keeps using `block_on` — not an error.
+    NotEligible(TdsClient),
+    /// The flip was attempted but constructing the blocking edge failed; the
+    /// connection is unusable.
+    Failed(Error),
+}
+
+/// Mirror of [`ActiveRowReadState`](crate::connection::tds_client) for the sync
+/// edge: the pause cursor carried between fetch calls.
+enum SyncRowState {
+    Idle,
+    RowPaused(Box<RowPauseState>),
+    PlpPaused(Box<PlpPauseState>),
+}
+
+/// A synchronous, reactor-free client exposing the TDS row-fetch surface over an
+/// owned blocking socket. Built by [`TdsClient::into_sync`]; reverted by
+/// [`TdsSyncClient::into_async`]. See the [module docs](self) for the
+/// owning-vs-borrowed rationale and drop semantics.
+///
+/// All connection/result-set state (metadata, INFO buffer, read-till-end flag,
+/// `count_map`, return values) lives on the owned [`TdsClient`] `inner`; token
+/// side-effects flow through [`TdsClient::apply_row_read_token`], the *same*
+/// handler the async path uses, so the two clients cannot drift. This wrapper
+/// adds only the blocking edge and the sync-side pause cursor.
+pub struct TdsSyncClient {
+    /// The async client, gutted of its socket + read residual (both now owned by
+    /// `reader`). It remains the single source of truth for all connection state
+    /// so token side-effects mutate the same fields as the async path, and so
+    /// [`into_async`](Self::into_async) can hand the socket back.
+    inner: TdsClient,
+    reader: BlockingPacketReader<StdTcpByteSource>,
+    /// Empty-slice fallback for [`get_metadata`](Self::get_metadata) when the
+    /// connection has no current metadata.
+    empty_metadata: Vec<ColumnMetadata>,
+    active: SyncRowState,
+    /// Per-request timeout applied as a read deadline before each fetch.
+    request_timeout: Option<Duration>,
+    /// Runtime handle captured at conversion, needed to re-register the socket
+    /// with the reactor in [`into_async`](Self::into_async).
+    runtime_handle: Option<Handle>,
+}
+
+impl TdsSyncClient {
+    /// Builds a sync client from an established (raw-TCP) connection. Called only
+    /// by [`TdsClient::into_sync`], which has already extracted the socket and
+    /// seeded the blocking reader with the transport's residual bytes.
+    pub(crate) fn from_established(
+        inner: TdsClient,
+        reader: BlockingPacketReader<StdTcpByteSource>,
+        runtime_handle: Option<Handle>,
+        request_timeout: Option<Duration>,
+    ) -> Self {
+        Self {
+            inner,
+            reader,
+            empty_metadata: Vec::new(),
+            active: SyncRowState::Idle,
+            request_timeout,
+            runtime_handle,
+        }
+    }
+
+    /// Reverts to the async [`TdsClient`], re-registering the owned socket with
+    /// the tokio reactor and handing back any unconsumed bytes so the async path
+    /// resumes byte-identically.
+    ///
+    /// Errors (fd known-dead / poisoned) if the runtime handle is missing or the
+    /// socket cannot be re-registered; the connection is then consumed and closes.
+    pub fn into_async(mut self) -> TdsResult<TdsClient> {
+        let handle = self.runtime_handle.take().ok_or_else(|| {
+            Error::UsageError(
+                "into_async requires the tokio runtime handle captured at into_sync; \
+                 the connection was created outside a runtime context"
+                    .to_string(),
+            )
+        })?;
+
+        let residual = self.reader.take_residual();
+        let std_stream = self.reader.into_source().into_stream();
+        // Re-arm non-blocking mode before re-registering with the reactor.
+        std_stream.set_nonblocking(true)?;
+
+        let _guard = handle.enter();
+        // `from_std` touches the reactor, so it must run inside `handle.enter()`.
+        let tokio_stream = tokio::net::TcpStream::from_std(std_stream)?;
+
+        let mut inner = self.inner;
+        inner
+            .transport
+            .restore_blocking_parts(tokio_stream, residual)?;
+        Ok(inner)
+    }
+
+    /// Fetches up to `max_rows` rows into `out`, recycling row buffers from
+    /// `spare` to amortize per-row allocation. Stops at `max_rows`, at the
+    /// result-set boundary, or when the set is exhausted; returns the count
+    /// fetched. Authored fresh as a thin loop over the reactor-free
+    /// [`next_row_into`](Self::next_row_into); correctness derives from that
+    /// method's differential parity with the async oracle.
+    pub fn fetch_rows_batch(
+        &mut self,
+        out: &mut Vec<Vec<ColumnValues>>,
+        mut spare: Vec<Vec<ColumnValues>>,
+        max_rows: usize,
+    ) -> TdsResult<usize> {
+        let col_count = self
+            .inner
+            .current_metadata
+            .as_ref()
+            .map_or(0, |m| m.columns.len());
+        let mut fetched = 0usize;
+        while fetched < max_rows {
+            if !self.maybe_has_unread_rows() {
+                break;
+            }
+            let recycled = spare.pop().unwrap_or_else(|| Vec::with_capacity(col_count));
+            let mut writer = DefaultRowWriter::from_recycled(recycled);
+            if self.get_next_row_into(&mut writer)? {
+                out.push(writer.take_row());
+                fetched += 1;
+            } else {
+                break;
+            }
+        }
+        Ok(fetched)
+    }
+
+    /// Fetches the next row, or `None` at the end of the result set. Sync twin of
+    /// [`ResultSet::next_row`](crate::connection::tds_client::ResultSet::next_row).
+    pub fn next_row(&mut self) -> TdsResult<Option<Vec<ColumnValues>>> {
+        if !self.maybe_has_unread_rows() {
+            return Ok(None);
+        }
+        let col_count = self
+            .inner
+            .current_metadata
+            .as_ref()
+            .map_or(0, |m| m.columns.len());
+        let mut writer = DefaultRowWriter::new(col_count);
+        if self.get_next_row_into(&mut writer)? {
+            Ok(Some(writer.take_row()))
+        } else {
+            Ok(None)
+        }
+    }
+
+    /// Decodes the next row directly into `writer`, returning `true` if a row was
+    /// written or `false` at the end of the result set. Sync twin of
+    /// [`ResultSet::next_row_into`](crate::connection::tds_client::ResultSet::next_row_into).
+    pub fn next_row_into(&mut self, writer: &mut (dyn RowWriter + Send)) -> TdsResult<bool> {
+        if !self.maybe_has_unread_rows() {
+            return Ok(false);
+        }
+        self.get_next_row_into(writer)
+    }
+
+    /// Drains and returns the informational (INFO/PRINT) messages captured so
+    /// far. Sync twin of
+    /// [`TdsClient::take_info_messages`](crate::connection::tds_client::TdsClient::take_info_messages).
+    pub fn take_info_messages(&mut self) -> Vec<SqlInfoMessage> {
+        self.inner.take_info_messages()
+    }
+
+    /// The current result set's column metadata, or an empty slice when none is
+    /// available. Sync twin of
+    /// [`ResultSet::get_metadata`](crate::connection::tds_client::ResultSet::get_metadata).
+    pub fn get_metadata(&self) -> &Vec<ColumnMetadata> {
+        self.inner
+            .current_metadata
+            .as_ref()
+            .map(|m| &m.columns)
+            .unwrap_or(&self.empty_metadata)
+    }
+
+    /// Whether more rows may remain in the current result set. Sync twin of
+    /// [`ResultSet::maybe_has_unread_rows`](crate::connection::tds_client::ResultSet::maybe_has_unread_rows).
+    pub fn maybe_has_unread_rows(&self) -> bool {
+        !self.inner.current_result_set_has_been_read_till_end
+    }
+
+    /// Whether the active PLP stream (if any) has reached its end. Sync twin of
+    /// [`ResultSet::active_plp_reached_end`](crate::connection::tds_client::ResultSet::active_plp_reached_end).
+    pub fn active_plp_reached_end(&self) -> bool {
+        match &self.active {
+            SyncRowState::PlpPaused(plp_state) => plp_state.reached_end(),
+            _ => true,
+        }
+    }
+
+    /// The collation of the active PLP stream, if any. Sync twin of
+    /// [`ResultSet::active_plp_collation`](crate::connection::tds_client::ResultSet::active_plp_collation).
+    pub fn active_plp_collation(&self) -> Option<SqlCollation> {
+        match &self.active {
+            SyncRowState::PlpPaused(plp_state) => plp_state.collation(),
+            _ => None,
+        }
+    }
+
+    /// The sync mirror of `TdsClient::get_next_row_into`: drives the L3 blocking
+    /// row driver over the owned buffer, resuming a paused row first and handling
+    /// non-row tokens to the result-set boundary.
+    fn get_next_row_into(&mut self, writer: &mut (dyn RowWriter + Send)) -> TdsResult<bool> {
+        let metadata = match &self.inner.current_metadata {
+            Some(metadata) => Arc::clone(metadata),
+            None => {
+                return Err(Error::UsageError(
+                    "No metadata found while fetching the next row. Have you executed a \
+                     row-returning query on the async client before converting?"
+                        .to_string(),
+                ));
+            }
+        };
+        // Always Encrypted decryption is not wired through the sync fetch path in
+        // v1: a non-empty CEK table means encrypted columns.
+        if !metadata.cek_table.is_empty() {
+            return Err(Error::UnimplementedFeature {
+                feature: "Always Encrypted columns over the synchronous fetch path".to_string(),
+                context: "convert back with into_async() and fetch encrypted result sets \
+                          on the async client"
+                    .to_string(),
+            });
+        }
+
+        let mut resume = match std::mem::replace(&mut self.active, SyncRowState::Idle) {
+            SyncRowState::Idle => None,
+            SyncRowState::RowPaused(pause_state) => Some(*pause_state),
+            SyncRowState::PlpPaused(plp_state) => {
+                // Resuming a PLP-paused row needs the chunked drain
+                // (`read_active_plp_bytes`), which is deferred from the v1 sync
+                // surface. Preserve the pause so the accessors stay accurate.
+                self.active = SyncRowState::PlpPaused(plp_state);
+                return Err(Error::UnimplementedFeature {
+                    feature: "resuming a PLP-paused row over the synchronous fetch path"
+                        .to_string(),
+                    context: "chunked PLP reads (read_active_plp_bytes) are deferred; \
+                              fetch large-object rows on the async client"
+                        .to_string(),
+                });
+            }
+        };
+
+        self.arm_deadline();
+        let context = ParserContext::ColumnMetadata(metadata, None);
+        loop {
+            let result =
+                drive_row_over_buffer_blocking(&mut self.reader, &context, resume.take(), writer)?;
+            match result {
+                RowReadResult::RowWritten => {
+                    writer.end_row();
+                    return Ok(true);
+                }
+                RowReadResult::RowPaused(pause_state) => {
+                    self.active = SyncRowState::RowPaused(Box::new(pause_state));
+                    return Ok(true);
+                }
+                RowReadResult::PlpPaused(plp_state) => {
+                    self.active = SyncRowState::PlpPaused(Box::new(plp_state));
+                    return Ok(true);
+                }
+                RowReadResult::Token(token) => {
+                    if let Some(has_row) = self.handle_row_read_token(token)? {
+                        return Ok(has_row);
+                    }
+                }
+            }
+        }
+    }
+
+    /// The sync fetch shell's non-row-token handler. Delegates every side-effect
+    /// to the shared [`TdsClient::apply_row_read_token`] (identical to the async
+    /// path), differing only in the ERROR drain: the rest of the batch is drained
+    /// to its terminal DONE over the blocking edge before
+    /// [`TdsClient::finalize_row_error`] builds the surfaced error.
+    fn handle_row_read_token(&mut self, token: Tokens) -> TdsResult<Option<bool>> {
+        match self.inner.apply_row_read_token(token)? {
+            TokenOutcome::Continue => Ok(None),
+            TokenOutcome::Terminal => Ok(Some(false)),
+            TokenOutcome::DrainThenError(mut all_errors) => {
+                all_errors.extend(self.drain_stream()?);
+                Err(self.inner.finalize_row_error(all_errors))
+            }
+        }
+    }
+
+    /// The sync mirror of `TdsClient::drain_stream`: after an ERROR token, read
+    /// the rest of the batch to its terminal DONE, collecting further ERROR
+    /// tokens so the surfaced error carries the full diagnostic chain.
+    ///
+    /// Every non-terminal side-effect (`Info`/`EnvChange`/`SessionState`/
+    /// `ReturnValue`) is routed through the shared
+    /// [`TdsClient::apply_row_read_token`] so it cannot drift from the async
+    /// drain; only the terminal DONE, collected ERRORs, and the `ReturnStatus`
+    /// capture are handled inline (mirroring the async arms exactly).
+    fn drain_stream(&mut self) -> TdsResult<Vec<SqlErrorInfo>> {
+        let mut collected_errors: Vec<SqlErrorInfo> = Vec::new();
+        let mut scratch = DefaultRowWriter::new(0);
+        let context = ParserContext::None(());
+        loop {
+            match drive_row_over_buffer_blocking(&mut self.reader, &context, None, &mut scratch)? {
+                RowReadResult::Token(token) => match token {
+                    Tokens::Done(done) | Tokens::DoneProc(done) | Tokens::DoneInProc(done)
+                        if !done.has_more() =>
+                    {
+                        break;
+                    }
+                    Tokens::Error(error_token) => {
+                        collected_errors.push(SqlErrorInfo::from(&error_token));
+                    }
+                    Tokens::ReturnStatus(return_status) => {
+                        self.inner.last_return_status = ReturnStatus::Received(return_status.value);
+                    }
+                    token @ (Tokens::Info(_)
+                    | Tokens::EnvChange(_)
+                    | Tokens::SessionState(_)
+                    | Tokens::ReturnValue(_)) => {
+                        self.inner.apply_row_read_token(token)?;
+                    }
+                    _ => {}
+                },
+                // No metadata context here, so no rows are decoded; anything but a
+                // token during drain is a protocol violation.
+                _ => {
+                    return Err(Error::ProtocolError(
+                        "Unexpected row payload while draining the token stream after an error"
+                            .to_string(),
+                    ));
+                }
+            }
+        }
+        Ok(collected_errors)
+    }
+
+    /// Refreshes the byte source's read deadline from the per-request timeout,
+    /// so cancel/deadline checks ride the same receive edge as the async path.
+    fn arm_deadline(&mut self) {
+        let deadline = self.request_timeout.map(|d| Instant::now() + d);
+        self.reader.source_mut().set_deadline(deadline);
+    }
+}
diff --git a/mssql-tds/src/connection/transport/network_transport.rs b/mssql-tds/src/connection/transport/network_transport.rs
index e875d10b..3eeb9058 100644
--- a/mssql-tds/src/connection/transport/network_transport.rs
+++ b/mssql-tds/src/connection/transport/network_transport.rs
@@ -448,6 +448,28 @@ pub trait Stream: AsyncRead + AsyncWrite + Unpin + Send + Sync {
     fn channel_binding_token(&self) -> Option<Vec<u8>> {
         None
     }
+
+    /// Whether this stream can be handed off to the synchronous blocking edge
+    /// (`TdsSyncClient`) as an owned `std::net::TcpStream`.
+    ///
+    /// Only a raw (plaintext) TCP stream qualifies: a TLS-wrapped stream owns
+    /// engine state that cannot be reconstructed from a bare socket, so it
+    /// reports `false` and the async client is kept for that connection. This is
+    /// a non-consuming probe so a caller can decline the flip without having
+    /// taken the stream.
+    fn supports_blocking_extraction(&self) -> bool {
+        false
+    }
+
+    /// Consumes the stream and yields its underlying blocking
+    /// `std::net::TcpStream`, or `None` when the stream is not a raw TCP socket
+    /// (e.g. TLS). Callers must gate on [`supports_blocking_extraction`] first so
+    /// a declined flip leaves the async stream intact.
+    ///
+    /// [`supports_blocking_extraction`]: Stream::supports_blocking_extraction
+    fn into_blocking_std(self: Box<Self>) -> Option<std::net::TcpStream> {
+        None
+    }
 }
 
 impl Stream for TcpStream {
@@ -476,6 +498,18 @@ impl Stream for TcpStream {
             Ok(_) => false,
         }
     }
+
+    fn supports_blocking_extraction(&self) -> bool {
+        true
+    }
+
+    fn into_blocking_std(self: Box<Self>) -> Option<std::net::TcpStream> {
+        let std_stream = (*self).into_std().ok()?;
+        // The synchronous edge drives blocking reads/writes; a tokio stream is
+        // registered non-blocking, so restore blocking mode for the std socket.
+        std_stream.set_nonblocking(false).ok()?;
+        Some(std_stream)
+    }
 }
 
 impl Stream for Box<dyn Stream> {
@@ -494,6 +528,14 @@ impl Stream for Box<dyn Stream> {
     fn channel_binding_token(&self) -> Option<Vec<u8>> {
         (**self).channel_binding_token()
     }
+
+    fn supports_blocking_extraction(&self) -> bool {
+        (**self).supports_blocking_extraction()
+    }
+
+    fn into_blocking_std(self: Box<Self>) -> Option<std::net::TcpStream> {
+        (*self).into_blocking_std()
+    }
 }
 
 pub(crate) struct NetworkTransport {
@@ -759,6 +801,12 @@ impl NetworkTransport {
             })?;
 
         info!("Successfully disabled TLS, reverting to unencrypted stream");
+        // TLS is now disabled and the reclaimed stream is a pure plaintext
+        // passthrough to the underlying socket. Mark the handshake complete so
+        // the `TlsOverTdsStream` reports itself in passthrough mode, which lets
+        // the connection expose its raw socket for the synchronous blocking edge.
+        let mut base_stream = base_stream;
+        base_stream.tls_handshake_completed();
         self.stream = Some(base_stream);
         Ok(())
     }
@@ -1391,6 +1439,43 @@ impl crate::connection::transport::tds_transport::TdsTransport for NetworkTransp
     fn connection_known_dead(&self) -> bool {
         self.known_dead
     }
+
+    fn take_blocking_parts(
+        &mut self,
+    ) -> Option<(std::net::TcpStream, crate::io::packet_buffer::ResidualBytes)> {
+        // Probe eligibility without consuming: a TLS stream cannot be flipped, so
+        // leave the transport untouched and let the caller keep the async client.
+        let eligible = self
+            .stream
+            .as_ref()
+            .is_some_and(|s| s.supports_blocking_extraction());
+        if !eligible {
+            return None;
+        }
+        let boxed = self.stream.take()?;
+        let std_stream = match boxed.into_blocking_std() {
+            Some(std_stream) => std_stream,
+            None => {
+                // Eligibility said raw TCP, so this is unreachable; if it ever
+                // fires the socket is already gone, so mark the connection dead
+                // rather than silently losing it.
+                self.known_dead = true;
+                return None;
+            }
+        };
+        let residual = self.tds_read_buffer.take_residual();
+        Some((std_stream, residual))
+    }
+
+    fn restore_blocking_parts(
+        &mut self,
+        stream: tokio::net::TcpStream,
+        residual: crate::io::packet_buffer::ResidualBytes,
+    ) -> TdsResult<()> {
+        self.stream = Some(Box::new(stream));
+        self.tds_read_buffer.seed_residual(&residual);
+        Ok(())
+    }
 }
 
 #[cfg(test)]
diff --git a/mssql-tds/src/connection/transport/ssl_handler.rs b/mssql-tds/src/connection/transport/ssl_handler.rs
index 5b37d485..f7fb1493 100644
--- a/mssql-tds/src/connection/transport/ssl_handler.rs
+++ b/mssql-tds/src/connection/transport/ssl_handler.rs
@@ -538,6 +538,23 @@ impl<S: Stream> Stream for TlsOverTdsStream<S> {
     fn is_connection_dead(&self) -> bool {
         self.wrapped_stream.is_connection_dead()
     }
+
+    fn supports_blocking_extraction(&self) -> bool {
+        // Once the handshake is complete this stream is a pure passthrough to
+        // the wrapped stream (see `poll_read`/`poll_write`). In "Login Only"
+        // encryption mode TLS is disabled after login, so the surviving
+        // `TlsOverTdsStream` wraps the original raw TCP socket. Delegate the
+        // blocking-extraction probe so such a connection can flip to the
+        // synchronous edge instead of being pinned to the async client.
+        self.has_completed_tls_handshake && self.wrapped_stream.supports_blocking_extraction()
+    }
+
+    fn into_blocking_std(self: Box<Self>) -> Option<std::net::TcpStream> {
+        if !self.has_completed_tls_handshake {
+            return None;
+        }
+        Box::new(self.wrapped_stream).into_blocking_std()
+    }
 }
 
 #[cfg(target_os = "macos")]
@@ -555,6 +572,18 @@ impl Stream for BufferedTdsStream {
     fn is_connection_dead(&self) -> bool {
         self.tls_over_tds_stream.is_connection_dead()
     }
+
+    fn supports_blocking_extraction(&self) -> bool {
+        self.buffer.as_ref().is_none_or(|b| b.is_empty())
+            && self.tls_over_tds_stream.supports_blocking_extraction()
+    }
+
+    fn into_blocking_std(self: Box<Self>) -> Option<std::net::TcpStream> {
+        if !self.buffer.as_ref().is_none_or(|b| b.is_empty()) {
+            return None;
+        }
+        Box::new(self.tls_over_tds_stream).into_blocking_std()
+    }
 }
 
 #[cfg(target_os = "macos")]
diff --git a/mssql-tds/src/connection/transport/tds_transport.rs b/mssql-tds/src/connection/transport/tds_transport.rs
index 3309cef5..8fd57d14 100644
--- a/mssql-tds/src/connection/transport/tds_transport.rs
+++ b/mssql-tds/src/connection/transport/tds_transport.rs
@@ -80,4 +80,34 @@ pub(crate) trait TdsTransport: TdsTokenStreamReader + Send + Sync + std::fmt::De
     fn connection_known_dead(&self) -> bool {
         false
     }
+
+    /// Detaches the raw blocking socket and any buffered residual bytes for the
+    /// synchronous fetch edge (`TdsSyncClient`), returning `None` when this
+    /// transport cannot be flipped (TLS, named pipes, mock). The probe is
+    /// non-consuming for ineligible transports: the caller's client stays intact
+    /// so it can keep using the async path.
+    ///
+    /// On `Some`, the stream is removed from this transport and the read buffer
+    /// is drained of its residual, so the transport must not be read again until
+    /// [`restore_blocking_parts`](TdsTransport::restore_blocking_parts) reinstates
+    /// both.
+    fn take_blocking_parts(
+        &mut self,
+    ) -> Option<(std::net::TcpStream, crate::io::packet_buffer::ResidualBytes)> {
+        None
+    }
+
+    /// Reinstates a previously detached socket (reverted to a tokio stream by the
+    /// synchronous edge) and reseeds the read buffer with the residual bytes that
+    /// straddled the flip, so the async path resumes byte-identically.
+    fn restore_blocking_parts(
+        &mut self,
+        _stream: tokio::net::TcpStream,
+        _residual: crate::io::packet_buffer::ResidualBytes,
+    ) -> TdsResult<()> {
+        Err(crate::error::Error::UnimplementedFeature {
+            feature: "blocking transport handoff".to_string(),
+            context: "this transport does not support synchronous blocking extraction".to_string(),
+        })
+    }
 }
diff --git a/mssql-tds/src/datatypes/row_writer.rs b/mssql-tds/src/datatypes/row_writer.rs
index eb8054e4..361c6966 100644
--- a/mssql-tds/src/datatypes/row_writer.rs
+++ b/mssql-tds/src/datatypes/row_writer.rs
@@ -119,6 +119,14 @@ impl DefaultRowWriter {
         }
     }
 
+    /// Creates a writer that reuses an existing (already-allocated) row buffer,
+    /// clearing it first. Lets batch fetchers recycle row allocations across a
+    /// result set instead of allocating one `Vec` per row.
+    pub fn from_recycled(mut row: Vec<ColumnValues>) -> Self {
+        row.clear();
+        Self { row }
+    }
+
     /// Takes the completed row, leaving the writer ready for reuse.
     pub fn take_row(&mut self) -> Vec<ColumnValues> {
         std::mem::take(&mut self.row)
diff --git a/mssql-tds/src/io.rs b/mssql-tds/src/io.rs
index 920d3dd0..302b404c 100644
--- a/mssql-tds/src/io.rs
+++ b/mssql-tds/src/io.rs
@@ -30,6 +30,7 @@ pub(crate) mod packet_buffer;
 pub mod packet_reader;
 pub mod packet_writer;
 pub mod reader_writer;
+pub(crate) mod std_byte_source;
 pub(crate) mod sync_token;
 pub(crate) mod tds_core;
 pub(crate) mod token_stream;
diff --git a/mssql-tds/src/io/blocking_reader.rs b/mssql-tds/src/io/blocking_reader.rs
index 01268c06..4f5ebea5 100644
--- a/mssql-tds/src/io/blocking_reader.rs
+++ b/mssql-tds/src/io/blocking_reader.rs
@@ -47,6 +47,37 @@ impl<S: BlockingByteSource> BlockingPacketReader<S> {
         }
     }
 
+    /// Creates a reader seeded with `residual` bytes carried over from an
+    /// async→blocking edge flip, so the first decode/refill continues exactly
+    /// where the async parser paused (see
+    /// [`PacketBuffer::seed_residual`](crate::io::packet_buffer::PacketBuffer::seed_residual)).
+    pub(crate) fn with_seeded_buffer(
+        source: S,
+        packet_size: usize,
+        residual: &crate::io::packet_buffer::ResidualBytes,
+    ) -> Self {
+        let mut buffer = PacketBuffer::with_packet_size(packet_size);
+        buffer.seed_residual(residual);
+        Self { source, buffer }
+    }
+
+    /// Drains the unconsumed bytes for handoff back across a blocking→async flip.
+    pub(crate) fn take_residual(&mut self) -> crate::io::packet_buffer::ResidualBytes {
+        self.buffer.take_residual()
+    }
+
+    /// Consumes the reader and returns its byte source (to recover the owned
+    /// socket on revert).
+    pub(crate) fn into_source(self) -> S {
+        self.source
+    }
+
+    /// Mutable access to the byte source, so the owning client can refresh the
+    /// per-request cancel/deadline policy before each fetch.
+    pub(crate) fn source_mut(&mut self) -> &mut S {
+        &mut self.source
+    }
+
     /// Reads one complete TDS packet synchronously and strips its 8-byte header,
     /// exposing the payload. The forward-progress guard lives in
     /// [`BlockingRowReader::refill_row_buffer_blocking`].
diff --git a/mssql-tds/src/io/packet_buffer.rs b/mssql-tds/src/io/packet_buffer.rs
index 35c23b43..728354d6 100644
--- a/mssql-tds/src/io/packet_buffer.rs
+++ b/mssql-tds/src/io/packet_buffer.rs
@@ -36,6 +36,25 @@ pub(crate) struct NeedBytes {
     pub(crate) shortfall: usize,
 }
 
+/// The unconsumed bytes carried by a [`PacketBuffer`] at an edge flip
+/// (async↔blocking), split into the two regions the framing body distinguishes.
+///
+/// `available` is already-stripped, ready-to-decode payload
+/// (`working_buffer[position..length]`); `pending` is raw, still-headered bytes
+/// of the *next* packet from a coalesced read
+/// (`working_buffer[pending_offset..+pending_bytes]`). Re-seeding both into a
+/// fresh buffer via [`PacketBuffer::seed_residual`] reproduces the exact state
+/// the parser would have continued from, so a flip mid-packet or mid-row-token
+/// is byte-identical to never flipping.
+#[cfg_attr(not(test), allow(dead_code))]
+#[derive(Debug, Default, Clone, PartialEq, Eq)]
+pub(crate) struct ResidualBytes {
+    /// Stripped payload bytes not yet consumed by the decoder.
+    pub(crate) available: Vec<u8>,
+    /// Raw (still-headered) surplus bytes belonging to the next packet.
+    pub(crate) pending: Vec<u8>,
+}
+
 /// Synchronous, I/O-free buffer of reassembled TDS packet payload bytes.
 pub(crate) struct PacketBuffer {
     working_buffer: Vec<u8>,
@@ -375,6 +394,61 @@ impl PacketBuffer {
         self.length = length;
     }
 
+    /// Extracts every unconsumed byte for handoff across an edge flip, leaving
+    /// the buffer drained.
+    ///
+    /// Splits into `available` (stripped payload `position..length`) and
+    /// `pending` (raw surplus of the next packet), the exact two regions
+    /// [`begin_refill`](Self::begin_refill) distinguishes, so
+    /// [`seed_residual`](Self::seed_residual) can rebuild an identical resume
+    /// state on the other edge.
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub(crate) fn take_residual(&mut self) -> ResidualBytes {
+        let available = self.working_buffer[self.position..self.length].to_vec();
+        let pending = if self.pending_bytes > 0 {
+            let start = self.pending_bytes_offset;
+            self.working_buffer[start..start + self.pending_bytes].to_vec()
+        } else {
+            Vec::new()
+        };
+        self.position = 0;
+        self.length = 0;
+        self.pending_bytes = 0;
+        self.pending_bytes_offset = 0;
+        ResidualBytes { available, pending }
+    }
+
+    /// Seeds this (freshly constructed) buffer with residual bytes taken from the
+    /// other edge, reproducing the `available` + `pending` layout so the next
+    /// decode and refill continue byte-identically. The buffer grows if the
+    /// residual exceeds its two-packet working span (a straddle spanning more
+    /// than the negotiated packet size).
+    #[cfg_attr(not(test), allow(dead_code))]
+    pub(crate) fn seed_residual(&mut self, residual: &ResidualBytes) {
+        let available_len = residual.available.len();
+        let pending_len = residual.pending.len();
+        let needed = available_len + pending_len;
+        // Keep a full packet of headroom past the residual so the next
+        // `begin_refill`/`refill_window` always has room to read a whole packet,
+        // even when a large straddle fills most of the two-packet span.
+        let required_capacity = needed + self.max_packet_size;
+        if self.working_buffer.len() < required_capacity {
+            self.working_buffer.resize(required_capacity, 0);
+        }
+        self.working_buffer[..available_len].copy_from_slice(&residual.available);
+        self.position = 0;
+        self.length = available_len;
+        if pending_len > 0 {
+            self.working_buffer[available_len..available_len + pending_len]
+                .copy_from_slice(&residual.pending);
+            self.pending_bytes = pending_len;
+            self.pending_bytes_offset = available_len;
+        } else {
+            self.pending_bytes = 0;
+            self.pending_bytes_offset = 0;
+        }
+    }
+
     /// Debug view of the raw bytes read for the packet at `base`.
     pub(crate) fn raw_packet(&self, base: usize, raw_len: usize) -> &[u8] {
         &self.working_buffer[base..base + raw_len]
diff --git a/mssql-tds/src/io/std_byte_source.rs b/mssql-tds/src/io/std_byte_source.rs
new file mode 100644
index 00000000..bb489ab8
--- /dev/null
+++ b/mssql-tds/src/io/std_byte_source.rs
@@ -0,0 +1,101 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+//! The blocking byte source over an owned `std::net::TcpStream`.
+//!
+//! [`StdTcpByteSource`] is the production implementor of [`BlockingByteSource`]:
+//! the synchronous fetch edge (`TdsSyncClient`) reads TDS packets straight off a
+//! blocking socket, with no reactor. It owns the R1 slice-poll edge policy — a
+//! bounded blocking read with an atomic cancel-check *between* slices — so
+//! timeout and cancellation ride the same receive edge the async
+//! [`ReceiveGuard`](crate::io::byte_source::ReceiveGuard) uses, at zero hot-path
+//! cost and without a running tokio runtime.
+
+use std::io::Read;
+use std::net::TcpStream;
+use std::time::{Duration, Instant};
+
+use tokio_util::sync::CancellationToken;
+
+use crate::core::TdsResult;
+use crate::error::Error::{OperationCancelledError, TimeoutError};
+use crate::error::TimeoutErrorType;
+use crate::io::byte_source::BlockingByteSource;
+
+/// Duration of one blocking read slice. A stalled read wakes at this cadence so
+/// the cancel token and request deadline are re-checked between slices; shorter
+/// slices tighten cancel latency at the cost of more wakeups.
+const SLICE_TIMEOUT: Duration = Duration::from_millis(100);
+
+/// A [`BlockingByteSource`] over an owned blocking [`TcpStream`].
+pub(crate) struct StdTcpByteSource {
+    stream: TcpStream,
+    /// Cooperative cancel shared with the owning client; observed between read
+    /// slices (a blocked `read` cannot itself be interrupted on Windows).
+    cancel: Option<CancellationToken>,
+    /// Absolute instant the in-flight fetch must complete by, refreshed by the
+    /// owning client before each request. `None` waits indefinitely.
+    deadline: Option<Instant>,
+}
+
+impl StdTcpByteSource {
+    /// Wraps an established blocking socket, arming the slice cadence so cancel
+    /// and deadline checks interleave with reads.
+    pub(crate) fn new(stream: TcpStream, cancel: Option<CancellationToken>) -> TdsResult<Self> {
+        stream.set_read_timeout(Some(SLICE_TIMEOUT))?;
+        Ok(Self {
+            stream,
+            cancel,
+            deadline: None,
+        })
+    }
+
+    /// Sets the absolute deadline for subsequent reads (the owning client derives
+    /// it from the per-request timeout before each fetch).
+    pub(crate) fn set_deadline(&mut self, deadline: Option<Instant>) {
+        self.deadline = deadline;
+    }
+
+    /// Consumes the source and returns the owned socket, so the client can revert
+    /// it to an async tokio stream.
+    pub(crate) fn into_stream(self) -> TcpStream {
+        self.stream
+    }
+}
+
+impl BlockingByteSource for StdTcpByteSource {
+    fn receive(&mut self, buffer: &mut [u8]) -> TdsResult<usize> {
+        loop {
+            if let Some(cancel) = &self.cancel
+                && cancel.is_cancelled()
+            {
+                return Err(OperationCancelledError(
+                    "blocking receive cancelled".to_string(),
+                ));
+            }
+            if let Some(deadline) = self.deadline
+                && Instant::now() >= deadline
+            {
+                return Err(TimeoutError(TimeoutErrorType::String(
+                    "blocking receive deadline elapsed".to_string(),
+                )));
+            }
+            match self.stream.read(buffer) {
+                Ok(n) => return Ok(n),
+                Err(e)
+                    if matches!(
+                        e.kind(),
+                        std::io::ErrorKind::WouldBlock
+                            | std::io::ErrorKind::TimedOut
+                            | std::io::ErrorKind::Interrupted
+                    ) =>
+                {
+                    // Slice expired (or a signal interrupted the read) without new
+                    // bytes; loop to re-check cancel/deadline, then read again.
+                    continue;
+                }
+                Err(e) => return Err(e.into()),
+            }
+        }
+    }
+}
diff --git a/mssql-tds/src/io/token_stream.rs b/mssql-tds/src/io/token_stream.rs
index bc963753..1dc289c8 100644
--- a/mssql-tds/src/io/token_stream.rs
+++ b/mssql-tds/src/io/token_stream.rs
@@ -7,6 +7,7 @@ use crate::datatypes::decoder::{
     decrypt_encrypted_column,
 };
 use crate::datatypes::row_writer::{DefaultRowWriter, RowWriter, write_column_value};
+use crate::datatypes::sql_string::{SqlString, get_encoding_type};
 use crate::datatypes::sqldatatypes::TdsDataType;
 use crate::datatypes::sync_decoder::{PlpProgress, plp_collect_step};
 use crate::io::packet_buffer::PacketBuffer;
@@ -960,14 +961,15 @@ fn resolve_header_token_blocking<R: BlockingRowReader>(
 /// Blocking sibling of [`decode_async_row_column`], restricted to the L3 scope.
 ///
 /// Reproduces the eager PLP arm of [`decode_async_row_column`] /
-/// `GenericDecoder::decode_into` for a non-encrypted `BigVarBinary` (varbinary
-/// (max)) cell: collect the value chunk-streamed via [`collect_plp_bytes_blocking`]
-/// then `write_bytes` / `write_null`, followed by the same post-column pause
-/// check. This is the single mirror of `decode_into`'s `BigVarBinary` arm — one
-/// arm, no new decode machine. Every other async-seam reason (encrypted cells,
-/// PLP strings, legacy LOBs, rare fallback types) is out of the L3 blocking scope
-/// and refuses via [`crate::error::Error::UnimplementedFeature`]; the differential
-/// corpus never drives those through this path.
+/// `GenericDecoder::decode_into` for non-encrypted PLP (`max`) cells:
+/// `varbinary(max)` collects chunk-streamed bytes via
+/// [`collect_plp_bytes_blocking`] then `write_bytes`, while `varchar(max)` /
+/// `nvarchar(max)` collect the same way and `write_string` through
+/// [`SqlString::new`] with the column's [`get_encoding_type`], mirroring
+/// `StringDecoder::decode_string_into`'s PLP arm. Every remaining async-seam
+/// reason (encrypted cells, non-PLP Text/NText LOBs, rare fallback types) stays
+/// out of the L3 blocking scope and refuses via
+/// [`crate::error::Error::UnimplementedFeature`].
 #[cfg_attr(not(test), allow(dead_code))]
 fn decode_blocking_async_column<R: BlockingRowReader>(
     reader: &mut R,
@@ -979,18 +981,32 @@ fn decode_blocking_async_column<R: BlockingRowReader>(
     let meta = &columns[col];
     let len = columns.len();
 
-    if meta.crypto_metadata.is_some() || meta.data_type != TdsDataType::BigVarBinary {
+    let plp_string = meta.is_plp()
+        && matches!(
+            meta.data_type,
+            TdsDataType::NVarChar
+                | TdsDataType::BigVarChar
+                | TdsDataType::NChar
+                | TdsDataType::BigChar
+        );
+
+    if meta.crypto_metadata.is_some()
+        || (meta.data_type != TdsDataType::BigVarBinary && !plp_string)
+    {
         return Err(crate::error::Error::UnimplementedFeature {
             feature: "blocking sync decode of an async-seam column".to_string(),
             context: format!(
                 "column '{}' ({:?}) is not in the L3 blocking scope (non-encrypted \
-                 varbinary(max) only)",
+                 varbinary(max) / varchar(max) / nvarchar(max) only)",
                 meta.column_name, meta.data_type
             ),
         });
     }
 
     match collect_plp_bytes_blocking(reader)? {
+        Some(bytes) if plp_string => {
+            writer.write_string(col, SqlString::new(bytes, get_encoding_type(meta)))
+        }
         Some(bytes) => writer.write_bytes(col, bytes),
         None => writer.write_null(col),
     }
diff --git a/mssql-tds/tests/test_sync_client.rs b/mssql-tds/tests/test_sync_client.rs
new file mode 100644
index 00000000..5404f0d7
--- /dev/null
+++ b/mssql-tds/tests/test_sync_client.rs
@@ -0,0 +1,528 @@
+// Copyright (c) Microsoft Corporation.
+// Licensed under the MIT License.
+
+//! Differential tests for the synchronous, reactor-free `TdsSyncClient`.
+//!
+//! Every sync fetch path is checked byte-identical against an all-async oracle
+//! run over the same mock result set, including the residual-byte straddle gate:
+//! flipping async->sync and sync->async mid-result-set (where buffered bytes
+//! straddle a packet, not a clean boundary) must reproduce the oracle exactly,
+//! and a clean-boundary flip (negative control) must also pass.
+//!
+//! The mock server runs on its own OS thread with its own runtime, so the sync
+//! client's blocking reads on the test thread never starve it — the same
+//! decoupling a real (remote) SQL peer provides.
+
+#[cfg(test)]
+mod sync_client_tests {
+    use std::sync::mpsc;
+    use std::thread::JoinHandle;
+
+    use mssql_mock_tds::query_response::{ColumnDefinition, ColumnValue, Row, SqlDataType};
+    use mssql_mock_tds::{MockTdsServer, QueryResponse};
+    use mssql_tds::connection::client_context::ClientContext;
+    use mssql_tds::connection::tds_client::{ResultSet, TdsClient};
+    use mssql_tds::connection::tds_sync_client::{SyncConversion, TdsSyncClient};
+    use mssql_tds::connection_provider::tds_connection_provider::TdsConnectionProvider;
+    use mssql_tds::core::{EncryptionOptions, EncryptionSetting};
+    use mssql_tds::datatypes::column_values::ColumnValues;
+    use mssql_tds::error::SqlInfoMessage;
+    use tokio::sync::oneshot;
+
+    const QUERY: &str = "SELECT ROWS";
+
+    /// A mock server bound on its own thread + runtime; shut down on drop.
+    struct TestServer {
+        addr: std::net::SocketAddr,
+        shutdown: Option<oneshot::Sender<()>>,
+        thread: Option<JoinHandle<()>>,
+    }
+
+    impl Drop for TestServer {
+        fn drop(&mut self) {
+            if let Some(tx) = self.shutdown.take() {
+                let _ = tx.send(());
+            }
+            if let Some(handle) = self.thread.take() {
+                let _ = handle.join();
+            }
+        }
+    }
+
+    /// Builds the (Int, NVarChar) result set both the oracle and the sync client
+    /// fetch. Varying string lengths make row byte-sizes non-uniform, so a
+    /// mid-result-set flip lands at an arbitrary offset inside the single
+    /// response packet (a genuine straddle, not a clean packet boundary).
+    fn make_response(row_count: usize) -> QueryResponse {
+        let columns = vec![
+            ColumnDefinition::new("id", SqlDataType::Int),
+            ColumnDefinition::new("label", SqlDataType::NVarChar),
+        ];
+        let rows = (0..row_count)
+            .map(|i| {
+                Row::new(vec![
+                    ColumnValue::Int(i as i32),
+                    ColumnValue::NVarChar(format!("row-{i}-{}", "x".repeat(i % 7))),
+                ])
+            })
+            .collect();
+        QueryResponse::new(columns, rows)
+    }
+
+    /// Same (Int, NVarChar) result set, but the server emits an ERROR token after
+    /// `after_rows` rows, then a terminal DONE — exercising the fetch-time
+    /// error/drain path on both edges.
+    fn make_error_response(row_count: usize, after_rows: usize) -> QueryResponse {
+        use mssql_mock_tds::query_response::{InfoMessage, MidStreamError};
+        make_response(row_count).with_error_after(MidStreamError {
+            after_rows,
+            number: 50_000,
+            state: 1,
+            severity: 16,
+            message: "mid-stream boom".to_string(),
+            // INFO emitted during the drain (after ERROR, before terminal DONE)
+            // so both drains must capture it byte-identically.
+            drain_info: vec![InfoMessage::new(50_001, 10, "post-error drain notice")],
+        })
+    }
+
+    fn start_server(response: QueryResponse) -> TestServer {
+        let (addr_tx, addr_rx) = mpsc::channel();
+        let (shutdown_tx, shutdown_rx) = oneshot::channel();
+        let thread = std::thread::spawn(move || {
+            let rt = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("server runtime builds");
+            rt.block_on(async move {
+                let server = MockTdsServer::new("127.0.0.1:0")
+                    .await
+                    .expect("mock server binds");
+                let addr = server.local_addr();
+                {
+                    let registry = server.query_registry();
+                    registry.lock().await.register(QUERY, response);
+                }
+                addr_tx.send(addr).expect("addr channel open");
+                let _ = server.run_with_shutdown(shutdown_rx).await;
+            });
+        });
+        let addr = addr_rx.recv().expect("server reports its address");
+        TestServer {
+            addr,
+            shutdown: Some(shutdown_tx),
+            thread: Some(thread),
+        }
+    }
+
+    async fn connect(addr: std::net::SocketAddr) -> TdsClient {
+        let datasource = format!("tcp:{},{}", addr.ip(), addr.port());
+        let mut context = ClientContext::default();
+        context.user_name = "sa".to_string();
+        context.password = "test-password".to_string();
+        context.database = "master".to_string();
+        context.encryption_options = EncryptionOptions {
+            mode: EncryptionSetting::PreferOff,
+            trust_server_certificate: true,
+            host_name_in_cert: None,
+            server_certificate: None,
+        };
+        TdsConnectionProvider {}
+            .create_client(context, &datasource, None)
+            .await
+            .expect("client connects to mock server")
+    }
+
+    /// The all-async oracle: every row fetched via `next_row().await`.
+    async fn async_oracle(addr: std::net::SocketAddr) -> Vec<Vec<ColumnValues>> {
+        let mut client = connect(addr).await;
+        client
+            .execute(QUERY.to_string(), ())
+            .await
+            .expect("oracle executes");
+        let mut rows = Vec::new();
+        if client.on_rows() {
+            while let Some(row) = client.next_row().await.expect("oracle next_row") {
+                rows.push(row);
+            }
+        }
+        client.close_query().await.expect("oracle closes query");
+        rows
+    }
+
+    async fn execute_then_sync(addr: std::net::SocketAddr) -> TdsSyncClient {
+        let mut client = connect(addr).await;
+        client
+            .execute(QUERY.to_string(), ())
+            .await
+            .expect("executes before flip");
+        match client.into_sync() {
+            SyncConversion::Converted(sync) => sync,
+            SyncConversion::NotEligible(_) => panic!("raw TCP transport must be sync-eligible"),
+            SyncConversion::Failed(err) => panic!("into_sync failed: {err:?}"),
+        }
+    }
+
+    /// Sync `next_row` reproduces the async oracle byte-identically.
+    #[tokio::test]
+    async fn differential_next_row_matches_async_oracle() {
+        let server = start_server(make_response(100));
+        let expected = Box::pin(async_oracle(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual = Vec::new();
+        while let Some(row) = sync.next_row().expect("sync next_row") {
+            actual.push(row);
+        }
+
+        assert_eq!(actual, expected);
+        assert!(!actual.is_empty());
+    }
+
+    /// Sync `fetch_rows_batch` (with spare-vec recycling) reproduces the oracle.
+    #[tokio::test]
+    async fn differential_fetch_rows_batch_matches_async_oracle() {
+        let server = start_server(make_response(100));
+        let expected = Box::pin(async_oracle(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual: Vec<Vec<ColumnValues>> = Vec::new();
+        // Seed a small pool so the first batch exercises the spare-recycling
+        // (`spare.pop()`) path; later batches fall back to fresh allocation.
+        let mut spare: Vec<Vec<ColumnValues>> = vec![Vec::with_capacity(2), Vec::with_capacity(2)];
+        loop {
+            let before = actual.len();
+            let fetched = sync
+                .fetch_rows_batch(&mut actual, std::mem::take(&mut spare), 32)
+                .expect("fetch_rows_batch");
+            assert_eq!(actual.len() - before, fetched);
+            if fetched == 0 {
+                break;
+            }
+        }
+
+        assert_eq!(actual, expected);
+    }
+
+    /// Straddle gate, async->sync: read some rows async (leaving residual mid
+    /// packet), flip, finish sync. The concatenation must match the oracle.
+    #[tokio::test]
+    async fn straddle_interleave_async_to_sync() {
+        let server = start_server(make_response(100));
+        let expected = Box::pin(async_oracle(server.addr)).await;
+
+        let mut client = Box::pin(connect(server.addr)).await;
+        client
+            .execute(QUERY.to_string(), ())
+            .await
+            .expect("executes");
+
+        let mut actual = Vec::new();
+        assert!(client.on_rows());
+        for _ in 0..30 {
+            let row = client
+                .next_row()
+                .await
+                .expect("async next_row")
+                .expect("row present before flip");
+            actual.push(row);
+        }
+
+        let mut sync = match client.into_sync() {
+            SyncConversion::Converted(sync) => sync,
+            other => panic!(
+                "expected Converted, got a different variant: {}",
+                variant(&other)
+            ),
+        };
+        while let Some(row) = sync.next_row().expect("sync next_row after flip") {
+            actual.push(row);
+        }
+
+        assert_eq!(actual, expected);
+    }
+
+    /// Straddle gate, sync->async: flip to sync, read some rows, revert to async,
+    /// finish async. Both residual handoffs must preserve the byte stream.
+    #[tokio::test]
+    async fn straddle_interleave_sync_to_async() {
+        let server = start_server(make_response(100));
+        let expected = Box::pin(async_oracle(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual = Vec::new();
+        for _ in 0..30 {
+            let row = sync
+                .next_row()
+                .expect("sync next_row")
+                .expect("row present before revert");
+            actual.push(row);
+        }
+
+        let mut client = sync.into_async().expect("into_async reverts");
+        while let Some(row) = client
+            .next_row()
+            .await
+            .expect("async next_row after revert")
+        {
+            actual.push(row);
+        }
+
+        assert_eq!(actual, expected);
+    }
+
+    /// Clean-boundary negative control: flip before reading any row, so the
+    /// residual begins at a clean token boundary (no straddle). Must also match.
+    #[tokio::test]
+    async fn clean_boundary_flip_matches_async_oracle() {
+        let server = start_server(make_response(100));
+        let expected = Box::pin(async_oracle(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual = Vec::new();
+        while let Some(row) = sync.next_row().expect("sync next_row") {
+            actual.push(row);
+        }
+
+        assert_eq!(actual, expected);
+    }
+
+    /// Reversibility: after a full sync fetch, `into_async` yields a working
+    /// async client that can run further control-plane work.
+    #[tokio::test]
+    async fn into_async_returns_a_working_client() {
+        let server = start_server(make_response(50));
+        let _ = Box::pin(async_oracle(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        while sync.next_row().expect("sync next_row").is_some() {}
+
+        let mut client = sync.into_async().expect("into_async reverts");
+        client
+            .close_query()
+            .await
+            .expect("close_query on reverted client");
+        client
+            .close_connection()
+            .await
+            .expect("close_connection on reverted client");
+    }
+
+    /// The publicly-observable connection state a fetch-time ERROR must leave
+    /// behind once its batch is drained to terminal DONE. Both shells project
+    /// this from the same two fields cleared by the shared `finalize_row_error`
+    /// (`current_metadata` -> `None`, `current_result_set_has_been_read_till_end`
+    /// -> `true`), so the sync drain must reproduce it byte-identically.
+    #[derive(Debug, PartialEq, Eq)]
+    struct TerminalState {
+        metadata_empty: bool,
+        maybe_has_unread_rows: bool,
+    }
+
+    /// The all-async error oracle: fetch rows until the mid-stream ERROR, then
+    /// capture the surfaced error's `Debug` form plus the post-drain terminal
+    /// state for a byte-identical compare.
+    async fn async_oracle_until_error(
+        addr: std::net::SocketAddr,
+    ) -> (
+        Vec<Vec<ColumnValues>>,
+        String,
+        TerminalState,
+        Vec<SqlInfoMessage>,
+    ) {
+        let mut client = connect(addr).await;
+        client
+            .execute(QUERY.to_string(), ())
+            .await
+            .expect("oracle executes");
+        let mut rows = Vec::new();
+        assert!(client.on_rows());
+        let err = loop {
+            match client.next_row().await {
+                Ok(Some(row)) => rows.push(row),
+                Ok(None) => panic!("expected a mid-stream error, got a clean end"),
+                Err(e) => break format!("{e:?}"),
+            }
+        };
+        let terminal = TerminalState {
+            metadata_empty: client.get_metadata().is_empty(),
+            maybe_has_unread_rows: client.maybe_has_unread_rows(),
+        };
+        let info = client.take_info_messages();
+        (rows, err, terminal, info)
+    }
+
+    /// Sync analog: fetch via `next_row` until the ERROR surfaces (driving the
+    /// sync blocking drain), returning the rows, the error's `Debug` form, the
+    /// post-drain terminal state, and the info captured during the drain.
+    fn sync_fetch_until_error(
+        sync: &mut TdsSyncClient,
+    ) -> (
+        Vec<Vec<ColumnValues>>,
+        String,
+        TerminalState,
+        Vec<SqlInfoMessage>,
+    ) {
+        let mut rows = Vec::new();
+        let err = loop {
+            match sync.next_row() {
+                Ok(Some(row)) => rows.push(row),
+                Ok(None) => panic!("expected a mid-stream error, got a clean end"),
+                Err(e) => break format!("{e:?}"),
+            }
+        };
+        let terminal = TerminalState {
+            metadata_empty: sync.get_metadata().is_empty(),
+            maybe_has_unread_rows: sync.maybe_has_unread_rows(),
+        };
+        let info = sync.take_info_messages();
+        (rows, err, terminal, info)
+    }
+
+    /// Differential error path (all sync): rows-before-error and the surfaced
+    /// error match the async oracle exactly, driving the sync ERROR drain.
+    #[tokio::test]
+    async fn differential_error_mid_fetch_matches_async_oracle() {
+        let server = start_server(make_error_response(100, 40));
+        let (expected_rows, expected_err, expected_terminal, expected_info) =
+            Box::pin(async_oracle_until_error(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let (actual_rows, actual_err, actual_terminal, actual_info) =
+            sync_fetch_until_error(&mut sync);
+
+        assert_eq!(actual_rows, expected_rows);
+        assert_eq!(actual_err, expected_err);
+        assert_eq!(actual_rows.len(), 40);
+        assert_eq!(actual_terminal, expected_terminal);
+        // The INFO emitted during the drain must be captured byte-identically by
+        // the sync blocking drain and the async `drain_stream`.
+        assert_eq!(actual_info, expected_info);
+        assert_eq!(actual_info.len(), 1);
+        // Pin the absolute terminal state, not just sync/async parity: the drain
+        // reached terminal DONE, so metadata is cleared and no rows remain.
+        assert!(expected_terminal.metadata_empty);
+        assert!(!expected_terminal.maybe_has_unread_rows);
+    }
+
+    /// Differential error path via `fetch_rows_batch`: the batched loop pushes the
+    /// rows-before-error, then propagates the drained ERROR, leaving the same
+    /// terminal state and captured INFO as the async oracle.
+    #[tokio::test]
+    async fn differential_fetch_rows_batch_error_matches_async_oracle() {
+        let server = start_server(make_error_response(100, 40));
+        let (expected_rows, expected_err, expected_terminal, expected_info) =
+            Box::pin(async_oracle_until_error(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual: Vec<Vec<ColumnValues>> = Vec::new();
+        let mut spare: Vec<Vec<ColumnValues>> = vec![Vec::with_capacity(2), Vec::with_capacity(2)];
+        let actual_err = loop {
+            match sync.fetch_rows_batch(&mut actual, std::mem::take(&mut spare), 32) {
+                Ok(0) => panic!("expected a mid-stream error, got a clean end"),
+                Ok(_) => continue,
+                Err(e) => break format!("{e:?}"),
+            }
+        };
+        let actual_terminal = TerminalState {
+            metadata_empty: sync.get_metadata().is_empty(),
+            maybe_has_unread_rows: sync.maybe_has_unread_rows(),
+        };
+        let actual_info = sync.take_info_messages();
+
+        assert_eq!(actual, expected_rows);
+        assert_eq!(actual.len(), 40);
+        assert_eq!(actual_err, expected_err);
+        assert_eq!(actual_terminal, expected_terminal);
+        assert_eq!(actual_info, expected_info);
+        assert!(actual_terminal.metadata_empty);
+        assert!(!actual_terminal.maybe_has_unread_rows);
+    }
+
+    /// Straddle error, async->sync: read rows async, flip mid-result-set, then the
+    /// ERROR surfaces on the sync edge (sync drain over transferred residual).
+    #[tokio::test]
+    async fn straddle_error_async_to_sync_matches_oracle() {
+        let server = start_server(make_error_response(100, 40));
+        let (expected_rows, expected_err, expected_terminal, expected_info) =
+            Box::pin(async_oracle_until_error(server.addr)).await;
+
+        let mut client = Box::pin(connect(server.addr)).await;
+        client
+            .execute(QUERY.to_string(), ())
+            .await
+            .expect("executes");
+        let mut actual = Vec::new();
+        assert!(client.on_rows());
+        for _ in 0..30 {
+            let row = client
+                .next_row()
+                .await
+                .expect("async next_row")
+                .expect("row present before flip");
+            actual.push(row);
+        }
+
+        let mut sync = match client.into_sync() {
+            SyncConversion::Converted(sync) => sync,
+            other => panic!("expected Converted, got {}", variant(&other)),
+        };
+        let (rest, actual_err, actual_terminal, actual_info) = sync_fetch_until_error(&mut sync);
+        actual.extend(rest);
+
+        assert_eq!(actual, expected_rows);
+        assert_eq!(actual_err, expected_err);
+        assert_eq!(actual_terminal, expected_terminal);
+        assert_eq!(actual_info, expected_info);
+        assert!(actual_terminal.metadata_empty);
+        assert!(!actual_terminal.maybe_has_unread_rows);
+    }
+
+    /// Straddle error, sync->async: read rows sync, revert mid-result-set, then the
+    /// ERROR surfaces on the async edge (async drain over transferred residual).
+    #[tokio::test]
+    async fn straddle_error_sync_to_async_matches_oracle() {
+        let server = start_server(make_error_response(100, 40));
+        let (expected_rows, expected_err, expected_terminal, expected_info) =
+            Box::pin(async_oracle_until_error(server.addr)).await;
+
+        let mut sync = Box::pin(execute_then_sync(server.addr)).await;
+        let mut actual = Vec::new();
+        for _ in 0..30 {
+            let row = sync
+                .next_row()
+                .expect("sync next_row")
+                .expect("row present before revert");
+            actual.push(row);
+        }
+
+        let mut client = sync.into_async().expect("into_async reverts");
+        let actual_err = loop {
+            match client.next_row().await {
+                Ok(Some(row)) => actual.push(row),
+                Ok(None) => panic!("expected a mid-stream error, got a clean end"),
+                Err(e) => break format!("{e:?}"),
+            }
+        };
+        let actual_terminal = TerminalState {
+            metadata_empty: client.get_metadata().is_empty(),
+            maybe_has_unread_rows: client.maybe_has_unread_rows(),
+        };
+        let actual_info = client.take_info_messages();
+
+        assert_eq!(actual, expected_rows);
+        assert_eq!(actual_err, expected_err);
+        assert_eq!(actual_terminal, expected_terminal);
+        assert_eq!(actual_info, expected_info);
+        assert!(actual_terminal.metadata_empty);
+        assert!(!actual_terminal.maybe_has_unread_rows);
+    }
+
+    fn variant(conversion: &SyncConversion) -> &'static str {
+        match conversion {
+            SyncConversion::Converted(_) => "Converted",
+            SyncConversion::NotEligible(_) => "NotEligible",
+            SyncConversion::Failed(_) => "Failed",
+        }
+    }
+}