Implement detached component (#9086)

## What changed?
Implement detached component as a Field option and Registrable Component
option.

Add detached boolean value to ComponentAttributes persistence proto
definition.

## Why?
Allow detached components to continue updates and task execution even if
parent node lifecycle is closed.

## How did you test it?
- [X] built
- [X] run locally and tested manually
- [X] covered by existing tests
- [X] added new unit test(s)
- [ ] added new functional test(s)

api/persistence/v1/chasm.pb.go

@@ -221,7 +221,11 @@ type ChasmComponentAttributes struct {
 	SideEffectTasks []*ChasmComponentAttributes_Task `protobuf:"bytes,2,rep,name=side_effect_tasks,json=sideEffectTasks,proto3" json:"side_effect_tasks,omitempty"`
 	// Tasks are ordered by their scheduled time, breaking ties by
 	// versioned transition and versioned_transition_offset.
-	PureTasks     []*ChasmComponentAttributes_Task `protobuf:"bytes,3,rep,name=pure_tasks,json=pureTasks,proto3" json:"pure_tasks,omitempty"`
+	PureTasks []*ChasmComponentAttributes_Task `protobuf:"bytes,3,rep,name=pure_tasks,json=pureTasks,proto3" json:"pure_tasks,omitempty"`
+	// When true, this component ignores parent lifecycle validation.
+	// Detached components can continue operating, accepting writes and executing
+	// tasks, even when their parent is closed/terminated.
+	Detached      bool `protobuf:"varint,4,opt,name=detached,proto3" json:"detached,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
@@ -277,6 +281,13 @@ func (x *ChasmComponentAttributes) GetPureTasks() []*ChasmComponentAttributes_Ta
 	return nil
 }
 
+func (x *ChasmComponentAttributes) GetDetached() bool {
+	if x != nil {
+		return x.Detached
+	}
+	return false
+}
+
 type ChasmDataAttributes struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
 	unknownFields protoimpl.UnknownFields
@@ -804,12 +815,13 @@ const file_temporal_server_api_persistence_v1_chasm_proto_rawDesc = "" +
 	"\x15collection_attributes\x18\r \x01(\v2=.temporal.server.api.persistence.v1.ChasmCollectionAttributesH\x00R\x14collectionAttributes\x12k\n" +
 	"\x12pointer_attributes\x18\x0e \x01(\v2:.temporal.server.api.persistence.v1.ChasmPointerAttributesH\x00R\x11pointerAttributesB\f\n" +
 	"\n" +
-	"attributes\"\x9f\x05\n" +
+	"attributes\"\xbb\x05\n" +
 	"\x18ChasmComponentAttributes\x12\x17\n" +
 	"\atype_id\x18\x01 \x01(\rR\x06typeId\x12m\n" +
 	"\x11side_effect_tasks\x18\x02 \x03(\v2A.temporal.server.api.persistence.v1.ChasmComponentAttributes.TaskR\x0fsideEffectTasks\x12`\n" +
 	"\n" +
-	"pure_tasks\x18\x03 \x03(\v2A.temporal.server.api.persistence.v1.ChasmComponentAttributes.TaskR\tpureTasks\x1a\x98\x03\n" +
+	"pure_tasks\x18\x03 \x03(\v2A.temporal.server.api.persistence.v1.ChasmComponentAttributes.TaskR\tpureTasks\x12\x1a\n" +
+	"\bdetached\x18\x04 \x01(\bR\bdetached\x1a\x98\x03\n" +
 	"\x04Task\x12\x17\n" +
 	"\atype_id\x18\x01 \x01(\rR\x06typeId\x12 \n" +
 	"\vdestination\x18\x02 \x01(\tR\vdestination\x12A\n" +

chasm/field.go

@@ -39,8 +39,14 @@ func NewComponentField[C Component](
 	c C,
 	options ...ComponentFieldOption,
 ) Field[C] {
+	opts := &componentFieldOptions{}
+	for _, o := range options {
+		o(opts)
+	}
+	internal := newFieldInternalWithValue(fieldTypeComponent, c)
+	internal.detached = opts.detached
 	return Field[C]{
-		Internal: newFieldInternalWithValue(fieldTypeComponent, c),
+		Internal: internal,
 	}
 }
 

chasm/field_internal.go

@@ -8,6 +8,10 @@ type fieldInternal struct {
 
 	// Pointer to the corresponding tree node. Can be nil for the just created fields.
 	node *Node
+
+	// Detached field option. When true, the node created from this field
+	// will be detached regardless of the component type's registration.
+	detached bool
 }
 
 func newFieldInternalWithValue(ft fieldType, v any) fieldInternal {

chasm/library_core.go

@@ -11,7 +11,7 @@ func (b *CoreLibrary) Name() string {
 
 func (b *CoreLibrary) Components() []*RegistrableComponent {
 	return []*RegistrableComponent{
-		NewRegistrableComponent[*Visibility]("vis"),
+		NewRegistrableComponent[*Visibility]("vis", WithDetached()),
 	}
 }
 

chasm/registrable_component.go

@@ -20,6 +20,7 @@ type (
 
 		ephemeral     bool
 		singleCluster bool
+		detached      bool
 
 		searchAttributesMapper *VisibilitySearchAttributesMapper
 	}
@@ -54,6 +55,22 @@ func WithSingleCluster() RegistrableComponentOption {
 	}
 }
 
+// WithDetached marks the registrable component as detached. Detached components ignore
+// parent lifecycle validation, allowing them to continue operating when their
+// parent is closed/terminated.
+// If a registrable component is not detached by default, a component definition
+// can specify its child as detached via ComponentFieldDetached() option.
+func WithDetached() RegistrableComponentOption {
+	return func(rc *RegistrableComponent) {
+		rc.detached = true
+	}
+}
+
+// IsDetached returns true if the component type is registered as detached.
+func (rc *RegistrableComponent) IsDetached() bool {
+	return rc.detached
+}
+
 // WithBusinessIDAlias allows specifying the business ID alias of the component.
 // This option must be specified if the archetype uses the Visibility component.
 func WithBusinessIDAlias(

chasm/registry_test.go

@@ -79,6 +79,29 @@ func (s *RegistryTestSuite) TestRegistry_RegisterComponents_Success() {
 	require.Nil(s.T(), rc3)
 }
 
+func (s *RegistryTestSuite) TestRegistry_RegisterComponents_WithDetached() {
+	r := chasm.NewRegistry(s.logger)
+	ctrl := gomock.NewController(s.T())
+	lib := chasm.NewMockLibrary(ctrl)
+	lib.EXPECT().Name().Return("TestLibrary").AnyTimes()
+	lib.EXPECT().Components().Return([]*chasm.RegistrableComponent{
+		chasm.NewRegistrableComponent[*chasm.MockComponent]("DetachedComponent", chasm.WithDetached()),
+	})
+	lib.EXPECT().Tasks().Return(nil)
+
+	err := r.Register(lib)
+	s.Require().NoError(err)
+
+	// Detached component should have IsDetached() return true
+	detachedRC, ok := r.Component("TestLibrary.DetachedComponent")
+	s.Require().True(ok)
+	s.Require().True(detachedRC.IsDetached())
+
+	// Verify that a component without WithDetached() has IsDetached() return false
+	normalRC := chasm.NewRegistrableComponent[*chasm.MockComponent]("NormalComponent")
+	s.Require().False(normalRC.IsDetached())
+}
+
 func (s *RegistryTestSuite) TestRegistry_RegisterTasks_Success() {
 	r := chasm.NewRegistry(s.logger)
 	ctrl := gomock.NewController(s.T())

chasm/test_component_test.go

@@ -124,7 +124,14 @@ func (tc *TestComponent) Memo(_ Context) proto.Message {
 }
 
 func (tsc1 *TestSubComponent1) LifecycleState(_ Context) LifecycleState {
-	return LifecycleStateRunning
+	switch tsc1.SubComponent1Data.GetStatus() {
+	case enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING:
+		return LifecycleStateRunning
+	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW:
+		return LifecycleStateCompleted
+	default:
+		return LifecycleStateFailed
+	}
 }
 
 func (tsc1 *TestSubComponent1) GetData() string {
@@ -132,7 +139,14 @@ func (tsc1 *TestSubComponent1) GetData() string {
 }
 
 func (tsc11 *TestSubComponent11) LifecycleState(_ Context) LifecycleState {
-	return LifecycleStateRunning
+	switch tsc11.SubComponent11Data.GetStatus() {
+	case enumspb.WORKFLOW_EXECUTION_STATUS_UNSPECIFIED, enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING:
+		return LifecycleStateRunning
+	case enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, enumspb.WORKFLOW_EXECUTION_STATUS_CONTINUED_AS_NEW:
+		return LifecycleStateCompleted
+	default:
+		return LifecycleStateFailed
+	}
 }
 
 func (tsc2 *TestSubComponent2) LifecycleState(_ Context) LifecycleState {

chasm/tree.go

@@ -414,13 +414,8 @@ func (n *Node) Component(
 			fmt.Errorf("%s", reflect.TypeOf(node.value).String()))
 	}
 
-	// Access check always begins on the target node's parent, and ignored for nodes
-	// without ancestors.
-	if node.parent != nil {
-		err := node.parent.validateAccess(validationContext)
-		if err != nil {
-			return nil, err
-		}
+	if err := node.validateAccess(validationContext); err != nil {
+		return nil, err
 	}
 
 	if ref.validationFn != nil {
@@ -440,7 +435,7 @@ func (n *Node) Component(
 //
 // When the context's intent is OperationIntentProgress, This check validates that
 // all of a node's ancestors are still in a running state, and can accept writes. In
-// the case of a newly-created node, a detached node, or an OperationIntentObserve
+// the case of a newly created node, a detached node, or an OperationIntentObserve
 // intent, the check is skipped.
 func (n *Node) validateAccess(ctx Context) error {
 	intent := operationIntentFromContext(ctx.getContext())
@@ -449,11 +444,21 @@ func (n *Node) validateAccess(ctx Context) error {
 		return nil
 	}
 
-	// TODO - check if this is a detached node, operations are always allowed.
+	// Detached nodes skip ancestor validation entirely.
+	if n.isDetached() || n.parent == nil {
+		return nil
+	}
 
-	if n.parent != nil {
-		err := n.parent.validateAccess(ctx)
-		if err != nil {
+	return n.parent.validateAccessHelper(ctx)
+}
+
+// validateAccessHelper is a helper method that validates both the current
+// node's lifecycle state AND its ancestors recursively.
+// Do not call this method directly, call validateAccess instead.
+func (n *Node) validateAccessHelper(ctx Context) error {
+	// Check ancestors first (if not detached).
+	if !n.isDetached() && n.parent != nil {
+		if err := n.parent.validateAccessHelper(ctx); err != nil {
 			return err
 		}
 	}
@@ -464,8 +469,7 @@ func (n *Node) validateAccess(ctx Context) error {
 	}
 
 	// Hydrate the component so we can access its LifecycleState.
-	err := n.prepareComponentValue(ctx)
-	if err != nil {
+	if err := n.prepareComponentValue(ctx); err != nil {
 		return err
 	}
 	componentValue, _ := n.value.(Component) //nolint:revive // unchecked-type-assertion
@@ -578,6 +582,10 @@ func (n *Node) isMap() bool {
 	return n.serializedNode.GetMetadata().GetCollectionAttributes() != nil
 }
 
+func (n *Node) isDetached() bool {
+	return n.serializedNode.GetMetadata().GetComponentAttributes().GetDetached()
+}
+
 func (n *Node) fieldType() fieldType {
 	if n.serializedNode.GetMetadata().GetComponentAttributes() != nil {
 		return fieldTypeComponent
@@ -1014,6 +1022,15 @@ func (n *Node) syncSubField(
 			}
 
 			childNode.setValueState(valueStateNeedSyncStructure)
+
+			// Set detached flag from field option or component type registration.
+			componentAttr := childNode.serializedNode.GetMetadata().GetComponentAttributes()
+			componentAttr.Detached = internal.detached
+			if !componentAttr.Detached {
+				if rc, ok := n.registry.componentFor(fieldValue); ok {
+					componentAttr.Detached = rc.IsDetached()
+				}
+			}
 		case fieldTypeData:
 			if err = assertStructPointer(reflect.TypeOf(fieldValue)); err != nil {
 				return
@@ -1773,16 +1790,11 @@ func (n *Node) validateTask(
 			fmt.Errorf("%s", reflect.TypeOf(taskInstance).Name()))
 	}
 
-	// TODO: visibility component should be an (implicitly) detached component.
-	// Remove this special case when detached node is implemented.
-	if registableTask.taskTypeID != visibilityTaskTypeID && n.parent != nil {
-		err := n.parent.validateAccess(validateContext)
+	if err := n.validateAccess(validateContext); err != nil {
 		if errors.Is(err, errAccessCheckFailed) {
 			return false, nil
 		}
-		if err != nil {
-			return false, err
-		}
+		return false, err
 	}
 
 	defer log.CapturePanic(n.logger, &retErr)

chasm/tree_test.go

@@ -1464,17 +1464,32 @@ func (s *nodeSuite) TestValidateAccess() {
 		setup func(*Node, Context) error
 	}{
 		{
-			name:            "access check applies only to ancestors",
+			name:            "access check applies only to ancestors (terminated)",
 			valid:           true,
 			intent:          OperationIntentProgress,
 			lifecycleStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
 			terminated:      false,
-			setup: func(target *Node, _ Context) error {
+			setup: func(target *Node, ctx Context) error {
 				// Set the terminated flag on the target node instead of an ancestor
 				target.terminated = true
 				return nil
 			},
 		},
+		{
+			name:            "access check applies only to ancestors (closed)",
+			valid:           true,
+			intent:          OperationIntentProgress,
+			lifecycleStatus: enumspb.WORKFLOW_EXECUTION_STATUS_RUNNING,
+			terminated:      false,
+			setup: func(target *Node, ctx Context) error {
+				if err := target.prepareComponentValue(ctx); err != nil {
+					return err
+				}
+				targetComponent, _ := target.value.(*TestSubComponent11)
+				targetComponent.SubComponent11Data.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+				return nil
+			},
+		},
 		{
 			name:            "read-only always succeeds",
 			intent:          OperationIntentObserve,
@@ -1503,6 +1518,20 @@ func (s *nodeSuite) TestValidateAccess() {
 			terminated:      true,
 			valid:           false,
 		},
+		{
+			name:            "detached node skips parent validation",
+			valid:           true,
+			intent:          OperationIntentProgress,
+			lifecycleStatus: enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED, // root is closed
+			terminated:      false,
+			setup: func(target *Node, _ Context) error {
+				// Set the parent node (SubComponent1) as detached.
+				// When validateParentAccess is called on a detached node, it skips
+				// ancestor validation entirely.
+				target.parent.serializedNode.GetMetadata().GetComponentAttributes().Detached = true
+				return nil
+			},
+		},
 	}
 
 	for _, tc := range testCases {
@@ -1534,10 +1563,8 @@ func (s *nodeSuite) TestValidateAccess() {
 				s.NoError(tc.setup(node, ctx))
 			}
 
-			// Validation always begins on the target node's parent.
-			parent := node.parent
-			s.NotNil(parent)
-			err = parent.validateAccess(ctx)
+			// Validation begins on the target node, checking ancestors only.
+			err = node.validateAccess(ctx)
 			if tc.valid {
 				s.NoError(err)
 			} else {
@@ -1549,6 +1576,70 @@ func (s *nodeSuite) TestValidateAccess() {
 
 }
 
+func (s *nodeSuite) TestGetComponent_DetachedNodeBypassesParentValidation() {
+	// Test that a detached node can be accessed even when its parent is closed.
+	root, err := s.newTestTree(testComponentSerializedNodes())
+	s.NoError(err)
+
+	targetPath := []string{"SubComponent1", "SubComponent11"}
+	targetNode, ok := root.findNode(targetPath)
+	s.True(ok)
+
+	// Mark the target node as detached.
+	targetNode.serializedNode.GetMetadata().GetComponentAttributes().Detached = true
+
+	// Close the root node (set lifecycle to COMPLETED).
+	ctx := NewMutableContext(
+		newContextWithOperationIntent(context.Background(), OperationIntentProgress),
+		root,
+	)
+	err = root.prepareComponentValue(ctx)
+	s.NoError(err)
+	rootComponent, ok := root.value.(*TestComponent)
+	s.True(ok)
+	rootComponent.ComponentData.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+
+	// GetComponent on the detached node should succeed despite root being closed.
+	ref := ComponentRef{
+		componentPath: targetPath,
+	}
+	component, err := root.Component(ctx, ref)
+	s.NoError(err)
+	s.NotNil(component)
+}
+
+func (s *nodeSuite) TestGetComponent_ClosedTargetSucceeds() {
+	// Test that a closed target component can still be accessed via Component()
+	// because we only check ancestor lifecycle, not the target's lifecycle.
+	root, err := s.newTestTree(testComponentSerializedNodes())
+	s.NoError(err)
+
+	targetPath := []string{"SubComponent1", "SubComponent11"}
+	targetNode, ok := root.findNode(targetPath)
+	s.True(ok)
+
+	ctx := NewMutableContext(
+		newContextWithOperationIntent(context.Background(), OperationIntentProgress),
+		root,
+	)
+
+	// Close the target node's lifecycle (set to COMPLETED).
+	err = targetNode.prepareComponentValue(ctx)
+	s.NoError(err)
+	targetComponent, ok := targetNode.value.(*TestSubComponent11)
+	s.True(ok)
+	targetComponent.SubComponent11Data.Status = enumspb.WORKFLOW_EXECUTION_STATUS_COMPLETED
+	s.True(targetComponent.LifecycleState(ctx).IsClosed())
+
+	// GetComponent on the closed target should succeed because we only check ancestors.
+	ref := ComponentRef{
+		componentPath: targetPath,
+	}
+	component, err := root.Component(ctx, ref)
+	s.NoError(err)
+	s.NotNil(component)
+}
+
 func (s *nodeSuite) TestGetComponent() {
 	errValidation := errors.New("some random validation error")
 

proto/internal/temporal/server/api/persistence/v1/chasm.proto

@@ -59,6 +59,10 @@ message ChasmComponentAttributes {
     // Tasks are ordered by their scheduled time, breaking ties by
     // versioned transition and versioned_transition_offset. 
     repeated Task pure_tasks = 3;
+    // When true, this component ignores parent lifecycle validation.
+    // Detached components can continue operating, accepting writes and executing
+    // tasks, even when their parent is closed/terminated.
+    bool detached = 4;
 }
 
 message ChasmDataAttributes {}