#349

Committed 03 Aug 2026 05:42AM UTC coverage: 68.221% (+0.08%) from 68.143%

Build # #349

Build Type

push

github

Committed by

web-flow

Commit Message

Add Standalone Activities to Temporal Nexus Operation Handler (#2918)

* Enable Nexus activity operations without regressing workflow updates

Compose standalone activity support with the workflow-update Nexus model already present on master. Shared token, callback, link, client, and cancellation paths retain both operation families.

Constraint: Preserve the workflow-update token and API contracts from master
Rejected: Choose one conflict side | each side would drop a supported Nexus operation family
Confidence: high
Scope-risk: moderate
Reversibility: clean
Directive: Keep activity execution at token type 2 and workflow update at token type 3
Tested: Focused temporal-sdk token, link, invoker, client, async activity, and cancellation tests
Not-tested: Full repository suite and real-server-only standalone activity cases

* Make sure we test attaching

* Make sure to handle ActivityAlreadyStartedException

* Add test with SANO

* Bump test server

Coverage Stats

7220 of 12610 branches covered (57.26%)

Branch coverage included in aggregate %.

182 of 283 new or added lines in 15 files covered. (64.31%)

13 existing lines in 3 files now uncovered.

29918 of 41828 relevant lines covered (71.53%)

0.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

50.0

/temporal-sdk/src/main/java/io/temporal/nexus/TemporalOperationHandler.java

package io.temporal.nexus;

import io.nexusrpc.OperationException;
import io.nexusrpc.handler.*;
import io.temporal.client.ActivityClient;
import io.temporal.client.ActivityClientOptions;
import io.temporal.client.WorkflowClient;
import io.temporal.common.Experimental;
import io.temporal.internal.nexus.CurrentNexusOperationContext;
import io.temporal.internal.nexus.InternalNexusOperationContext;
import io.temporal.internal.nexus.OperationToken;
import io.temporal.internal.nexus.OperationTokenUtil;

/**
 * Generic Nexus operation handler backed by Temporal. Implements {@link OperationHandler} and
 * provides a composable way to map Temporal operations (start workflow, start activity, etc.) to
 * Nexus operations.
 *
 * <p>Usage example:
 *
 * <pre>{@code
 * @OperationImpl
 * public OperationHandler<TransferInput, TransferResult> startTransfer() {
 *   return TemporalOperationHandler.create((context, client, input) -> {
 *     return client.startWorkflow(
 *         TransferWorkflow.class,
 *         TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
 *         WorkflowOptions.newBuilder()
 *             .setWorkflowId("transfer-" + input.getTransferId())
 *             .build());
 *   });
 * }
 * }</pre>
 *
 * <p>This class supports subclassing to customize cancel behavior. Override {@link
 * #cancelWorkflowRun} to change how workflow-run (token type {@code t:1}) cancellations are
 * handled, or {@link #cancelActivityExecution} to change how activity-execution (token type {@code
 * t:2}) cancellations are handled. The {@link #start} and {@link #cancel} methods should not be
 * overridden — they contain the core dispatch logic.
 *
 * @param <T> the input type
 * @param <R> the result type
 */
@Experimental
public class TemporalOperationHandler<T, R> implements OperationHandler<T, R> {

  /**
   * Handler invoked when a Nexus start operation request is received.
   *
   * @param <T> the input type
   * @param <R> the result type
   */
  @FunctionalInterface
  public interface StartHandler<T, R> {
    TemporalOperationResult<R> apply(
        TemporalOperationStartContext context, TemporalNexusClient client, T input)
        throws OperationException;
  }

  private final StartHandler<T, R> startHandler;

  protected TemporalOperationHandler(StartHandler<T, R> startHandler) {
    this.startHandler = startHandler;
  }

  /**
   * Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link
   * #cancelWorkflowRun} or {@link #cancelActivityExecution} to customize cancel behavior.
   *
   * @param startHandler the handler to invoke on start operation requests
   * @return an operation handler backed by the given start handler
   */
  public static <T, R> TemporalOperationHandler<T, R> create(StartHandler<T, R> startHandler) {
    return new TemporalOperationHandler<>(startHandler);
  }

  @Override
  public final OperationStartResult<R> start(
      OperationContext ctx, OperationStartDetails details, T input) throws OperationException {
    InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();
    TemporalNexusClient client =
        new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);

    TemporalOperationStartContext startContext = new TemporalOperationStartContext(ctx, details);
    TemporalOperationResult<R> result = startHandler.apply(startContext, client, input);

    if (result.isSync()) {
      return OperationStartResult.newSyncBuilder(result.getSyncResult()).build();
    } else if (result.isAsync()) {
      return OperationStartResult.<R>newAsyncBuilder(result.getAsyncOperationToken()).build();
    } else {
      throw new HandlerException(
          HandlerException.ErrorType.INTERNAL,
          new IllegalStateException("TemporalOperationResult must be either sync or async"));
    }
  }

  @Override
  public final void cancel(OperationContext ctx, OperationCancelDetails details) {
    OperationToken token;
    try {
      token = OperationTokenUtil.loadOperationToken(details.getOperationToken());
    } catch (IllegalArgumentException e) {
      throw new HandlerException(
          HandlerException.ErrorType.BAD_REQUEST, "failed to parse operation token", e);
    }

    TemporalOperationCancelContext cancelContext = new TemporalOperationCancelContext(ctx, details);
    switch (token.getType()) {
      case WORKFLOW_RUN:
        cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId()));
        break;
      case WORKFLOW_UPDATE:
        cancelUpdateWorkflow(
            cancelContext,
            new CancelUpdateWorkflowInput(
                token.getWorkflowId(), token.getRunId(), token.getUpdateId()));
        break;
      case ACTIVITY_EXECUTION:
        cancelActivityExecution(
            cancelContext,
            new CancelActivityExecutionInput(token.getActivityId(), token.getRunId()));
        break;
      default:
        throw new HandlerException(
            HandlerException.ErrorType.BAD_REQUEST,
            new IllegalArgumentException("unsupported operation token type: " + token.getType()));
    }
  }

  /**
   * Called when a cancel request is received for a workflow-run token (type=1). Override to
   * customize cancel behavior.
   *
   * <p>Default behavior: cancels the underlying workflow.
   *
   * @param context the cancel context
   * @param input describes the workflow run to cancel
   */
  protected void cancelWorkflowRun(
      TemporalOperationCancelContext context, CancelWorkflowRunInput input) {
    WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();
    client.newUntypedWorkflowStub(input.getWorkflowId()).cancel();
  }

  /**
   * Called when a cancel request is received for a workflow update token. Override to customize
   * cancel behavior.
   *
   * <p>Default behavior: not implemented. There is no server primitive to cancel an in-flight
   * workflow update.
   *
   * @param context the cancel context
   * @param input describes the update to cancel
   */
  protected void cancelUpdateWorkflow(
      TemporalOperationCancelContext context, CancelUpdateWorkflowInput input) {
    throw new HandlerException(
        HandlerException.ErrorType.NOT_IMPLEMENTED,
        new UnsupportedOperationException("cannot cancel an UpdateWorkflow operation"));
  }

  /**
   * Called when a cancel request is received for an activity-execution token (type=2). Override to
   * customize cancel behavior.
   *
   * <p>Default behavior: requests cancellation of the underlying standalone activity execution.
   *
   * @param context the cancel context
   * @param input describes the activity execution to cancel
   */
  protected void cancelActivityExecution(
      TemporalOperationCancelContext context, CancelActivityExecutionInput input) {
    WorkflowClient wc = CurrentNexusOperationContext.get().getWorkflowClient();
    ActivityClient ac =
        ActivityClient.newInstance(
            wc.getWorkflowServiceStubs(),
            ActivityClientOptions.newBuilder()
                .setNamespace(wc.getOptions().getNamespace())
                .setDataConverter(wc.getOptions().getDataConverter())
                .setIdentity(wc.getOptions().getIdentity())
                .build());
    ac.getHandle(input.getActivityId(), input.getRunId()).cancel();
  }
}

1	package io.temporal.nexus;
2
3	import io.nexusrpc.OperationException;
4	import io.nexusrpc.handler.*;
5	import io.temporal.client.ActivityClient;
6	import io.temporal.client.ActivityClientOptions;
7	import io.temporal.client.WorkflowClient;
8	import io.temporal.common.Experimental;
9	import io.temporal.internal.nexus.CurrentNexusOperationContext;
10	import io.temporal.internal.nexus.InternalNexusOperationContext;
11	import io.temporal.internal.nexus.OperationToken;
12	import io.temporal.internal.nexus.OperationTokenUtil;
13
14	/**
15	* Generic Nexus operation handler backed by Temporal. Implements {@link OperationHandler} and
16	* provides a composable way to map Temporal operations (start workflow, start activity, etc.) to
17	* Nexus operations.
18	*
19	* <p>Usage example:
20	*
21	* <pre>{@code
22	* @OperationImpl
23	* public OperationHandler<TransferInput, TransferResult> startTransfer() {
24	* return TemporalOperationHandler.create((context, client, input) -> {
25	* return client.startWorkflow(
26	* TransferWorkflow.class,
27	* TransferWorkflow::transfer, input.getFromAccount(), input.getToAccount(),
28	* WorkflowOptions.newBuilder()
29	* .setWorkflowId("transfer-" + input.getTransferId())
30	* .build());
31	* });
32	* }
33	* }</pre>
34	*
35	* <p>This class supports subclassing to customize cancel behavior. Override {@link
36	* #cancelWorkflowRun} to change how workflow-run (token type {@code t:1}) cancellations are
37	* handled, or {@link #cancelActivityExecution} to change how activity-execution (token type {@code
38	* t:2}) cancellations are handled. The {@link #start} and {@link #cancel} methods should not be
39	* overridden — they contain the core dispatch logic.
40	*
41	* @param <T> the input type
42	* @param <R> the result type
43	*/
44	@Experimental
45	public class TemporalOperationHandler<T, R> implements OperationHandler<T, R> {
46
47	/**
48	* Handler invoked when a Nexus start operation request is received.
49	*
50	* @param <T> the input type
51	* @param <R> the result type
52	*/
53	@FunctionalInterface
54	public interface StartHandler<T, R> {
55	TemporalOperationResult<R> apply(
56	TemporalOperationStartContext context, TemporalNexusClient client, T input)
57	throws OperationException;
58	}
59
60	private final StartHandler<T, R> startHandler;
61
62	protected TemporalOperationHandler(StartHandler<T, R> startHandler) {	1✔
63	this.startHandler = startHandler;	1✔
64	}	1✔
65
66	/**
67	* Creates a {@link TemporalOperationHandler} from a start handler. Subclass and override {@link
68	* #cancelWorkflowRun} or {@link #cancelActivityExecution} to customize cancel behavior.
69	*
70	* @param startHandler the handler to invoke on start operation requests
71	* @return an operation handler backed by the given start handler
72	*/
73	public static <T, R> TemporalOperationHandler<T, R> create(StartHandler<T, R> startHandler) {
74	return new TemporalOperationHandler<>(startHandler);	1✔
75	}
76
77	@Override
78	public final OperationStartResult<R> start(
79	OperationContext ctx, OperationStartDetails details, T input) throws OperationException {
80	InternalNexusOperationContext nexusCtx = CurrentNexusOperationContext.get();	1✔
81	TemporalNexusClient client =	1✔
82	new TemporalNexusClientImpl(nexusCtx.getWorkflowClient(), ctx, details);	1✔
83
84	TemporalOperationStartContext startContext = new TemporalOperationStartContext(ctx, details);	1✔
85	TemporalOperationResult<R> result = startHandler.apply(startContext, client, input);	1✔
86
87	if (result.isSync()) {	1✔
88	return OperationStartResult.newSyncBuilder(result.getSyncResult()).build();	1✔
89	} else if (result.isAsync()) {	1!
90	return OperationStartResult.<R>newAsyncBuilder(result.getAsyncOperationToken()).build();	1✔
91	} else {
92	throw new HandlerException(	×
93	HandlerException.ErrorType.INTERNAL,
94	new IllegalStateException("TemporalOperationResult must be either sync or async"));
95	}
96	}
97
98	@Override
99	public final void cancel(OperationContext ctx, OperationCancelDetails details) {
100	OperationToken token;
101	try {
102	token = OperationTokenUtil.loadOperationToken(details.getOperationToken());	1✔
103	} catch (IllegalArgumentException e) {	×
104	throw new HandlerException(	×
105	HandlerException.ErrorType.BAD_REQUEST, "failed to parse operation token", e);
106	}	1✔
107
108	TemporalOperationCancelContext cancelContext = new TemporalOperationCancelContext(ctx, details);	1✔
109	switch (token.getType()) {	1!
110	case WORKFLOW_RUN:
111	cancelWorkflowRun(cancelContext, new CancelWorkflowRunInput(token.getWorkflowId()));	1✔
112	break;	1✔
113	case WORKFLOW_UPDATE:
114	cancelUpdateWorkflow(	×
115	cancelContext,
116	new CancelUpdateWorkflowInput(
117	token.getWorkflowId(), token.getRunId(), token.getUpdateId()));	×
118	break;	×
119	case ACTIVITY_EXECUTION:
NEW 120	cancelActivityExecution(	×
121	cancelContext,
NEW 122	new CancelActivityExecutionInput(token.getActivityId(), token.getRunId()));	×
NEW 123	break;	×
124	default:
125	throw new HandlerException(	×
126	HandlerException.ErrorType.BAD_REQUEST,
127	new IllegalArgumentException("unsupported operation token type: " + token.getType()));	×
128	}
129	}	1✔
130
131	/**
132	* Called when a cancel request is received for a workflow-run token (type=1). Override to
133	* customize cancel behavior.
134	*
135	* <p>Default behavior: cancels the underlying workflow.
136	*
137	* @param context the cancel context
138	* @param input describes the workflow run to cancel
139	*/
140	protected void cancelWorkflowRun(
141	TemporalOperationCancelContext context, CancelWorkflowRunInput input) {
142	WorkflowClient client = CurrentNexusOperationContext.get().getWorkflowClient();	1✔
143	client.newUntypedWorkflowStub(input.getWorkflowId()).cancel();	1✔
144	}	1✔
145
146	/**
147	* Called when a cancel request is received for a workflow update token. Override to customize
148	* cancel behavior.
149	*
150	* <p>Default behavior: not implemented. There is no server primitive to cancel an in-flight
151	* workflow update.
152	*
153	* @param context the cancel context
154	* @param input describes the update to cancel
155	*/
156	protected void cancelUpdateWorkflow(
157	TemporalOperationCancelContext context, CancelUpdateWorkflowInput input) {
158	throw new HandlerException(	×
159	HandlerException.ErrorType.NOT_IMPLEMENTED,
160	new UnsupportedOperationException("cannot cancel an UpdateWorkflow operation"));
161	}
162
163	/**
164	* Called when a cancel request is received for an activity-execution token (type=2). Override to
165	* customize cancel behavior.
166	*
167	* <p>Default behavior: requests cancellation of the underlying standalone activity execution.
168	*
169	* @param context the cancel context
170	* @param input describes the activity execution to cancel
171	*/
172	protected void cancelActivityExecution(
173	TemporalOperationCancelContext context, CancelActivityExecutionInput input) {
NEW 174	WorkflowClient wc = CurrentNexusOperationContext.get().getWorkflowClient();	×
NEW 175	ActivityClient ac =	×
NEW 176	ActivityClient.newInstance(	×
NEW 177	wc.getWorkflowServiceStubs(),	×
NEW 178	ActivityClientOptions.newBuilder()	×
NEW 179	.setNamespace(wc.getOptions().getNamespace())	×
NEW 180	.setDataConverter(wc.getOptions().getDataConverter())	×
NEW 181	.setIdentity(wc.getOptions().getIdentity())	×
NEW 182	.build());	×
NEW 183	ac.getHandle(input.getActivityId(), input.getRunId()).cancel();	×
NEW 184	}	×
185	}

temporalio / sdk-java / #349

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous