@@ -278,6 +278,40 @@ def _get_statuses_from_mutations_exception_group(
     return statuses
 
 
+def _get_status_from_operation_error(exc: Exception) -> status_pb2.Status:
+    """
+    Helper function that returns a Status object for an exception raised by a
+    single-row operation, used by the legacy synchronous shim.
+
+    When an operation runs out of time, :func:`_retry_exception_factory` raises a
+    synthetic ``DeadlineExceeded`` and chains the errors actually seen on the wire
+    as a ``RetryExceptionGroup`` cause. Reporting the synthetic wrapper would mask
+    the real failure, so unwrap it and report the terminal error instead, matching
+    what :func:`_get_statuses_from_mutations_exception_group` reports for the
+    equivalent bulk operation.
+
+    The unwrapping is deliberately limited to that case. When retries stop for any
+    other reason the primary exception *is* the terminal error and the group holds
+    only the earlier attempts, so unwrapping would report a stale error. The
+    synthetic exception is told apart by having no underlying response: errors
+    returned by the server are built from a grpc error and carry one.
+
+    Args:
+        exc: An exception raised by a single-row operation.
+    Returns:
+        status_pb2.Status: A Status proto object.
+    """
+    cause = exc.__cause__
+    if (
+        isinstance(exc, core_exceptions.DeadlineExceeded)
+        and exc.response is None
+        and isinstance(cause, RetryExceptionGroup)
+        and cause.exceptions
+    ):
+        return _get_status(cause.exceptions[-1])
+    return _get_status(exc)
+
+
 def _get_status(exc: Optional[Exception]) -> status_pb2.Status:
     """
     Helper function that returns a Status object corresponding to the given exception.
@@ -287,12 +321,15 @@ 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],
+    if isinstance(exc, core_exceptions.GoogleAPICallError):
+        return status_pb2.Status(
+            code=(
+                exc.grpc_status_code.value[0]
+                if exc.grpc_status_code is not None
+                else code_pb2.UNKNOWN
+            ),
+            # str() on an api call error prepends its code, which is the thing
+            # that is missing here, so report the message on its own.
             message=exc.message,
             details=exc.details,
         )
 
@@ -24,6 +24,7 @@
 
 from google.cloud.bigtable.data import mutations
 from google.cloud.bigtable.data import read_modify_write_rules as rmw_rules
+from google.cloud.bigtable.data._helpers import _get_status_from_operation_error
 
 MAX_MUTATIONS = 100000
 """The maximum number of mutations that a row can accumulate."""
@@ -451,7 +452,10 @@ def commit(self):
         in the row are left unchanged unless explicitly changed by a mutation.
 
         After committing the accumulated mutations, resets the local
-        mutations to an empty list.
+        mutations to an empty list. This happens regardless of the outcome:
+        the mutations are also discarded when the request fails, so a failed
+        :meth:`commit` cannot be retried by calling :meth:`commit` again on
+        the same row. Re-add the mutations to retry them.
 
         For example:
 
@@ -478,13 +482,7 @@ def commit(self):
             return status_pb2.Status(code=code_pb2.OK)
         except GoogleAPICallError as e:
             # If the RPC call returns an error, extract the error into a status object, if possible.
-            return status_pb2.Status(
-                code=e.grpc_status_code.value[0]
-                if e.grpc_status_code is not None
-                else code_pb2.UNKNOWN,
-                message=e.message,
-                details=getattr(e, "details", []),
-            )
+            return _get_status_from_operation_error(e)
         finally:
             self.clear()
 
 
@@ -479,6 +479,80 @@ def test_direct_row_commit_with_unknown_exception():
     assert app_profile_id == call_args.app_profile_id[0]
 
 
+def _commit_with_operation_error(exc):
+    """Commit a row whose underlying ``mutate_row`` fails with ``exc``."""
+    credentials = _make_credentials()
+    client = _make_client(project="project-id", credentials=credentials, admin=True)
+    table = _Table("projects/more-stuff", client=client, app_profile_id=None)
+    row = _make_direct_row(b"row_key", table)
+    row.set_cell("column_family_id", b"column", b"bytes-value")
+
+    with mock.patch.object(table._table_impl, "mutate_row") as mutate_row_mock:
+        mutate_row_mock.side_effect = exc
+        return row.commit()
+
+
+def test_direct_row_commit_unwraps_retry_exception_group():
+    """On timeout the data client raises a synthetic DeadlineExceeded chained
+    with the errors seen on the wire. Report the terminal wire error, so that a
+    single-row commit agrees with the equivalent bulk mutate_rows."""
+    from google.api_core.exceptions import ServiceUnavailable
+    from google.api_core.retry import RetryFailureReason
+    from google.rpc import code_pb2
+
+    from google.cloud.bigtable.data._helpers import _retry_exception_factory
+
+    source_exc, _ = _retry_exception_factory(
+        [ServiceUnavailable("first failure"), ServiceUnavailable("final failure")],
+        RetryFailureReason.TIMEOUT,
+        60.0,
+    )
+
+    result = _commit_with_operation_error(source_exc)
+
+    assert result.code == code_pb2.Code.UNAVAILABLE
+    assert result.message == "final failure"
+
+
+def test_direct_row_commit_reports_terminal_error_when_not_timed_out():
+    """When retries stop for any other reason the primary exception already is
+    the terminal error and the chained group holds only earlier attempts, so it
+    must not be unwrapped."""
+    from google.api_core.exceptions import PermissionDenied, ServiceUnavailable
+    from google.api_core.retry import RetryFailureReason
+    from google.rpc import code_pb2
+
+    from google.cloud.bigtable.data._helpers import _retry_exception_factory
+
+    source_exc, _ = _retry_exception_factory(
+        [ServiceUnavailable("earlier attempt"), PermissionDenied("terminal failure")],
+        RetryFailureReason.NON_RETRYABLE_ERROR,
+        60.0,
+    )
+
+    result = _commit_with_operation_error(source_exc)
+
+    assert result.code == code_pb2.Code.PERMISSION_DENIED
+    assert result.message == "terminal failure"
+
+
+def test_direct_row_commit_keeps_deadline_exceeded_returned_by_server():
+    """A DeadlineExceeded that came back from the server is the terminal error
+    and is reported as-is, unlike the locally built one."""
+    from google.api_core.exceptions import DeadlineExceeded, ServiceUnavailable
+    from google.rpc import code_pb2
+
+    from google.cloud.bigtable.data.exceptions import RetryExceptionGroup
+
+    exc = DeadlineExceeded("server deadline exceeded", response=mock.Mock())
+    exc.__cause__ = RetryExceptionGroup([ServiceUnavailable("earlier attempt")])
+
+    result = _commit_with_operation_error(exc)
+
+    assert result.code == code_pb2.Code.DEADLINE_EXCEEDED
+    assert result.message == "server deadline exceeded"
+
+
 def test_direct_row_commit_with_invalid_argument():
     from google.cloud.bigtable_v2.services.bigtable import BigtableClient