diff --git a/dapr/ext/workflow/_durabletask/worker.py b/dapr/ext/workflow/_durabletask/worker.py
index bdc9ff48..f7d0c04c 100644
--- a/dapr/ext/workflow/_durabletask/worker.py
+++ b/dapr/ext/workflow/_durabletask/worker.py
@@ -18,7 +18,7 @@
 import threading
 import warnings
 from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timedelta
+from datetime import datetime, timedelta, timezone
 from threading import Event, Thread
 from types import GeneratorType
 from typing import Any, Generator, Iterator, Optional, Sequence, TypeVar, Union
@@ -1399,6 +1399,12 @@ def create_timer_internal(
         id = self.next_sequence_number()
         if isinstance(fire_at, timedelta):
             fire_at = self.current_utc_datetime + fire_at
+        # Normalize timezone-aware datetimes to naive UTC so they can be safely
+        # compared with current_utc_datetime (which is always naive UTC). A
+        # tzinfo whose utcoffset() is None is still naive and must be left as-is
+        # (calling astimezone() on it would raise ValueError).
+        if isinstance(fire_at, datetime) and fire_at.utcoffset() is not None:
+            fire_at = fire_at.astimezone(timezone.utc).replace(tzinfo=None)
         action = ph.new_create_timer_action(id, fire_at, origin=origin)
         self._pending_actions[id] = action
 
diff --git a/tests/ext/workflow/durabletask/test_orchestration_executor.py b/tests/ext/workflow/durabletask/test_orchestration_executor.py
index fa9d570f..fab9aa7c 100644
--- a/tests/ext/workflow/durabletask/test_orchestration_executor.py
+++ b/tests/ext/workflow/durabletask/test_orchestration_executor.py
@@ -155,6 +155,66 @@ def delay_orchestrator(ctx: task.OrchestrationContext, _):
     assert complete_action.result.value == '"done"'  # results are JSON-encoded
 
 
+def test_create_timer_with_timezone_aware_datetime():
+    """A timezone-aware fire_at must schedule the timer at the same absolute
+    instant as its naive-UTC equivalent."""
+
+    fire_at_aware = datetime(2030, 6, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=9)))
+    fire_at_utc = datetime(2030, 6, 1, 0, 0, 0)
+
+    def delay_orchestrator(ctx: task.OrchestrationContext, _):
+        yield ctx.create_timer(fire_at_aware)
+        return 'done'
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(delay_orchestrator)
+
+    start_time = datetime(2030, 1, 1, 12, 0, 0)
+    new_events = [
+        helpers.new_workflow_started_event(start_time),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+    ]
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER)
+    result = executor.execute(TEST_INSTANCE_ID, [], new_events)
+    actions = result.actions
+
+    assert actions is not None
+    assert len(actions) == 1
+    assert actions[0].HasField('createTimer')
+    assert actions[0].createTimer.fireAt.ToDatetime() == fire_at_utc
+
+
+def test_timer_fired_completion_with_timezone_aware_datetime():
+    """A timer created from a timezone-aware datetime must replay cleanly
+    against history recorded in naive UTC (as the sidecar stores it)."""
+
+    fire_at_aware = datetime(2030, 6, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=9)))
+    fire_at_utc = datetime(2030, 6, 1, 0, 0, 0)
+
+    def delay_orchestrator(ctx: task.OrchestrationContext, _):
+        yield ctx.create_timer(fire_at_aware)
+        return 'done'
+
+    registry = worker._Registry()
+    name = registry.add_orchestrator(delay_orchestrator)
+
+    start_time = datetime(2030, 1, 1, 12, 0, 0)
+    old_events = [
+        helpers.new_workflow_started_event(start_time),
+        helpers.new_execution_started_event(name, TEST_INSTANCE_ID, encoded_input=None),
+        helpers.new_timer_created_event(1, fire_at_utc),
+    ]
+    new_events = [helpers.new_timer_fired_event(1, fire_at_utc)]
+
+    executor = worker._OrchestrationExecutor(registry, TEST_LOGGER)
+    result = executor.execute(TEST_INSTANCE_ID, old_events, new_events)
+    actions = result.actions
+
+    complete_action = get_and_validate_single_complete_workflow_action(actions)
+    assert complete_action.workflowStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert complete_action.result.value == '"done"'  # results are JSON-encoded
+
+
 def test_schedule_activity_actions():
     """Test the actions output for the call_activity orchestrator method"""