@@ -9,10 +9,6 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 FIXED
 
-- Async sub-orchestration calls without an explicit `return_type` now expose
-`Any` rather than promising the child function's annotated result type.
-Custom result reconstruction still requires `return_type`; runtime decoding is
-unchanged.
 - Activity and sub-orchestration `retry_timeout` decisions now use replay-safe
 orchestration time, preventing retries from expiring solely because history is
 replayed later. Retry calculations also accept timezone-aware timestamps.
 
@@ -100,6 +100,8 @@ allocate action IDs, serialize input, or consume buffered events. Cancelling its
 its first step starts no durable operation. Operation inputs are not copied:
 mutations made before the coroutine's first step are observed, and relative
 timer delays start from that execution point. Keep or copy inputs accordingly.
+Dropping an operation without awaiting it or wrapping it in a task schedules
+nothing, and Python reports that the coroutine was never awaited.
 
 Cold coroutines are single-use, like ordinary Python coroutines. For an
 explicitly started operation with multiple independent waiters, retain its
@@ -118,6 +120,27 @@ Cancelling either borrower does not cancel the underlying operation. Direct
 an awaitable durable handle is not a coroutine. Do not retry that expression
 assuming the eager operation was rolled back.
 
+To identify the winner of a race, check whether an operation has completed, or
+read its result later, use eager handles with `task.when_any`, or wrap cold
+operations in `asyncio` tasks:
+
+```python
+eager = ctx.as_task_context()
+approval = eager.wait_for_external_event("approval", data_type=bool)
+deadline = eager.create_timer(timedelta(hours=1))
+winner = await task.when_any([approval, deadline])  # The identical handle wins.
+if winner is approval:
+    deadline.cancel()
+    approved = approval.get_result()
+else:
+    report = asyncio.create_task(ctx.call_activity("report", return_type=str))
+    finished = report.done()  # Completion check without waiting.
+    summary = await report  # Result, or the operation's failure.
+```
+
+The losing handle stays inspectable through `is_complete` and, once complete,
+`get_result()`.
+
 The worker classifies each orchestrator once, when it is registered:
 
 1. A callable that is itself a coroutine function (including partials, bound
 
@@ -230,6 +230,8 @@ async def create_timer(self, fire_at: datetime | timedelta) -> None:
         coroutine is constructed. Cancellation removes uncommitted timer work
         where possible; it cannot retract a timer already present in history.
         """
+        if isinstance(fire_at, timedelta) and not math.isfinite(fire_at.total_seconds()):
+            raise ValueError("Timer delays must be finite.")
         timer = self._context.create_timer(fire_at)
         try:
             await timer
 
@@ -19,6 +19,7 @@
 from durabletask.internal import helpers
 from durabletask.internal import orchestrator_service_pb2 as pb
 from tests.durabletask.test_coroutine_orchestrators import (
+    INSTANCE_ID,
     NOW,
     _completion,  # pyright: ignore[reportPrivateUsage]
     _execute,  # pyright: ignore[reportPrivateUsage]
@@ -32,6 +33,56 @@ def test_additive_public_api_has_no_new_abstract_requirements() -> None:
     assert task.AsyncOrchestrationContext is AsyncOrchestrationContext
 
 
+# Intentional differences between the contexts; anything else is drift.
+_ASYNC_ONLY_MEMBERS = {"as_task_context"}
+_EAGER_ONLY_MEMBERS: set[str] = set()
+_ASYNC_OPERATIONS = {
+    "call_activity", "call_sub_orchestrator", "call_entity", "lock_entities",
+    "create_timer", "wait_for_external_event",
+}
+_EXTRA_ASYNC_PARAMETERS = {"wait_for_external_event": {"timeout"}}
+
+
+def _public_members(cls: type) -> set[str]:
+    return {name for name in dir(cls) if not name.startswith("_")}
+
+
+def test_async_context_public_api_matches_task_context() -> None:
+    async_members = _public_members(AsyncOrchestrationContext)
+    eager_members = _public_members(task.OrchestrationContext)
+    assert async_members - eager_members == _ASYNC_ONLY_MEMBERS
+    assert eager_members - async_members == _EAGER_ONLY_MEMBERS
+
+    for name in sorted(async_members & eager_members):
+        async_member = inspect.getattr_static(AsyncOrchestrationContext, name)
+        eager_member = inspect.getattr_static(task.OrchestrationContext, name)
+        assert isinstance(async_member, property) == isinstance(eager_member, property), name
+        if isinstance(async_member, property):
+            continue
+        assert inspect.iscoroutinefunction(async_member) == (name in _ASYNC_OPERATIONS), name
+        async_parameters = set(inspect.signature(async_member).parameters)
+        eager_parameters = set(inspect.signature(eager_member).parameters)
+        assert async_parameters - eager_parameters == _EXTRA_ASYNC_PARAMETERS.get(name, set()), name
+        assert eager_parameters <= async_parameters, name
+
+
+def test_async_context_metadata_matches_task_context() -> None:
+    observed: list[tuple[Any, Any]] = []
+
+    async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> None:
+        eager = ctx.as_task_context()
+        for name in ("instance_id", "parent_instance_id", "version", "current_utc_datetime", "is_replaying"):
+            observed.append((getattr(ctx, name), getattr(eager, name)))
+        await ctx.call_activity("work")
+        observed.append((ctx.is_replaying, eager.is_replaying))
+
+    old = _start() + [helpers.new_task_scheduled_event(1, "work")]
+    _execute(orchestrator, old, [helpers.new_task_completed_event(1, "null")])
+    assert observed and all(async_value == eager_value for async_value, eager_value in observed)
+    assert observed[0][0] == INSTANCE_ID
+    assert observed[4][0] is True and observed[-1][0] is False
+
+
 def test_public_async_context_annotations_resolve_at_runtime() -> None:
     assert get_type_hints(AsyncOrchestrationContext.call_activity)["return"] is Any
     assert get_type_hints(AsyncOrchestrationContext.call_sub_orchestrator)["return"] is Any
@@ -348,6 +399,63 @@ async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> int:
     assert json.loads(_completion(result).result.value) == 42
 
 
+@pytest.mark.parametrize("seconds", [float("inf"), float("-inf"), float("nan")])
+def test_nonfinite_timer_delay_rejected_before_scheduling(seconds: float) -> None:
+    class InfiniteDelay(timedelta):
+        def total_seconds(self) -> float:
+            return seconds
+
+    async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> str:
+        with pytest.raises(ValueError, match="finite"):
+            await ctx.create_timer(InfiniteDelay())
+        return await ctx.call_activity("work", return_type=str)
+
+    actions = _execute(orchestrator, [], _start()).actions
+    assert [(action.id, action.scheduleTask.name) for action in actions] == [(1, "work")]
+
+
+@pytest.mark.parametrize("approved_first", [True, False])
+def test_winner_completion_and_result_idioms(approved_first: bool) -> None:
+    observed: list[Any] = []
+
+    async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> str:
+        eager = ctx.as_task_context()
+        approval = eager.wait_for_external_event("approval", data_type=bool)
+        deadline = eager.create_timer(timedelta(hours=1))
+        winner = await task.when_any([approval, deadline])
+        observed.append((winner is approval, approval.is_complete, deadline.is_complete))
+        if winner is approval:
+            deadline.cancel()
+            return f"approved:{approval.get_result()}"
+
+        report = asyncio.create_task(ctx.call_activity("report", return_type=str))
+        await asyncio.sleep(0)
+        observed.append(report.done())
+        return f"timed-out:{await report}"
+
+    deadline = NOW + timedelta(hours=1)
+    old = _start() + [helpers.new_timer_created_event(1, deadline)]
+    if approved_first:
+        new = [helpers.new_event_raised_event("approval", "true")]
+        for _ in range(2):
+            observed.clear()
+            complete = _completion(_execute(orchestrator, old, new))
+            assert json.loads(complete.result.value) == "approved:True"
+            assert observed == [(True, True, False)]
+        return
+
+    fired = [helpers.new_orchestrator_started_event(deadline), helpers.new_timer_fired_event(1, deadline)]
+    actions = _execute(orchestrator, old, fired).actions
+    assert [action.scheduleTask.name for action in actions] == ["report"]
+    assert observed == [(False, False, True), False]
+    old += fired + [helpers.new_task_scheduled_event(2, "report")]
+    for _ in range(2):
+        observed.clear()
+        complete = _completion(_execute(orchestrator, old, [helpers.new_task_completed_event(2, '"sent"')]))
+        assert json.loads(complete.result.value) == "timed-out:sent"
+        assert observed == [(False, False, True), False]
+
+
 def test_owned_timer_cancellation_does_not_cancel_borrowed_timer() -> None:
     async def orchestrator(ctx: task.AsyncOrchestrationContext, _: Any) -> None:
         borrowed = ctx.as_task_context().create_timer(timedelta(seconds=10))
 
@@ -646,7 +646,7 @@ def test_untracked_task_finalized_after_dispatch_close_is_quiet(monkeypatch: pyt
     untracked: list[asyncio.Task[Any]] = []
 
     async def orchestrator(ctx: task.OrchestrationContext, _: Any) -> None:
-        untracked.append(asyncio.Task(ctx.call_activity("background").wait()))
+        untracked.append(asyncio.Task(ctx.call_activity("background")))
         await ctx.call_activity("work")
 
     _execute(orchestrator, [], _start())