#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

96.39

/temporal-sdk/src/main/java/io/temporal/internal/nexus/OperationTokenUtil.java

package io.temporal.internal.nexus;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JavaType;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.ObjectWriter;
import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
import com.google.common.base.Strings;
import java.util.Base64;

public class OperationTokenUtil {
  private static final ObjectMapper mapper = new ObjectMapper().registerModule(new Jdk8Module());
  private static final ObjectWriter ow = mapper.writer();
  private static final Base64.Decoder decoder = Base64.getUrlDecoder();
  private static final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();

  /**
   * Load and validate an operation token without asserting the token type. Use this for cancel
   * dispatch where the token type determines the cancel behavior.
   *
   * @throws IllegalArgumentException if the operation token is malformed or has invalid structure
   */
  public static OperationToken loadOperationToken(String operationToken) {
    OperationToken token;
    try {
      JavaType reference = mapper.getTypeFactory().constructType(OperationToken.class);
      token = mapper.readValue(decoder.decode(operationToken), reference);
    } catch (Exception e) {
      throw new IllegalArgumentException("Failed to parse operation token: " + e.getMessage());
    }
    if (token.getVersion() != null && token.getVersion() != 0) {
      throw new IllegalArgumentException("Invalid operation token: unexpected version field");
    }
    if (Strings.isNullOrEmpty(token.getNamespace())) {
      throw new IllegalArgumentException("Invalid operation token: missing namespace(ns)");
    }
    switch (token.getType()) {
      case WORKFLOW_RUN:
      case WORKFLOW_UPDATE:
        if (Strings.isNullOrEmpty(token.getWorkflowId())) {
          throw new IllegalArgumentException("Invalid operation token: missing workflow ID (wid)");
        }
        break;
      case ACTIVITY_EXECUTION:
        if (Strings.isNullOrEmpty(token.getActivityId())) {
          throw new IllegalArgumentException("Invalid operation token: missing activity ID (aid)");
        }
        break;
      default:
        throw new IllegalArgumentException(
            "Invalid operation token: unknown operation token type: " + token.getType());
    }
    return token;
  }

  /**
   * Load a workflow run operation token, asserting that the token type is {@link
   * OperationTokenType#WORKFLOW_RUN}.
   *
   * @throws IllegalArgumentException if the operation token is invalid or not a workflow run token
   */
  public static OperationToken loadWorkflowRunOperationToken(String operationToken) {
    OperationToken token = loadOperationToken(operationToken);
    if (!token.getType().equals(OperationTokenType.WORKFLOW_RUN)) {
      throw new IllegalArgumentException(
          "Invalid workflow run token: incorrect operation token type: " + token.getType());
    }
    return token;
  }

  /**
   * Load a workflow update operation token, asserting that the token type is {@link
   * OperationTokenType#WORKFLOW_UPDATE}.
   *
   * @throws IllegalArgumentException if the operation token is invalid or not a workflow update
   *     token
   */
  public static OperationToken loadWorkflowUpdateOperationToken(String operationToken) {
    OperationToken token = loadOperationToken(operationToken);
    if (!token.getType().equals(OperationTokenType.WORKFLOW_UPDATE)) {
      throw new IllegalArgumentException(
          "Invalid workflow update token: incorrect operation token type: " + token.getType());
    }
    if (Strings.isNullOrEmpty(token.getUpdateId())) {
      throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)");
    }
    return token;
  }

  /**
   * Extract the workflow ID from a workflow run operation token.
   *
   * @throws IllegalArgumentException if the operation token is invalid
   */
  public static String loadWorkflowIdFromOperationToken(String operationToken) {
    return loadWorkflowRunOperationToken(operationToken).getWorkflowId();
  }

  /**
   * Load an activity execution operation token, asserting that the token type is {@link
   * OperationTokenType#ACTIVITY_EXECUTION}.
   *
   * @throws IllegalArgumentException if the operation token is invalid or not an activity execution
   *     token
   */
  public static OperationToken loadActivityExecutionOperationToken(String operationToken) {
    OperationToken token = loadOperationToken(operationToken);
    if (!token.getType().equals(OperationTokenType.ACTIVITY_EXECUTION)) {
      throw new IllegalArgumentException(
          "Invalid activity execution token: incorrect operation token type: " + token.getType());
    }
    return token;
  }

  /**
   * Extract the activity ID from an activity execution operation token.
   *
   * @throws IllegalArgumentException if the operation token is invalid
   */
  public static String loadActivityIdFromOperationToken(String operationToken) {
    return loadActivityExecutionOperationToken(operationToken).getActivityId();
  }

  /** Generate a workflow run operation token from a workflow ID and namespace. */
  public static String generateWorkflowRunOperationToken(String workflowId, String namespace)
      throws JsonProcessingException {
    String json =
        ow.writeValueAsString(
            new OperationToken(OperationTokenType.WORKFLOW_RUN, namespace, workflowId));
    return encoder.encodeToString(json.getBytes());
  }

  /** Generate a workflow update operation token from namespace, workflowId, runId, updateId */
  public static String generateWorkflowUpdateOperationToken(
      String namespace, String workflowId, String runId, String updateId)
      throws JsonProcessingException {
    if (Strings.isNullOrEmpty(namespace)) {
      throw new IllegalArgumentException("Invalid workflow update token: missing namespace(ns)");
    }
    if (Strings.isNullOrEmpty(workflowId)) {
      throw new IllegalArgumentException(
          "Invalid workflow update token: missing workflow ID (wid)");
    }
    if (Strings.isNullOrEmpty(updateId)) {
      throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)");
    }
    runId = Strings.emptyToNull(runId); // empty runId is allowed but should not be serialized
    String json = ow.writeValueAsString(new OperationToken(namespace, workflowId, runId, updateId));
    return encoder.encodeToString(json.getBytes());
  }

  /**
   * Generate an activity execution operation token from an activity ID and namespace.
   *
   * <p>This overload omits the run ID. Use it when writing the token into the Nexus operation-token
   * callback header — that token is generated before the start RPC completes, so the run ID is not
   * yet known.
   */
  public static String generateActivityExecutionOperationToken(String activityId, String namespace)
      throws JsonProcessingException {
    return generateActivityExecutionOperationToken(activityId, null, namespace);
  }

  /**
   * Generate an activity execution operation token from an activity ID, run ID, and namespace. The
   * {@code runId} is included only when non-null.
   *
   * <p>This overload is used for the operation token returned to the Nexus caller from a start
   * operation — at that point the start RPC has completed and the run ID is known. The header token
   * written into the activity completion callback must NOT carry a run ID; use {@link
   * #generateActivityExecutionOperationToken(String, String)} for that path.
   */
  public static String generateActivityExecutionOperationToken(
      String activityId, String runId, String namespace) throws JsonProcessingException {
    String json =
        ow.writeValueAsString(
            new OperationToken(
                OperationTokenType.ACTIVITY_EXECUTION, namespace, null, activityId, runId));
    return encoder.encodeToString(json.getBytes());
  }

  private OperationTokenUtil() {}
}

1	package io.temporal.internal.nexus;
2
3	import com.fasterxml.jackson.core.JsonProcessingException;
4	import com.fasterxml.jackson.databind.JavaType;
5	import com.fasterxml.jackson.databind.ObjectMapper;
6	import com.fasterxml.jackson.databind.ObjectWriter;
7	import com.fasterxml.jackson.datatype.jdk8.Jdk8Module;
8	import com.google.common.base.Strings;
9	import java.util.Base64;
10
11	public class OperationTokenUtil {
12	private static final ObjectMapper mapper = new ObjectMapper().registerModule(new Jdk8Module());	1✔
13	private static final ObjectWriter ow = mapper.writer();	1✔
14	private static final Base64.Decoder decoder = Base64.getUrlDecoder();	1✔
15	private static final Base64.Encoder encoder = Base64.getUrlEncoder().withoutPadding();	1✔
16
17	/**
18	* Load and validate an operation token without asserting the token type. Use this for cancel
19	* dispatch where the token type determines the cancel behavior.
20	*
21	* @throws IllegalArgumentException if the operation token is malformed or has invalid structure
22	*/
23	public static OperationToken loadOperationToken(String operationToken) {
24	OperationToken token;
25	try {
26	JavaType reference = mapper.getTypeFactory().constructType(OperationToken.class);	1✔
27	token = mapper.readValue(decoder.decode(operationToken), reference);	1✔
28	} catch (Exception e) {	1✔
29	throw new IllegalArgumentException("Failed to parse operation token: " + e.getMessage());	1✔
30	}	1✔
31	if (token.getVersion() != null && token.getVersion() != 0) {	1✔
32	throw new IllegalArgumentException("Invalid operation token: unexpected version field");	1✔
33	}
34	if (Strings.isNullOrEmpty(token.getNamespace())) {	1✔
35	throw new IllegalArgumentException("Invalid operation token: missing namespace(ns)");	1✔
36	}
37	switch (token.getType()) {	1✔
38	case WORKFLOW_RUN:
39	case WORKFLOW_UPDATE:
40	if (Strings.isNullOrEmpty(token.getWorkflowId())) {	1✔
41	throw new IllegalArgumentException("Invalid operation token: missing workflow ID (wid)");	1✔
42	}
43	break;
44	case ACTIVITY_EXECUTION:
45	if (Strings.isNullOrEmpty(token.getActivityId())) {	1✔
46	throw new IllegalArgumentException("Invalid operation token: missing activity ID (aid)");	1✔
47	}
48	break;
49	default:
50	throw new IllegalArgumentException(	1✔
51	"Invalid operation token: unknown operation token type: " + token.getType());	1✔
52	}
53	return token;	1✔
54	}
55
56	/**
57	* Load a workflow run operation token, asserting that the token type is {@link
58	* OperationTokenType#WORKFLOW_RUN}.
59	*
60	* @throws IllegalArgumentException if the operation token is invalid or not a workflow run token
61	*/
62	public static OperationToken loadWorkflowRunOperationToken(String operationToken) {
63	OperationToken token = loadOperationToken(operationToken);	1✔
64	if (!token.getType().equals(OperationTokenType.WORKFLOW_RUN)) {	1✔
65	throw new IllegalArgumentException(	1✔
66	"Invalid workflow run token: incorrect operation token type: " + token.getType());	1✔
67	}
68	return token;	1✔
69	}
70
71	/**
72	* Load a workflow update operation token, asserting that the token type is {@link
73	* OperationTokenType#WORKFLOW_UPDATE}.
74	*
75	* @throws IllegalArgumentException if the operation token is invalid or not a workflow update
76	* token
77	*/
78	public static OperationToken loadWorkflowUpdateOperationToken(String operationToken) {
79	OperationToken token = loadOperationToken(operationToken);	1✔
80	if (!token.getType().equals(OperationTokenType.WORKFLOW_UPDATE)) {	1✔
81	throw new IllegalArgumentException(	1✔
82	"Invalid workflow update token: incorrect operation token type: " + token.getType());	1✔
83	}
84	if (Strings.isNullOrEmpty(token.getUpdateId())) {	1✔
85	throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)");	1✔
86	}
87	return token;	1✔
88	}
89
90	/**
91	* Extract the workflow ID from a workflow run operation token.
92	*
93	* @throws IllegalArgumentException if the operation token is invalid
94	*/
95	public static String loadWorkflowIdFromOperationToken(String operationToken) {
96	return loadWorkflowRunOperationToken(operationToken).getWorkflowId();	1✔
97	}
98
99	/**
100	* Load an activity execution operation token, asserting that the token type is {@link
101	* OperationTokenType#ACTIVITY_EXECUTION}.
102	*
103	* @throws IllegalArgumentException if the operation token is invalid or not an activity execution
104	* token
105	*/
106	public static OperationToken loadActivityExecutionOperationToken(String operationToken) {
107	OperationToken token = loadOperationToken(operationToken);	1✔
108	if (!token.getType().equals(OperationTokenType.ACTIVITY_EXECUTION)) {	1!
NEW 109	throw new IllegalArgumentException(	×
NEW 110	"Invalid activity execution token: incorrect operation token type: " + token.getType());	×
111	}
112	return token;	1✔
113	}
114
115	/**
116	* Extract the activity ID from an activity execution operation token.
117	*
118	* @throws IllegalArgumentException if the operation token is invalid
119	*/
120	public static String loadActivityIdFromOperationToken(String operationToken) {
121	return loadActivityExecutionOperationToken(operationToken).getActivityId();	1✔
122	}
123
124	/** Generate a workflow run operation token from a workflow ID and namespace. */
125	public static String generateWorkflowRunOperationToken(String workflowId, String namespace)
126	throws JsonProcessingException {
127	String json =	1✔
128	ow.writeValueAsString(	1✔
129	new OperationToken(OperationTokenType.WORKFLOW_RUN, namespace, workflowId));
130	return encoder.encodeToString(json.getBytes());	1✔
131	}
132
133	/** Generate a workflow update operation token from namespace, workflowId, runId, updateId */
134	public static String generateWorkflowUpdateOperationToken(
135	String namespace, String workflowId, String runId, String updateId)
136	throws JsonProcessingException {
137	if (Strings.isNullOrEmpty(namespace)) {	1✔
138	throw new IllegalArgumentException("Invalid workflow update token: missing namespace(ns)");	1✔
139	}
140	if (Strings.isNullOrEmpty(workflowId)) {	1✔
141	throw new IllegalArgumentException(	1✔
142	"Invalid workflow update token: missing workflow ID (wid)");
143	}
144	if (Strings.isNullOrEmpty(updateId)) {	1✔
145	throw new IllegalArgumentException("Invalid workflow update token: missing update ID (uid)");	1✔
146	}
147	runId = Strings.emptyToNull(runId); // empty runId is allowed but should not be serialized	1✔
148	String json = ow.writeValueAsString(new OperationToken(namespace, workflowId, runId, updateId));	1✔
149	return encoder.encodeToString(json.getBytes());	1✔
150	}
151
152	/**
153	* Generate an activity execution operation token from an activity ID and namespace.
154	*
155	* <p>This overload omits the run ID. Use it when writing the token into the Nexus operation-token
156	* callback header — that token is generated before the start RPC completes, so the run ID is not
157	* yet known.
158	*/
159	public static String generateActivityExecutionOperationToken(String activityId, String namespace)
160	throws JsonProcessingException {
161	return generateActivityExecutionOperationToken(activityId, null, namespace);	1✔
162	}
163
164	/**
165	* Generate an activity execution operation token from an activity ID, run ID, and namespace. The
166	* {@code runId} is included only when non-null.
167	*
168	* <p>This overload is used for the operation token returned to the Nexus caller from a start
169	* operation — at that point the start RPC has completed and the run ID is known. The header token
170	* written into the activity completion callback must NOT carry a run ID; use {@link
171	* #generateActivityExecutionOperationToken(String, String)} for that path.
172	*/
173	public static String generateActivityExecutionOperationToken(
174	String activityId, String runId, String namespace) throws JsonProcessingException {
175	String json =	1✔
176	ow.writeValueAsString(	1✔
177	new OperationToken(
178	OperationTokenType.ACTIVITY_EXECUTION, namespace, null, activityId, runId));
179	return encoder.encodeToString(json.getBytes());	1✔
180	}
181
182	private OperationTokenUtil() {}
183	}

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