Standalone Activity: preserve server-generated request IDs across restarts (#9724)

## What changed?
When generating a request ID server-side, set it on the request struct
before any cloning so that the mutation is re-used by all retries.

## Why?
Without this, there is a bug, although I have not attempted to repro it:

1. Request arrives at Frontend without `requestID`
2. Inside retry interceptor loop, `requestID` is generated and set on a
cloned copy
3. Request proceeds to history, starts the execution, but then some
networking condition in the cell causes `RetryableInterceptor ` not to
receive the Ack (it sees a context expiry)
4. Frontend retries, **generating a new request ID**. But meanwhile the
activity has completed. This would be rare, but technically possible.
5. The default reuse policy permits a second execution to be started.
This would be a bug: the second start should have been prevented by the
request ID. If the user's activity lacks idempotency protection it will
lead to incorrectness in the user's systems.


## How did you test it?

- [x] built
- [x] added weak new unit test(s) for the Start case.
- [x] manually tested:

```diff
commit fa2476c
Author: Dan Davison <dan.davison@temporal.io>
Date:   2 days ago

    Not-for-merge functional test for request-ID stability across server retries
    
    Use a package-level atomic to fail StartActivityExecution once after the
    activity is created at history, triggering the RetryableInterceptor.
    Without the fix, the retry generates a new request ID and gets
    ActivityExecutionAlreadyStarted. With the fix, the retry reuses the
    same request ID and the dedup succeeds.

diff --git a/chasm/lib/activity/frontend.go b/chasm/lib/activity/frontend.go
index c013b17..2a7b96b 100644
--- a/chasm/lib/activity/frontend.go
+++ b/chasm/lib/activity/frontend.go
@@ -2,6 +2,7 @@
 
 import (
 	"context"
+	"sync/atomic"
 
 	"github.com/google/uuid"
 	apiactivitypb "go.temporal.io/api/activity/v1" //nolint:importas
@@ -35,6 +36,10 @@ type FrontendHandler interface {
 
 var ErrStandaloneActivityDisabled = serviceerror.NewUnimplemented("Standalone activity is disabled")
 
+// TestStartFailOnce, when set to true, causes the next StartActivityExecution to return Unavailable
+// after the activity is created. It fires once (CAS to false).
+var TestStartFailOnce atomic.Bool
+
 type frontendHandler struct {
 	FrontendHandler
 	client            activitypb.ActivityServiceClient
@@ -100,6 +105,13 @@ func (h *frontendHandler) StartActivityExecution(ctx context.Context, req *workf
 		NamespaceId:     namespaceID.String(),
 		FrontendRequest: modifiedReq,
 	})
+	if err != nil {
+		return nil, err
+	}
+
+	if TestStartFailOnce.CompareAndSwap(true, false) {
+		return nil, serviceerror.NewUnavailable("test: injected failure after successful creation")
+	}
 
 	return resp.GetFrontendResponse(), err
 }
diff --git a/tests/standalone_activity_test.go b/tests/standalone_activity_test.go
index 6afc7b6..d5ded28 100644
--- a/tests/standalone_activity_test.go
+++ b/tests/standalone_activity_test.go
@@ -270,6 +270,36 @@ func (s *standaloneActivityTestSuite) TestIDConflictPolicy() {
 	})
 }
 
+func (s *standaloneActivityTestSuite) TestServerGeneratedRequestIDStableAcrossRetries() {
+	t := s.T()
+	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
+	defer cancel()
+
+	activityID := testcore.RandomizeStr(t.Name())
+	taskQueue := testcore.RandomizeStr(t.Name())
+
+	// Make the handler fail once with a retryable error so the RetryableInterceptor retries.
+	activity.TestStartFailOnce.Store(true)
+
+	resp, err := s.FrontendClient().StartActivityExecution(ctx, &workflowservice.StartActivityExecutionRequest{
+		Namespace:    s.Namespace().String(),
+		ActivityId:   activityID,
+		ActivityType: s.tv.ActivityType(),
+		Identity:     s.tv.WorkerIdentity(),
+		Input:        defaultInput,
+		TaskQueue: &taskqueuepb.TaskQueue{
+			Name: taskQueue,
+		},
+		StartToCloseTimeout: durationpb.New(defaultStartToCloseTimeout),
+		// No RequestId — server generates one.
+	})
+	// With the fix, the retry uses the same request ID, so history recognizes it as a dedup
+	// and succeeds (with Started=false). Without the fix, the retry generates a new request ID
+	// and gets ActivityExecutionAlreadyStarted.
+	require.NoError(t, err)
+	require.NotNil(t, resp)
+}
+
 func (s *standaloneActivityTestSuite) TestPollActivityTaskQueue() {
 	t := s.T()
 	ctx, cancel := context.WithTimeout(t.Context(), 10*time.Second)
``` 

## Potential risks
Could introduce incorrectness into Standalone Activity


<!-- CURSOR_SUMMARY -->
---

> [!NOTE]
> **Medium Risk**
> Changes request ID generation semantics for standalone activity
Start/Cancel/Terminate paths to improve deduplication across retries;
risk is moderate because it touches request mutation behavior that
affects idempotency and retry interactions.
> 
> **Overview**
> Ensures standalone activity requests reuse a **single** `RequestId`
across frontend retries by generating the server-side ID *before*
cloning/mutating the request (so subsequent retry attempts see the same
ID).
> 
> Removes the prior pre-mutation cloning for
`TerminateActivityExecution` and `RequestCancelActivityExecution`
request-ID population, and adds a unit test (`frontend_test.go`)
asserting `StartActivityExecution` keeps a stable `RequestId` across
multiple `validateAndPopulateStartRequest` calls.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
885f60d. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: dandavison

chasm/lib/activity/frontend.go

@@ -294,8 +294,6 @@ func (h *frontendHandler) TerminateActivityExecution(
 	}
 
 	if req.GetRequestId() == "" {
-		// Since this mutates the request, we clone it first so that any retries use the original request.
-		req = common.CloneProto(req)
 		req.RequestId = uuid.NewString()
 	}
 
@@ -333,8 +331,6 @@ func (h *frontendHandler) RequestCancelActivityExecution(
 	}
 
 	if req.GetRequestId() == "" {
-		// Since this mutates the request, we clone it first so that any retries use the original request.
-		req = common.CloneProto(req)
 		req.RequestId = uuid.NewString()
 	}
 
@@ -362,6 +358,9 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 	req *workflowservice.StartActivityExecutionRequest,
 	namespaceID namespace.ID,
 ) (*workflowservice.StartActivityExecutionRequest, error) {
+	if req.GetRequestId() == "" {
+		req.RequestId = uuid.NewString()
+	}
 	// Since validation includes mutation of the request, we clone it first so that any retries use the original request.
 	req = common.CloneProto(req)
 	activityType := req.ActivityType.GetName()
@@ -400,10 +399,6 @@ func (h *frontendHandler) validateAndPopulateStartRequest(
 func (h *frontendHandler) validateAndNormalizeStartActivityExecutionRequest(
 	req *workflowservice.StartActivityExecutionRequest,
 ) error {
-	if req.GetRequestId() == "" {
-		req.RequestId = uuid.NewString()
-	}
-
 	maxIDLengthLimit := h.config.MaxIDLengthLimit()
 
 	if len(req.GetRequestId()) > maxIDLengthLimit {

chasm/lib/activity/frontend_test.go

@@ -0,0 +1,65 @@
+package activity
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	commonpb "go.temporal.io/api/common/v1"
+	taskqueuepb "go.temporal.io/api/taskqueue/v1"
+	"go.temporal.io/api/workflowservice/v1"
+	"go.temporal.io/server/common/log"
+	"go.temporal.io/server/common/namespace"
+	"google.golang.org/protobuf/types/known/durationpb"
+)
+
+// TestRequestIdStableAcrossRetries verifies that a request ID is re-used
+// across retries, even if server-generated.
+func TestRequestIdStableAcrossRetries(t *testing.T) {
+	h := &frontendHandler{
+		config: &Config{
+			BlobSizeLimitError:         defaultBlobSizeLimitError,
+			BlobSizeLimitWarn:          defaultBlobSizeLimitWarn,
+			MaxIDLengthLimit:           func() int { return defaultMaxIDLengthLimit },
+			DefaultActivityRetryPolicy: getDefaultRetrySettings,
+		},
+		logger: log.NewNoopLogger(),
+	}
+	nsID := namespace.ID("test-namespace-id")
+
+	newReq := func(requestId string) *workflowservice.StartActivityExecutionRequest {
+		return &workflowservice.StartActivityExecutionRequest{
+			Namespace:  "test-namespace",
+			ActivityId: "test-activity",
+			ActivityType: &commonpb.ActivityType{
+				Name: "test-type",
+			},
+			TaskQueue: &taskqueuepb.TaskQueue{
+				Name: "test-queue",
+			},
+			StartToCloseTimeout: durationpb.New(time.Minute),
+			RequestId:           requestId,
+		}
+	}
+
+	// Simulate two RetryableInterceptor attempts: both call
+	// validateAndPopulateStartRequest with the same request pointer.
+	validateTwoAttempts := func(t *testing.T, req *workflowservice.StartActivityExecutionRequest) {
+		t.Helper()
+		clone1, err := h.validateAndPopulateStartRequest(req, nsID)
+		require.NoError(t, err)
+		require.NotEmpty(t, clone1.RequestId)
+
+		clone2, err := h.validateAndPopulateStartRequest(req, nsID)
+		require.NoError(t, err)
+		require.Equal(t, clone1.RequestId, clone2.RequestId)
+	}
+
+	t.Run("server-generated", func(t *testing.T) {
+		validateTwoAttempts(t, newReq(""))
+	})
+
+	t.Run("client-provided", func(t *testing.T) {
+		validateTwoAttempts(t, newReq("my-request-id"))
+	})
+}