Add AwaitOptions with timerSummary for Async.await() and Workflow.await()

Add overloads for await methods that accept AwaitOptions, allowing users
to specify a timer summary that appears in the workflow history UI/CLI.

- Add AwaitOptions class with timerSummary field
- Add Async.await(Duration, AwaitOptions, Supplier) for non-blocking waits
- Add Workflow.await(Duration, AwaitOptions, Supplier) for blocking waits
- Add test validating timer summary is set in workflow history

Author: mfateev

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptor.java

@@ -762,6 +762,19 @@ public DynamicUpdateHandler getHandler() {
    */
   Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> unblockCondition);
 
+  /**
+   * Asynchronously wait until unblockCondition evaluates to true or timeout expires.
+   *
+   * @param timeout maximum time to wait for the condition
+   * @param options options for the await operation, including timer summary
+   * @param unblockCondition condition that should return true to indicate completion
+   * @return Promise that completes with true if the condition was satisfied, false if the timeout
+   *     expired, or exceptionally with CanceledFailure if the enclosing CancellationScope is
+   *     canceled
+   */
+  Promise<Boolean> awaitAsync(
+      Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition);
+
   Promise<Void> newTimer(Duration duration);
 
   Promise<Void> newTimer(Duration duration, TimerOptions options);

temporal-sdk/src/main/java/io/temporal/common/interceptors/WorkflowOutboundCallsInterceptorBase.java

@@ -2,6 +2,7 @@
 
 import com.uber.m3.tally.Scope;
 import io.temporal.common.SearchAttributeUpdate;
+import io.temporal.workflow.AwaitOptions;
 import io.temporal.workflow.Functions.Func;
 import io.temporal.workflow.MutableSideEffectOptions;
 import io.temporal.workflow.Promise;
@@ -85,6 +86,12 @@ public Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> unblockCo
     return next.awaitAsync(timeout, unblockCondition);
   }
 
+  @Override
+  public Promise<Boolean> awaitAsync(
+      Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+    return next.awaitAsync(timeout, options, unblockCondition);
+  }
+
   @Override
   public Promise<Void> newTimer(Duration duration) {
     return next.newTimer(duration);

temporal-sdk/src/main/java/io/temporal/internal/sync/SyncWorkflowContext.java

@@ -1460,6 +1460,90 @@ public Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> unblockCo
     return result;
   }
 
+  @Override
+  public Promise<Boolean> awaitAsync(
+      Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+    // Check if condition is already true
+    setReadOnly(true);
+    try {
+      if (unblockCondition.get()) {
+        return Workflow.newPromise(true);
+      }
+    } finally {
+      setReadOnly(false);
+    }
+
+    CompletablePromise<Boolean> result = Workflow.newPromise();
+
+    // Capture cancellation state - the condition will be evaluated from the runner thread
+    // where CancellationScope.current() is not available
+    AtomicBoolean cancelled = new AtomicBoolean(false);
+
+    // Create timer - need access to cancellation handle
+    AtomicReference<Functions.Proc1<RuntimeException>> timerCancellation = new AtomicReference<>();
+    AtomicBoolean timerCompleted = new AtomicBoolean(false);
+
+    @Nullable
+    UserMetadata userMetadata =
+        makeUserMetaData(options.getTimerSummary(), null, dataConverterWithCurrentWorkflowContext);
+
+    timerCancellation.set(
+        replayContext.newTimer(
+            timeout,
+            userMetadata,
+            (e) -> {
+              // Set timer flag directly so condition watcher sees it immediately
+              if (e == null) {
+                timerCompleted.set(true);
+              }
+              // Timer cancellation exceptions are ignored - we just care if it fired
+            }));
+
+    // Register with current CancellationScope for timer cancellation
+    CancellationScope.current()
+        .getCancellationRequest()
+        .thenApply(
+            (r) -> {
+              timerCancellation.get().apply(new CanceledFailure(r));
+              return r;
+            });
+
+    Functions.Proc cancelHandle =
+        registerConditionWatcher(
+            () -> {
+              if (cancelled.get()) {
+                throw new CanceledFailure("cancelled");
+              }
+              return unblockCondition.get() || timerCompleted.get();
+            },
+            (e) -> {
+              // Complete promise directly so blocked threads see it immediately
+              if (e != null) {
+                result.completeExceptionally(e);
+              } else {
+                boolean conditionMet = unblockCondition.get();
+                result.complete(conditionMet);
+                if (conditionMet && !timerCompleted.get()) {
+                  // Cancel timer since condition was met first
+                  timerCancellation.get().apply(new CanceledFailure("condition met"));
+                }
+              }
+            });
+
+    // Handle cancellation - complete result promise
+    CancellationScope.current()
+        .getCancellationRequest()
+        .thenApply(
+            (r) -> {
+              cancelled.set(true);
+              result.completeExceptionally(new CanceledFailure(r));
+              cancelHandle.apply(); // Remove the watcher
+              return r;
+            });
+
+    return result;
+  }
+
   @SuppressWarnings("deprecation")
   @Override
   public void continueAsNew(ContinueAsNewInput input) {

temporal-sdk/src/main/java/io/temporal/internal/sync/WorkflowInternal.java

@@ -536,6 +536,15 @@ public static Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> un
     return getWorkflowOutboundInterceptor().awaitAsync(timeout, unblockCondition);
   }
 
+  public static Promise<Boolean> awaitAsync(
+      Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+    assertNotReadOnly("awaitAsync");
+    // Don't wrap the condition with setReadOnly here - the condition will be evaluated
+    // from the runner thread (not a workflow thread), so getRootWorkflowContext() won't work.
+    // SyncWorkflowContext.evaluateConditionWatchers handles setReadOnly directly.
+    return getWorkflowOutboundInterceptor().awaitAsync(timeout, options, unblockCondition);
+  }
+
   public static <R> R sideEffect(Class<R> resultClass, Type resultType, Func<R> func) {
     assertNotReadOnly("side effect");
     return getWorkflowOutboundInterceptor().sideEffect(resultClass, resultType, func);

temporal-sdk/src/main/java/io/temporal/workflow/Async.java

@@ -267,6 +267,30 @@ public static Promise<Boolean> await(
     return WorkflowInternal.awaitAsync(timeout, unblockCondition);
   }
 
+  /**
+   * Asynchronously wait until unblockCondition evaluates to true or timeout expires.
+   *
+   * @param timeout maximum time to wait for the condition
+   * @param options options for the await operation, including timer summary
+   * @param unblockCondition condition that should return true to indicate completion. The condition
+   *     is called on every state transition, so it should never call any blocking operations or
+   *     contain code that mutates workflow state.
+   * @return Promise that completes with:
+   *     <ul>
+   *       <li>true if the condition was satisfied
+   *       <li>false if the timeout expired before the condition was satisfied
+   *       <li>exceptionally with CanceledFailure if the enclosing CancellationScope is canceled
+   *     </ul>
+   *
+   * @see Workflow#await(Duration, AwaitOptions, java.util.function.Supplier) for a blocking version
+   */
+  public static Promise<Boolean> await(
+      Duration timeout,
+      AwaitOptions options,
+      java.util.function.Supplier<Boolean> unblockCondition) {
+    return WorkflowInternal.awaitAsync(timeout, options, unblockCondition);
+  }
+
   /** Prohibits instantiation. */
   private Async() {}
 }

temporal-sdk/src/main/java/io/temporal/workflow/AwaitOptions.java

@@ -0,0 +1,87 @@
+package io.temporal.workflow;
+
+import io.temporal.common.Experimental;
+import java.util.Objects;
+
+/** Options for await operations that involve timers. */
+public final class AwaitOptions {
+
+  public static AwaitOptions.Builder newBuilder() {
+    return new AwaitOptions.Builder();
+  }
+
+  public static AwaitOptions.Builder newBuilder(AwaitOptions options) {
+    return new AwaitOptions.Builder(options);
+  }
+
+  public static AwaitOptions getDefaultInstance() {
+    return DEFAULT_INSTANCE;
+  }
+
+  private static final AwaitOptions DEFAULT_INSTANCE;
+
+  static {
+    DEFAULT_INSTANCE = AwaitOptions.newBuilder().build();
+  }
+
+  public static final class Builder {
+    private String timerSummary;
+
+    private Builder() {}
+
+    private Builder(AwaitOptions options) {
+      if (options == null) {
+        return;
+      }
+      this.timerSummary = options.timerSummary;
+    }
+
+    /**
+     * Single-line fixed summary for the timer used by timed await operations. This will appear in
+     * UI/CLI. Can be in single-line Temporal Markdown format.
+     *
+     * <p>Default is none/empty.
+     */
+    @Experimental
+    public Builder setTimerSummary(String timerSummary) {
+      this.timerSummary = timerSummary;
+      return this;
+    }
+
+    public AwaitOptions build() {
+      return new AwaitOptions(timerSummary);
+    }
+  }
+
+  private final String timerSummary;
+
+  private AwaitOptions(String timerSummary) {
+    this.timerSummary = timerSummary;
+  }
+
+  public String getTimerSummary() {
+    return timerSummary;
+  }
+
+  public Builder toBuilder() {
+    return new Builder(this);
+  }
+
+  @Override
+  public String toString() {
+    return "AwaitOptions{" + "timerSummary='" + timerSummary + '\'' + '}';
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) return true;
+    if (o == null || getClass() != o.getClass()) return false;
+    AwaitOptions that = (AwaitOptions) o;
+    return Objects.equals(timerSummary, that.timerSummary);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(timerSummary);
+  }
+}

temporal-sdk/src/main/java/io/temporal/workflow/Workflow.java

@@ -630,6 +630,32 @@ public static boolean await(Duration timeout, Supplier<Boolean> unblockCondition
         });
   }
 
+  /**
+   * Block current workflow thread until unblockCondition is evaluated to true or timeout passes.
+   *
+   * @param timeout time to unblock even if unblockCondition is not satisfied.
+   * @param options options for the await operation, including timer summary
+   * @param unblockCondition condition that should return true to indicate that thread should
+   *     unblock. The condition is called on every state transition, so it should not contain any
+   *     code that mutates any workflow state. It should also not contain any time based conditions.
+   *     Use timeout parameter for those.
+   * @return false if timed out.
+   * @throws CanceledFailure if thread (or current {@link CancellationScope} was canceled).
+   * @see Async#await(Duration, AwaitOptions, java.util.function.Supplier) for a non-blocking
+   *     version that returns a Promise
+   */
+  public static boolean await(
+      Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+    return WorkflowInternal.awaitAsync(
+            timeout,
+            options,
+            () -> {
+              CancellationScope.throwCanceled();
+              return unblockCondition.get();
+            })
+        .get();
+  }
+
   /**
    * Invokes function retrying in case of failures according to retry options. Synchronous variant.
    * Use {@link Async#retry(RetryOptions, Optional, Functions.Func)} for asynchronous functions.

temporal-sdk/src/test/java/io/temporal/workflow/AsyncAwaitTest.java

@@ -2,7 +2,12 @@
 
 import static org.junit.Assert.*;
 
+import io.temporal.api.common.v1.WorkflowExecution;
+import io.temporal.api.enums.v1.EventType;
+import io.temporal.api.history.v1.HistoryEvent;
+import io.temporal.client.WorkflowStub;
 import io.temporal.failure.CanceledFailure;
+import io.temporal.testUtils.HistoryUtils;
 import io.temporal.testing.internal.SDKTestWorkflowRule;
 import io.temporal.workflow.shared.TestWorkflows.TestWorkflow1;
 import java.time.Duration;
@@ -106,6 +111,21 @@ public void testAsyncAwaitConditionThrows() {
     assertEquals("caught:simulated error", result);
   }
 
+  static final String awaitTimerSummary = "await-timer-summary";
+
+  @Test
+  public void testAwaitWithOptionsSetsTimerSummary() {
+    TestWorkflow1 workflow = testWorkflowRule.newWorkflowStubTimeoutOptions(TestWorkflow1.class);
+    String result = workflow.execute("await-with-options");
+    assertEquals("await-with-options:false", result);
+
+    // Verify the timer summary is set in the workflow history
+    WorkflowExecution exec = WorkflowStub.fromTyped(workflow).getExecution();
+    HistoryEvent timerStartedEvent =
+        testWorkflowRule.getHistoryEvent(exec.getWorkflowId(), EventType.EVENT_TYPE_TIMER_STARTED);
+    HistoryUtils.assertEventMetadata(timerStartedEvent, awaitTimerSummary, null);
+  }
+
   /** Combined workflow that handles all test scenarios. */
   public static class TestAsyncAwaitWorkflow implements TestWorkflow1 {
     private boolean condition1 = false;
@@ -140,6 +160,8 @@ public String execute(String testCase) {
           return testTimedCancellation();
         case "condition-throws":
           return testConditionThrows();
+        case "await-with-options":
+          return testAwaitWithOptions();
         default:
           return "unknown test case";
       }
@@ -359,5 +381,14 @@ private String testConditionThrows() {
         return "caught:" + e.getMessage();
       }
     }
+
+    private String testAwaitWithOptions() {
+      // Use Async.await with AwaitOptions to set a timer summary
+      AwaitOptions options = AwaitOptions.newBuilder().setTimerSummary(awaitTimerSummary).build();
+      // Use a condition that will never be true, so it times out
+      Promise<Boolean> promise = Async.await(Duration.ofMillis(100), options, () -> false);
+      boolean result = promise.get();
+      return "await-with-options:" + result;
+    }
   }
 }

temporal-testing/src/main/java/io/temporal/testing/TestActivityEnvironmentInternal.java

@@ -404,6 +404,12 @@ public Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> unblockCo
       throw new UnsupportedOperationException("not implemented");
     }
 
+    @Override
+    public Promise<Boolean> awaitAsync(
+        Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+      throw new UnsupportedOperationException("not implemented");
+    }
+
     @Override
     public Promise<Void> newTimer(Duration duration) {
       throw new UnsupportedOperationException("not implemented");

temporal-testing/src/main/java/io/temporal/testing/internal/TracingWorkerInterceptor.java

@@ -249,6 +249,15 @@ public Promise<Boolean> awaitAsync(Duration timeout, Supplier<Boolean> unblockCo
       return next.awaitAsync(timeout, unblockCondition);
     }
 
+    @Override
+    public Promise<Boolean> awaitAsync(
+        Duration timeout, AwaitOptions options, Supplier<Boolean> unblockCondition) {
+      if (!WorkflowUnsafe.isReplaying()) {
+        trace.add("awaitAsync " + timeout + " " + options);
+      }
+      return next.awaitAsync(timeout, options, unblockCondition);
+    }
+
     @Override
     public Promise<Void> newTimer(Duration duration) {
       if (!WorkflowUnsafe.isReplaying()) {