diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py
index 25c7a4e76dfc..b9584d6fe1ff 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/batcher.py
@@ -20,7 +20,10 @@
 
 from google.api_core import exceptions as core_exceptions
 
-from google.cloud.bigtable.data.exceptions import MutationsExceptionGroup
+from google.cloud.bigtable.data.exceptions import (
+    MutationsExceptionGroup,
+    RetryExceptionGroup,
+)
 from google.cloud.bigtable.data.mutations import RowMutationEntry
 
 FLUSH_COUNT = 100  # after this many elements, send out the batch
@@ -137,8 +140,12 @@ def _close_batcher(self):
                 # return that error to the user. In standard execution paths,
                 # FailedMutationEntryError always has an Exception cause;
                 # defensively wrap in Unknown if __cause__ is None.
-                if error.__cause__ is not None:
-                    self._exceptions.put(error.__cause__)
+                cause = error.__cause__
+                if isinstance(cause, RetryExceptionGroup) and cause.exceptions:
+                    # Use the terminal exception from the retry chain.
+                    cause = cause.exceptions[-1]
+                if cause is not None:
+                    self._exceptions.put(cause)
                 else:
                     unknown_err = core_exceptions.Unknown(str(error))
                     unknown_err.__cause__ = error
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py
index b49a09555b79..907344cab409 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_async/mutations_batcher.py
@@ -605,11 +605,11 @@ async def _wait_for_batch_results(
             return []
         exceptions: list[FailedMutationEntryError] = []
         for task, batch in tasks:
-            if CrossSync.is_async:
-                # futures don't need to be awaited in sync mode
-                await task
             try:
-                exc_list = task.result()
+                if CrossSync.is_async:
+                    exc_list = await task
+                else:
+                    exc_list = task.result()
                 if exc_list:
                     # expect a list of FailedMutationEntryError objects
                     for exc in exc_list:
diff --git a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py
index 2bdce069e8e0..1cb4a3b88fee 100644
--- a/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py
+++ b/packages/google-cloud-bigtable/google/cloud/bigtable/data/_helpers.py
@@ -33,6 +33,7 @@
 from google.api_core import exceptions as core_exceptions
 from google.api_core import retry as retries
 from google.api_core.retry import RetryFailureReason, exponential_sleep_generator
+from google.protobuf import any_pb2, message
 from google.rpc import code_pb2, status_pb2
 
 from google.cloud.bigtable.data.exceptions import (
@@ -287,22 +288,61 @@ def _get_status(exc: Optional[Exception]) -> status_pb2.Status:
     Returns:
         status_pb2.Status: A Status proto object.
     """
-    if (
-        isinstance(exc, core_exceptions.GoogleAPICallError)
-        and exc.grpc_status_code is not None
-    ):
-        return status_pb2.Status(  # type: ignore[unreachable]
-            code=exc.grpc_status_code.value[0],
-            message=exc.message,
-            details=exc.details,
-        )
+    if isinstance(exc, core_exceptions.GoogleAPICallError):
+        grpc_code = exc.grpc_status_code
+        if grpc_code is not None:
+            try:
+                details: list[any_pb2.Any] = []
+                for detail in exc.details or ():
+                    try:
+                        if isinstance(detail, any_pb2.Any):
+                            details.append(detail)
+                        elif isinstance(detail, message.Message):
+                            packed = any_pb2.Any()
+                            packed.Pack(detail)
+                            details.append(packed)
+                        elif isinstance(getattr(detail, "_pb", None), message.Message):
+                            packed = any_pb2.Any()
+                            packed.Pack(detail._pb)
+                            details.append(packed)
+                    except Exception:
+                        continue
+                return status_pb2.Status(
+                    code=grpc_code.value[0],
+                    message=exc.message or "",
+                    details=details,
+                )
+            except Exception:
+                pass
 
     return status_pb2.Status(
         code=code_pb2.UNKNOWN,
-        message=str(exc) if exc else "An unknown error has occurred",
+        message=_exc_message(exc),
     )
 
 
+def _exc_message(exc: Optional[Exception]) -> str:
+    """
+    Helper function that renders an exception as a string, without raising.
+
+    ``GoogleAPICallError.__str__`` reads ``exc.details``, which raises if the
+    exception was built with a malformed ``details`` value, so fall back to the
+    raw message when that happens.
+
+    Args:
+        exc: The exception to render.
+    Returns:
+        str: A description of the exception.
+    """
+    if exc is None:
+        return "An unknown error has occurred"
+    try:
+        return str(exc)
+    except Exception:
+        raw_message = getattr(exc, "message", None)
+        return raw_message if isinstance(raw_message, str) else type(exc).__name__
+
+
 def _validate_timeouts(
     operation_timeout: float, attempt_timeout: float | None, allow_none: bool = False
 ):
@@ -337,11 +377,18 @@ def _get_error_type(
 
     Args:
       - call_code: Exception type or gRPC status code.
+    Raises:
+      - ValueError: if call_code is not a valid retryable gRPC error status code
+        or exception type.
     """
     if isinstance(call_code, type):
         return call_code
-    else:
-        return type(core_exceptions.from_grpc_status(call_code, ""))
+    if isinstance(call_code, bool):
+        raise ValueError(f"Invalid gRPC error status code: {call_code!r}")
+    error_type = type(core_exceptions.from_grpc_status(call_code, ""))
+    if error_type is core_exceptions.GoogleAPICallError:
+        raise ValueError(f"Invalid gRPC error status code: {call_code!r}")
+    return error_type
 
 
 def _get_retryable_errors(
diff --git a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py
index a6b7194cfbbb..71739c769921 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_async/test_mutations_batcher.py
@@ -1106,6 +1106,55 @@ async def test__execute_mutate_rows_batch_completed_callback_coroutine(self):
                 else:
                     assert called_with == []
 
+    @CrossSync.pytest
+    async def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self):
+        from google.rpc import code_pb2, status_pb2
+
+        with mock.patch.object(CrossSync._MutateRowsOperation, "start") as mutate_rows:
+            mutate_rows.side_effect = RuntimeError("unexpected error")
+            table = mock.Mock()
+            table.default_mutate_rows_operation_timeout = 17
+            table.default_mutate_rows_attempt_timeout = 13
+            table.default_mutate_rows_retryable_errors = ()
+            callback = mock.Mock()
+            async with self._make_one(table) as instance:
+                instance._user_batch_completed_callback = callback
+                batch = [self._make_mutation(), self._make_mutation()]
+                with pytest.raises(RuntimeError, match="unexpected error"):
+                    await instance._execute_mutate_rows(batch, mock.Mock())
+                callback.assert_called_once_with(
+                    [
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                    ]
+                )
+
+    @CrossSync.pytest
+    async def test__wait_for_batch_results_unexpected_task_exception(self):
+        """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError."""
+        batch = [self._make_mutation(), self._make_mutation()]
+        expected_err = RuntimeError("task failed")
+
+        async with self._make_one() as instance:
+            with mock.patch.object(
+                instance, "_execute_mutate_rows", side_effect=expected_err
+            ):
+                with mock.patch.object(
+                    instance._flow_control, "add_to_flow"
+                ) as flow_mock:
+
+                    async def gen(x):
+                        yield x
+
+                    flow_mock.side_effect = lambda x: gen(x)
+                    await instance._flush_internal(batch)
+                    assert len(instance._oldest_exceptions) == 2
+                    for failed_entry in instance._oldest_exceptions:
+                        assert failed_entry.index is None
+                        assert failed_entry.__cause__ is expected_err
+                    instance._oldest_exceptions.clear()
+                    instance._newest_exceptions.clear()
+
     @CrossSync.pytest
     async def test__raise_exceptions(self):
         """Raise exceptions and reset error state"""
diff --git a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py
index 339323e30514..58b3547f36d1 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/_sync_autogen/test_mutations_batcher.py
@@ -983,6 +983,54 @@ def test__execute_mutate_rows_batch_completed_callback_coroutine(self):
                 assert result == []
                 assert called_with == []
 
+    def test__execute_mutate_rows_batch_completed_callback_unexpected_error(self):
+        from google.rpc import code_pb2, status_pb2
+
+        with mock.patch.object(
+            CrossSync._Sync_Impl._MutateRowsOperation, "start"
+        ) as mutate_rows:
+            mutate_rows.side_effect = RuntimeError("unexpected error")
+            table = mock.Mock()
+            table.default_mutate_rows_operation_timeout = 17
+            table.default_mutate_rows_attempt_timeout = 13
+            table.default_mutate_rows_retryable_errors = ()
+            callback = mock.Mock()
+            with self._make_one(table) as instance:
+                instance._user_batch_completed_callback = callback
+                batch = [self._make_mutation(), self._make_mutation()]
+                with pytest.raises(RuntimeError, match="unexpected error"):
+                    instance._execute_mutate_rows(batch, mock.Mock())
+                callback.assert_called_once_with(
+                    [
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                        status_pb2.Status(code=code_pb2.UNKNOWN),
+                    ]
+                )
+
+    def test__wait_for_batch_results_unexpected_task_exception(self):
+        """Unexpected exceptions raised by batch tasks should be wrapped in FailedMutationEntryError."""
+        batch = [self._make_mutation(), self._make_mutation()]
+        expected_err = RuntimeError("task failed")
+        with self._make_one() as instance:
+            with mock.patch.object(
+                instance, "_execute_mutate_rows", side_effect=expected_err
+            ):
+                with mock.patch.object(
+                    instance._flow_control, "add_to_flow"
+                ) as flow_mock:
+
+                    def gen(x):
+                        yield x
+
+                    flow_mock.side_effect = lambda x: gen(x)
+                    instance._flush_internal(batch)
+                    assert len(instance._oldest_exceptions) == 2
+                    for failed_entry in instance._oldest_exceptions:
+                        assert failed_entry.index is None
+                        assert failed_entry.__cause__ is expected_err
+                    instance._oldest_exceptions.clear()
+                    instance._newest_exceptions.clear()
+
     def test__raise_exceptions(self):
         """Raise exceptions and reset error state"""
         from google.cloud.bigtable.data import exceptions
diff --git a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py
index 3e44ec7c3d04..418fe4e28c16 100644
--- a/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py
+++ b/packages/google-cloud-bigtable/tests/unit/data/test__helpers.py
@@ -16,7 +16,8 @@
 import mock
 import pytest
 from google.api_core import exceptions as core_exceptions
-from google.rpc import code_pb2, status_pb2
+from google.protobuf import any_pb2
+from google.rpc import code_pb2, error_details_pb2, status_pb2
 
 import google.cloud.bigtable.data._helpers as _helpers
 import google.cloud.bigtable.data.exceptions as bt_exceptions
@@ -351,6 +352,72 @@ def test_get_statuses_from_mutations_exception_group(
         )
         assert statuses[0] == expected_status
 
+    def test_get_status_packs_unpacked_error_details(self):
+        err_info = error_details_pb2.ErrorInfo(
+            reason="IAM_PERMISSION_DENIED",
+            domain="bigtable.googleapis.com",
+            metadata={"resource": "projects/p/instances/i/tables/t"},
+        )
+        retry_info = error_details_pb2.RetryInfo()
+        already_packed = any_pb2.Any()
+        already_packed.Pack(retry_info)
+
+        proto_plus_wrapper = mock.Mock()
+        proto_plus_wrapper._pb = error_details_pb2.DebugInfo(detail="stack")
+
+        exc = core_exceptions.PermissionDenied(
+            "Permission denied",
+            details=[err_info, already_packed, proto_plus_wrapper, "ignored"],
+        )
+
+        status = _helpers._get_status(exc)
+        assert status.code == code_pb2.PERMISSION_DENIED
+        assert status.message == "Permission denied"
+        assert len(status.details) == 3
+
+        unpacked_err_info = error_details_pb2.ErrorInfo()
+        assert status.details[0].Unpack(unpacked_err_info)
+        assert unpacked_err_info == err_info
+
+        assert status.details[1] == already_packed
+
+        unpacked_debug_info = error_details_pb2.DebugInfo()
+        assert status.details[2].Unpack(unpacked_debug_info)
+        assert unpacked_debug_info == proto_plus_wrapper._pb
+
+    def test_get_status_handles_malformed_details_and_exceptions(self):
+        err_info = error_details_pb2.ErrorInfo(
+            reason="IAM_PERMISSION_DENIED",
+            domain="bigtable.googleapis.com",
+        )
+
+        class BrokenWrapper:
+            @property
+            def _pb(self):
+                raise RuntimeError("broken _pb property")
+
+        exc = core_exceptions.PermissionDenied(
+            None,
+            details=[BrokenWrapper(), err_info],
+        )
+        status = _helpers._get_status(exc)
+        assert status.code == code_pb2.PERMISSION_DENIED
+        assert status.message == ""
+        assert len(status.details) == 1
+
+        unpacked_err_info = error_details_pb2.ErrorInfo()
+        assert status.details[0].Unpack(unpacked_err_info)
+        assert unpacked_err_info == err_info
+
+        # Non-iterable details falls back to UNKNOWN without raising
+        malformed_exc = core_exceptions.PermissionDenied(
+            "Permission denied",
+            details=123,  # type: ignore[arg-type]
+        )
+        fallback_status = _helpers._get_status(malformed_exc)
+        assert fallback_status.code == code_pb2.UNKNOWN
+        assert "Permission denied" in fallback_status.message
+
 
 class TestGetRetryableErrors:
     @pytest.mark.parametrize(
@@ -397,6 +464,24 @@ def test_get_retryable_errors(self, input_codes, input_table, expected):
         result = _helpers._get_retryable_errors(input_codes, fake_table)
         assert result == expected
 
+    @pytest.mark.parametrize(
+        "invalid_code",
+        [
+            0,
+            grpc.StatusCode.OK,
+            -1,
+            999,
+            True,
+            False,
+            "INVALID",
+        ],
+    )
+    def test_get_error_type_invalid_status_code(self, invalid_code):
+        with pytest.raises(ValueError, match="Invalid gRPC error status code"):
+            _helpers._get_error_type(invalid_code)
+        with pytest.raises(ValueError, match="Invalid gRPC error status code"):
+            _helpers._get_retryable_errors([invalid_code], mock.Mock())
+
 
 class TestTrackedBackoffGenerator:
     def test_tracked_backoff_generator_history(self):