fix(core): keep TypeDef root kind consistent for factory custom serializers

java/fory-core/src/main/java/org/apache/fory/meta/NativeTypeDefDecoder.java

@@ -230,6 +230,12 @@ static TypeDef decodeTypeDef(
     // Native TypeDef can carry class-layer fields even when the root wire type is an enum,
     // map, or other non-struct wrapper. Validate the resolved root class kind instead.
     if (rootClass != null) {
+      if (!hasFieldMetadata && Types.isExtType(rootTypeId)) {
+        // Extension roots need their actual serializer: a metadata-only TypeInfo may still have
+        // a provisional struct kind. Resolve and retain it through the normal owner here, not
+        // while building field metadata, which can recurse into serializer construction.
+        resolver.getTypeInfo(rootClass);
+      }
       int expectedRootTypeId = resolver.getTypeDefRootTypeId(rootClass, hasFieldMetadata);
       if (!isCompatibleRootKind(expectedRootTypeId, rootTypeId, !rootClassLayerRegistered)) {
         throw new DeserializationException(

java/fory-core/src/main/java/org/apache/fory/resolver/ClassResolver.java

@@ -1981,18 +1981,18 @@ private TypeDef buildTypeDef(TypeInfo typeInfo, Class<? extends Serializer> seri
     TypeDef typeDef;
     Preconditions.checkArgument(
         serializerClass != UnknownClassSerializers.UnknownStructSerializer.class);
-    if (needToWriteTypeDef(serializerClass) || needsCollectionFieldTypeDef(serializerClass)) {
+    // Meta sharing needs the struct field schema even when compatible mode is disabled.
+    if (isStructSerializerClass(serializerClass) || needsCollectionFieldTypeDef(serializerClass)) {
       // Default collection/map serializers remain non-struct roots, but their wrapper fields still
       // need TypeDef metadata so remote compatible readers can evolve those fields.
       typeDef = typeDefMap.computeIfAbsent(typeInfo.type, cls -> TypeDef.buildTypeDef(this, cls));
     } else {
-      // Some type will use other serializers such MapSerializer and so on.
+      // Field schemas may already exist for class layers or metadata probes. An empty serializer
+      // root has a different shape and belongs to TypeInfo, not the class's field-schema map.
       typeDef =
-          typeDefMap.computeIfAbsent(
-              typeInfo.type,
-              cls ->
-                  NativeTypeDefEncoder.buildTypeDefWithFieldInfos(
-                      this, cls, Collections.emptyList()));
+          cacheTypeDef(
+              NativeTypeDefEncoder.buildTypeDefWithFieldInfos(
+                  this, typeInfo.type, Collections.emptyList()));
     }
     typeInfo.typeDef = typeDef;
     return typeDef;

java/fory-core/src/main/java/org/apache/fory/resolver/TypeResolver.java

@@ -1231,10 +1231,20 @@ private TypeDef matchingLocalTypeDef(long headerHash, Class<?> cls) {
     }
     // A declared polymorphic target can be an unregistered interface or abstract class. It has no
     // concrete local metadata owner, so do not materialize a TypeDef merely to probe for a hit.
-    if (getTypeInfo(cls, false) == null) {
+    TypeInfo typeInfo = getTypeInfo(cls, false);
+    if (typeInfo == null) {
       return null;
     }
-    TypeDef localTypeDef = getTypeDef(cls, true);
+    TypeDef localTypeDef = typeInfo.typeDef;
+    if (localTypeDef == null && typeInfo.serializer != null) {
+      localTypeDef = buildTypeDef(typeInfo);
+    }
+    if (localTypeDef != null && TypeDef.headerHash(localTypeDef.getId()) == headerHash) {
+      return localTypeDef;
+    }
+    // Class layers can use a field schema independently of the serializer's root definition.
+    // Keep this probe metadata-only: creating serializers here changes registration/codegen order.
+    localTypeDef = getTypeDef(cls, true);
     return TypeDef.headerHash(localTypeDef.getId()) == headerHash ? localTypeDef : null;
   }
 

java/fory-core/src/test/java/org/apache/fory/resolver/ClassResolverTest.java

@@ -719,6 +719,40 @@ public void testIdEnumDoesNotUseTypeDefMetaLimits() {
     serDeCheck(fory, TestNeedToWriteReferenceClass.A);
   }
 
+  @Test(dataProvider = "enableCodegen")
+  public void testColdLocalTypeDef(boolean codegen) {
+    for (boolean named : new boolean[] {false, true}) {
+      for (boolean typed : new boolean[] {false, true}) {
+        ForyBuilder builder =
+            Fory.builder()
+                .withXlang(false)
+                .withCompatible(true)
+                .withScopedMetaShare(true)
+                .withMaxSchemaVersionsPerType(1)
+                .withCodegen(codegen)
+                .withAsyncCompilation(false);
+        Fory writer = builder.build();
+        Fory reader = builder.build();
+        if (named) {
+          writer.register(Foo.class, "test.Foo");
+          reader.register(Foo.class, "test.Foo");
+        } else {
+          writer.register(Foo.class, 101);
+          reader.register(Foo.class, 101);
+        }
+        TypeResolver resolver = reader.getTypeResolver();
+        assertNull(resolver.getTypeInfo(Foo.class, false).getSerializer());
+        Foo value = new Foo();
+        byte[] bytes = writer.serialize(value);
+        Foo copy = typed ? reader.deserialize(bytes, Foo.class) : (Foo) reader.deserialize(bytes);
+        assertEquals(copy, value);
+        TypeDef localTypeDef = writer.getTypeResolver().getTypeInfo(Foo.class).getTypeDef();
+        // An exact local schema must not consume the allowance for remote schema versions.
+        assertNull(resolver.getCheckedRemoteTypeDef(TypeDef.headerHash(localTypeDef.getId())));
+      }
+    }
+  }
+
   @Test
   public void testIdExtDoesNotUseTypeDefMetaLimits() {
     Fory fory =

java/fory-core/src/test/java/org/apache/fory/serializer/SerializerFactoryTest.java

@@ -34,8 +34,10 @@
 import org.apache.fory.context.ReadContext;
 import org.apache.fory.context.WriteContext;
 import org.apache.fory.memory.MemoryBuffer;
+import org.apache.fory.resolver.ClassResolver;
 import org.apache.fory.resolver.TypeResolver;
 import org.testng.Assert;
+import org.testng.annotations.DataProvider;
 import org.testng.annotations.Test;
 
 public class SerializerFactoryTest {
@@ -181,4 +183,172 @@ private static void assertKryoSerializer(Fory fory) {
     Object a2 = fory.deserialize(fory.serialize(a));
     Assert.assertEquals(a, a2);
   }
+
+  public interface Restriction {}
+
+  public static class RestrictionValue implements Restriction {
+    public String code;
+  }
+
+  public static class RestrictionHolder {
+    public Restriction restriction;
+  }
+
+  public static class FactoryBean {
+    public int value;
+    public FactoryBean next;
+  }
+
+  public enum FactoryEnum {
+    VALUE
+  }
+
+  private static class RestrictionSerializer extends Serializer<RestrictionValue> {
+    RestrictionSerializer(TypeResolver resolver) {
+      super(resolver.getConfig(), RestrictionValue.class);
+    }
+
+    @Override
+    public void write(WriteContext ctx, RestrictionValue value) {
+      ctx.writeRef(value.code);
+    }
+
+    @Override
+    public RestrictionValue read(ReadContext ctx) {
+      RestrictionValue value = new RestrictionValue();
+      ctx.reference(value);
+      value.code = (String) ctx.readRef();
+      return value;
+    }
+  }
+
+  @DataProvider
+  public Object[][] factoryModes() {
+    return new Object[][] {{false}, {true}};
+  }
+
+  @DataProvider
+  public Object[][] metaFactoryModes() {
+    List<Object[]> modes = new ArrayList<>();
+    for (boolean codegen : new boolean[] {false, true}) {
+      for (boolean registered : new boolean[] {false, true}) {
+        for (boolean metadataFirst : new boolean[] {false, true}) {
+          modes.add(new Object[] {codegen, registered, metadataFirst});
+        }
+      }
+    }
+    return modes.toArray(new Object[0][]);
+  }
+
+  private static Fory metaShareFory(boolean codegen, SerializerFactory factory) {
+    return Fory.builder()
+        .withXlang(false)
+        .withCompatible(true)
+        .withScopedMetaShare(true)
+        .withRefTracking(true)
+        .requireClassRegistration(false)
+        .withCodegen(codegen)
+        .withAsyncCompilation(false)
+        .withSerializerFactory(factory)
+        .build();
+  }
+
+  private static SerializerFactory restrictionFactory(AtomicInteger creations) {
+    return (resolver, cls) -> {
+      if (cls == RestrictionValue.class) {
+        creations.incrementAndGet();
+        return new RestrictionSerializer(resolver);
+      }
+      return null;
+    };
+  }
+
+  @Test(dataProvider = "metaFactoryModes")
+  public void testMetaShareFactory(boolean codegen, boolean registered, boolean metadataFirst) {
+    AtomicInteger writerCreations = new AtomicInteger();
+    AtomicInteger readerCreations = new AtomicInteger();
+    Fory writer = metaShareFory(codegen, restrictionFactory(writerCreations));
+    Fory reader = metaShareFory(codegen, restrictionFactory(readerCreations));
+    if (registered) {
+      writer.register(RestrictionValue.class);
+      reader.register(RestrictionValue.class);
+    }
+    if (metadataFirst) {
+      ClassResolver resolver = (ClassResolver) reader.getTypeResolver();
+      resolver.getTypeIdForTypeDef(RestrictionValue.class);
+      Assert.assertNull(resolver.getSerializer(RestrictionValue.class, false));
+    }
+    RestrictionValue value = new RestrictionValue();
+    value.code = "A";
+    if (metadataFirst) {
+      RestrictionValue copy = reader.deserialize(writer.serialize(value), RestrictionValue.class);
+      Assert.assertEquals(copy.code, value.code);
+    }
+    RestrictionHolder holder = new RestrictionHolder();
+    holder.restriction = value;
+    for (int i = 0; i < 2; i++) {
+      RestrictionHolder copy =
+          reader.deserialize(writer.serialize(holder), RestrictionHolder.class);
+      Assert.assertEquals(((RestrictionValue) copy.restriction).code, "A");
+    }
+    RestrictionHolder restored =
+        writer.deserialize(reader.serialize(holder), RestrictionHolder.class);
+    Assert.assertEquals(((RestrictionValue) restored.restriction).code, "A");
+    Assert.assertEquals(writerCreations.get(), 1);
+    Assert.assertEquals(readerCreations.get(), 1);
+  }
+
+  @Test(dataProvider = "factoryModes")
+  @SuppressWarnings({"rawtypes", "unchecked"})
+  public void testMetaShareEnumFactory(boolean codegen) {
+    SerializerFactory factory =
+        (resolver, cls) ->
+            cls == FactoryEnum.class ? new EnumSerializer(resolver.getConfig(), (Class) cls) : null;
+    Fory writer = metaShareFory(codegen, factory);
+    Fory reader = metaShareFory(codegen, factory);
+    Assert.assertSame(reader.deserialize(writer.serialize(FactoryEnum.VALUE)), FactoryEnum.VALUE);
+  }
+
+  @Test(dataProvider = "factoryModes")
+  public void testMetaShareStructFactory(boolean codegen) {
+    SerializerFactory factory =
+        (resolver, cls) -> cls == FactoryBean.class ? new ObjectSerializer<>(resolver, cls) : null;
+    Fory writer = metaShareFory(codegen, factory);
+    Fory reader = metaShareFory(codegen, factory);
+    FactoryBean value = new FactoryBean();
+    value.value = 42;
+    value.next = value;
+    FactoryBean copy = (FactoryBean) reader.deserialize(writer.serialize(value));
+    Assert.assertEquals(copy.value, value.value);
+    Assert.assertSame(copy.next, copy);
+    // Reading metadata must not leave a partially constructed serializer for a later write.
+    FactoryBean restored = (FactoryBean) writer.deserialize(reader.serialize(copy));
+    Assert.assertEquals(restored.value, value.value);
+    Assert.assertSame(restored.next, restored);
+  }
+
+  @Test(dataProvider = "factoryModes")
+  public void testMetaShareFactoryFailure(boolean codegen) {
+    Fory writer = metaShareFory(codegen, restrictionFactory(new AtomicInteger()));
+    AtomicInteger attempts = new AtomicInteger();
+    Fory reader =
+        metaShareFory(
+            codegen,
+            (resolver, cls) -> {
+              if (cls != RestrictionValue.class) {
+                return null;
+              }
+              if (attempts.incrementAndGet() == 1) {
+                throw new IllegalStateException("Serializer construction failed");
+              }
+              return new RestrictionSerializer(resolver);
+            });
+    RestrictionValue value = new RestrictionValue();
+    value.code = "A";
+    byte[] bytes = writer.serialize(value);
+    Assert.expectThrows(RuntimeException.class, () -> reader.deserialize(bytes));
+    RestrictionValue copy = (RestrictionValue) reader.deserialize(bytes);
+    Assert.assertEquals(copy.code, value.code);
+    Assert.assertEquals(attempts.get(), 2);
+  }
 }