Wire workflow reset to support CHASM Nexus operations (probe-and-fallback)

service/history/configs/config.go

@@ -2,6 +2,7 @@ package configs
 
 import (
 	"go.temporal.io/server/chasm/lib/callback"
+	"go.temporal.io/server/chasm/lib/nexusoperation"
 	"go.temporal.io/server/common"
 	"go.temporal.io/server/common/dynamicconfig"
 	"go.temporal.io/server/common/namespace"
@@ -69,6 +70,7 @@ type Config struct {
 	MaxCallbacksPerExecution              dynamicconfig.IntPropertyFnWithNamespaceFilter
 	MaxCallbacksPerUpdateID               dynamicconfig.IntPropertyFnWithNamespaceFilter
 	EnableChasm                           dynamicconfig.BoolPropertyFnWithNamespaceFilter
+	EnableChasmNexusWorkflowOperations    dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	EnableCHASMCallbacks                  dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	EnableCHASMSignalBacklinks            dynamicconfig.BoolPropertyFnWithNamespaceFilter
 	EnableWorkflowUpdateCallbacks         dynamicconfig.BoolPropertyFnWithNamespaceFilter
@@ -494,6 +496,7 @@ func NewConfig(
 		MaxCallbacksPerExecution:              callback.MaxPerExecution.Get(dc),
 		MaxCallbacksPerUpdateID:               dynamicconfig.MaxCallbacksPerUpdateID.Get(dc),
 		EnableChasm:                           dynamicconfig.EnableChasm.Get(dc),
+		EnableChasmNexusWorkflowOperations:    nexusoperation.EnableChasmWorkflowOperations.Get(dc),
 		ChasmMaxInMemoryPureTasks:             dynamicconfig.ChasmMaxInMemoryPureTasks.Get(dc),
 
 		EnableCHASMSchedulerCreation:  dynamicconfig.EnableCHASMSchedulerCreation.Get(dc),

service/history/handler_test.go

@@ -5,9 +5,15 @@ import (
 	"errors"
 	"testing"
 
+	"github.com/nexus-rpc/sdk-go/nexus"
 	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	commonpb "go.temporal.io/api/common/v1"
+	failurepb "go.temporal.io/api/failure/v1"
+	nexuspb "go.temporal.io/api/nexus/v1"
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
+	tokenspb "go.temporal.io/server/api/token/v1"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/membership"
 	"go.temporal.io/server/common/metrics"
@@ -70,3 +76,170 @@ func TestDescribeHistoryHost(t *testing.T) {
 	})
 	assert.NoError(t, err)
 }
+
+func TestScheduledEventIDFromStateMachineRef(t *testing.T) {
+	testCases := []struct {
+		name    string
+		ref     *persistencespb.StateMachineRef
+		wantID  int64
+		wantErr bool
+	}{
+		{
+			name:   "recovers id from last key",
+			ref:    &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{{Type: "nexusoperations.Operation", Id: "42"}}},
+			wantID: 42,
+		},
+		{
+			name: "uses the last (deepest) key",
+			ref: &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{
+				{Type: "nexusoperations.Operation", Id: "7"},
+				{Type: "nexusoperations.Cancelation", Id: "99"},
+			}},
+			wantID: 99,
+		},
+		{
+			name:    "empty path is an error",
+			ref:     &persistencespb.StateMachineRef{},
+			wantErr: true,
+		},
+		{
+			name:    "non-numeric id is an error",
+			ref:     &persistencespb.StateMachineRef{Path: []*persistencespb.StateMachineKey{{Type: "x", Id: "not-a-number"}}},
+			wantErr: true,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			id, err := scheduledEventIDFromStateMachineRef(tc.ref)
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.wantID, id)
+		})
+	}
+}
+
+func TestChasmNexusCompletionFromHSMRequest(t *testing.T) {
+	testCases := []struct {
+		name    string
+		request *historyservice.CompleteNexusOperationRequest
+		verify  func(t *testing.T, completion *persistencespb.ChasmNexusCompletion)
+	}{
+		{
+			name: "success carries the payload",
+			request: &historyservice.CompleteNexusOperationRequest{
+				State:          string(nexus.OperationStateSucceeded),
+				OperationToken: "op-token",
+				Completion:     &tokenspb.NexusOperationCompletion{RequestId: "req-1"},
+				Outcome:        &historyservice.CompleteNexusOperationRequest_Success{Success: &commonpb.Payload{Data: []byte("ok")}},
+			},
+			verify: func(t *testing.T, completion *persistencespb.ChasmNexusCompletion) {
+				require.Equal(t, "req-1", completion.GetRequestId())
+				require.Equal(t, "op-token", completion.GetOperationToken())
+				require.NotNil(t, completion.GetSuccess())
+				require.Equal(t, []byte("ok"), completion.GetSuccess().GetData())
+			},
+		},
+		{
+			name: "failure is converted to a temporal failure",
+			request: &historyservice.CompleteNexusOperationRequest{
+				State:      string(nexus.OperationStateFailed),
+				Completion: &tokenspb.NexusOperationCompletion{RequestId: "req-2"},
+				Outcome:    &historyservice.CompleteNexusOperationRequest_Failure{Failure: &nexuspb.Failure{Message: "boom"}},
+			},
+			verify: func(t *testing.T, completion *persistencespb.ChasmNexusCompletion) {
+				require.NotNil(t, completion.GetFailure())
+				require.Equal(t, "boom", completion.GetFailure().GetMessage())
+			},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			completion, err := chasmNexusCompletionFromHSMRequest(tc.request)
+			require.NoError(t, err)
+			tc.verify(t, completion)
+		})
+	}
+}
+
+func TestScheduledEventIDFromComponentPath(t *testing.T) {
+	testCases := []struct {
+		name    string
+		path    []string
+		wantID  int64
+		wantErr bool
+	}{
+		{name: "recovers id from Operations path", path: []string{"Operations", "42"}, wantID: 42},
+		{name: "uses the last segment", path: []string{"Operations", "7", "13"}, wantID: 13},
+		{name: "empty path is an error", path: nil, wantErr: true},
+		{name: "non-numeric segment is an error", path: []string{"Operations", "nope"}, wantErr: true},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			id, err := scheduledEventIDFromComponentPath(tc.path)
+			if tc.wantErr {
+				require.Error(t, err)
+				return
+			}
+			require.NoError(t, err)
+			require.Equal(t, tc.wantID, id)
+		})
+	}
+}
+
+func TestNexusOperationErrorFromChasmRequest(t *testing.T) {
+	testCases := []struct {
+		name      string
+		request   *historyservice.CompleteNexusOperationChasmRequest
+		wantOpErr bool
+		wantState nexus.OperationState // checked only when wantOpErr is true
+	}{
+		{
+			name: "success yields no operation error",
+			request: &historyservice.CompleteNexusOperationChasmRequest{
+				Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"},
+				Outcome:    &historyservice.CompleteNexusOperationChasmRequest_Success{Success: &commonpb.Payload{}},
+			},
+			wantOpErr: false,
+		},
+		{
+			name: "generic failure yields a failed operation error",
+			request: &historyservice.CompleteNexusOperationChasmRequest{
+				Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"},
+				Outcome:    &historyservice.CompleteNexusOperationChasmRequest_Failure{Failure: &failurepb.Failure{Message: "boom"}},
+			},
+			wantOpErr: true,
+			wantState: nexus.OperationStateFailed,
+		},
+		{
+			name: "canceled failure yields a canceled operation error",
+			request: &historyservice.CompleteNexusOperationChasmRequest{
+				Completion: &tokenspb.NexusOperationCompletion{RequestId: "r"},
+				Outcome: &historyservice.CompleteNexusOperationChasmRequest_Failure{Failure: &failurepb.Failure{
+					Message:     "canceled",
+					FailureInfo: &failurepb.Failure_CanceledFailureInfo{CanceledFailureInfo: &failurepb.CanceledFailureInfo{}},
+				}},
+			},
+			wantOpErr: true,
+			wantState: nexus.OperationStateCanceled,
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			opErr, err := nexusOperationErrorFromChasmRequest(tc.request)
+			require.NoError(t, err)
+			if tc.wantOpErr {
+				require.NotNil(t, opErr)
+				require.Equal(t, tc.wantState, opErr.State)
+			} else {
+				require.Nil(t, opErr)
+			}
+		})
+	}
+}

service/history/history_engine.go

@@ -244,7 +244,8 @@ func NewEngineWithShardContext(
 		historyEngImpl.queueProcessors[processor.Category()] = processor
 	}
 
-	historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.GetMetricsHandler(), logger)
+	historyEngImpl.eventsReapplier = ndc.NewEventsReapplier(shard.StateMachineRegistry(), shard.GetMetricsHandler(), logger).
+		WithChasmWorkflowRegistry(shard.ChasmWorkflowRegistry())
 
 	if shard.GetClusterMetadata().IsGlobalNamespaceEnabled() {
 		historyEngImpl.replicationAckMgr = replication.NewAckManager(

service/history/interfaces/shard_context.go

@@ -10,6 +10,7 @@ import (
 	"go.temporal.io/server/api/historyservice/v1"
 	persistencespb "go.temporal.io/server/api/persistence/v1"
 	"go.temporal.io/server/chasm"
+	chasmworkflow "go.temporal.io/server/chasm/lib/workflow"
 	"go.temporal.io/server/common/archiver"
 	"go.temporal.io/server/common/clock"
 	"go.temporal.io/server/common/cluster"
@@ -119,6 +120,8 @@ type (
 		GetFinalizer() *finalizer.Finalizer
 
 		ChasmRegistry() *chasm.Registry
+		// ChasmWorkflowRegistry returns the CHASM workflow library's event/command registry.
+		ChasmWorkflowRegistry() *chasmworkflow.Registry
 		EndpointRegistry() chasm.EndpointRegistry
 
 		BusinessIDReuseRateLimiter(namespaceID namespace.ID, businessID string, archetypeID chasm.ArchetypeID) quotas.RateLimiter

service/history/ndc/events_reapplier.go

@@ -8,6 +8,7 @@ import (
 	historypb "go.temporal.io/api/history/v1"
 	"go.temporal.io/api/serviceerror"
 	enumsspb "go.temporal.io/server/api/enums/v1"
+	chasmworkflow "go.temporal.io/server/chasm/lib/workflow"
 	"go.temporal.io/server/common/log"
 	"go.temporal.io/server/common/metrics"
 	"go.temporal.io/server/service/history/hsm"
@@ -27,9 +28,10 @@ type (
 	}
 
 	EventsReapplierImpl struct {
-		stateMachineRegistry *hsm.Registry
-		metricsHandler       metrics.Handler
-		logger               log.Logger
+		stateMachineRegistry  *hsm.Registry
+		chasmWorkflowRegistry *chasmworkflow.Registry
+		metricsHandler        metrics.Handler
+		logger                log.Logger
 	}
 )
 
@@ -46,6 +48,14 @@ func NewEventsReapplier(
 	}
 }
 
+// WithChasmWorkflowRegistry sets the CHASM workflow registry used to resolve CHASM-backed events on the
+// reapply path. Kept separate from NewEventsReapplier so the constructor signature stays backward compatible
+// for external callers.
+func (r *EventsReapplierImpl) WithChasmWorkflowRegistry(chasmWorkflowRegistry *chasmworkflow.Registry) *EventsReapplierImpl {
+	r.chasmWorkflowRegistry = chasmWorkflowRegistry
+	return r
+}
+
 func (r *EventsReapplierImpl) ReapplyEvents(
 	ctx context.Context,
 	ms historyi.MutableState,
@@ -57,7 +67,7 @@ func (r *EventsReapplierImpl) ReapplyEvents(
 	if !ms.IsWorkflowExecutionRunning() {
 		return nil, serviceerror.NewInternal("unable to reapply events to closed workflow.")
 	}
-	reappliedEvents, err := reapplyEvents(ctx, ms, updateRegistry, r.stateMachineRegistry, historyEvents, nil, runID, false)
+	reappliedEvents, err := reapplyEvents(ctx, ms, updateRegistry, r.stateMachineRegistry, r.chasmWorkflowRegistry, historyEvents, nil, runID, false)
 	if err != nil {
 		return nil, err
 	}