diff --git a/.changelog/4901.fixed b/.changelog/4901.fixed
new file mode 100644
index 0000000000..1670453d07
--- /dev/null
+++ b/.changelog/4901.fixed
@@ -0,0 +1 @@
+`opentelemetry-instrumentation-asyncio`: fix `to_thread` instrumentation so spans and duration metrics measure the actual function execution instead of ending immediately, and record the `exception` state when the function raises
diff --git a/instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py b/instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py
index 4acf422242..9aa6d1ca87 100644
--- a/instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py
+++ b/instrumentation/opentelemetry-instrumentation-asyncio/src/opentelemetry/instrumentation/asyncio/__init__.py
@@ -224,30 +224,37 @@ def wrap_taskgroup_create_task(method, instance, args, kwargs) -> None:
 
     def trace_to_thread(self, func: callable):
         """
-        Trace a function, but if already instrumented, skip double-wrapping.
+        Wrap a function so that its execution in the worker thread is
+        measured and, if enabled, traced.
         """
-        if _is_instrumented(func):
-            return func
-
-        start = default_timer()
         func_name = getattr(func, "__name__", None)
         if func_name is None and isinstance(func, functools.partial):
             func_name = func.func.__name__
-        span = (
-            self._tracer.start_span(f"{ASYNCIO_PREFIX} to_thread-" + func_name)
-            if func_name in self._to_thread_name_to_trace
-            else None
-        )
-        attr = {"type": "to_thread", "name": func_name}
-        exception = None
-        try:
-            attr["state"] = "finished"
-            return func
-        except Exception:
-            attr["state"] = "exception"
-            raise
-        finally:
-            self.record_process(start, attr, span, exception)
+
+        @functools.wraps(func)
+        def wrapper(*args, **kwargs):
+            start = default_timer()
+            span = (
+                self._tracer.start_span(
+                    f"{ASYNCIO_PREFIX} to_thread-" + func_name
+                )
+                if func_name in self._to_thread_name_to_trace
+                else None
+            )
+            attr = {"type": "to_thread", "name": func_name}
+            exception = None
+            try:
+                result = func(*args, **kwargs)
+                attr["state"] = "finished"
+                return result
+            except Exception as exc:
+                exception = exc
+                attr["state"] = "exception"
+                raise
+            finally:
+                self.record_process(start, attr, span, exception)
+
+        return wrapper
 
     def trace_item(self, coro_or_future):
         """Trace a coroutine or future item."""
diff --git a/instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_to_thread.py b/instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_to_thread.py
index 3e6d52cf97..b319cbf642 100644
--- a/instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_to_thread.py
+++ b/instrumentation/opentelemetry-instrumentation-asyncio/tests/test_asyncio_to_thread.py
@@ -2,6 +2,7 @@
 # SPDX-License-Identifier: Apache-2.0
 import asyncio
 import functools
+import time
 from unittest.mock import patch
 
 # pylint: disable=no-name-in-module
@@ -10,7 +11,7 @@
     OTEL_PYTHON_ASYNCIO_TO_THREAD_FUNCTION_NAMES_TO_TRACE,
 )
 from opentelemetry.test.test_base import TestBase
-from opentelemetry.trace import get_tracer
+from opentelemetry.trace import StatusCode, get_tracer
 
 
 class TestAsyncioToThread(TestBase):
@@ -58,6 +59,83 @@ async def to_thread():
                     self.assertEqual(point.attributes["type"], "to_thread")
                     self.assertEqual(point.attributes["name"], "multiply")
 
+    def test_to_thread_duration_covers_execution(self):
+        def multiply(x, y):
+            time.sleep(0.1)
+            return x * y
+
+        async def to_thread():
+            result = await asyncio.to_thread(multiply, 2, 3)
+            assert result == 6
+
+        asyncio.run(to_thread())
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertGreaterEqual(span.end_time - span.start_time, 0.1 * 10**9)
+
+        for metric in (
+            self.memory_metrics_reader.get_metrics_data()
+            .resource_metrics[0]
+            .scope_metrics[0]
+            .metrics
+        ):
+            if metric.name == "asyncio.process.duration":
+                for point in metric.data.data_points:
+                    self.assertGreaterEqual(point.sum, 0.1)
+
+    def test_to_thread_exception(self):
+        def multiply(x, y):
+            raise ValueError("fail")
+
+        async def to_thread():
+            await asyncio.to_thread(multiply, 2, 3)
+
+        with self.assertRaises(ValueError):
+            asyncio.run(to_thread())
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 1)
+        span = spans[0]
+        self.assertEqual(span.name, "asyncio to_thread-multiply")
+        self.assertEqual(span.status.status_code, StatusCode.ERROR)
+        self.assertEqual(len(span.events), 1)
+        self.assertEqual(span.events[0].name, "exception")
+
+        for metric in (
+            self.memory_metrics_reader.get_metrics_data()
+            .resource_metrics[0]
+            .scope_metrics[0]
+            .metrics
+        ):
+            if metric.name == "asyncio.process.duration":
+                for point in metric.data.data_points:
+                    self.assertEqual(point.attributes["state"], "exception")
+
+    def test_to_thread_repeated_calls(self):
+        def multiply(x, y):
+            return x * y
+
+        async def to_thread():
+            assert await asyncio.to_thread(multiply, 2, 3) == 6
+            assert await asyncio.to_thread(multiply, 4, 5) == 20
+
+        asyncio.run(to_thread())
+
+        spans = self.memory_exporter.get_finished_spans()
+        self.assertEqual(len(spans), 2)
+
+        for metric in (
+            self.memory_metrics_reader.get_metrics_data()
+            .resource_metrics[0]
+            .scope_metrics[0]
+            .metrics
+        ):
+            if metric.name == "asyncio.process.created":
+                for point in metric.data.data_points:
+                    self.assertEqual(point.value, 2)
+
     def test_to_thread_partial_func(self):
         def multiply(x, y):
             return x * y